Stop treating RST_STREAM as EOF

In hq, it's important to know when the stream is reset as opposed to ending naturally.  You need to know this so that you can acknowledge any potentially unread header blocks.

This changes the code so that resets cause subsequent reads to immediately fail with a recognizable error code.  Receiving RST_STREAM stops further reading.  It already caused existing data to be discarded.  Finally, when a read fails on a reset stream, the state moves to the previously unused ResetRead state.

Author: martinthomson

connection_test.go

@@ -405,7 +405,7 @@ func (h *testReceiveHandler) StreamReadable(s RecvStream) {
 			break
 		case ErrorWouldBlock:
 			return
-		case ErrorStreamIsClosed, ErrorConnIsClosed, io.EOF:
+		case ErrorStreamReset, ErrorConnIsClosed, io.EOF:
 			h.done = true
 			return
 		default:
@@ -565,7 +565,7 @@ func TestSendReceiveStreamRst(t *testing.T) {
 	ss := pair.server.GetStream(4)
 	b := make([]byte, 1024)
 	n, err = ss.Read(b)
-	assertEquals(t, err, io.EOF)
+	assertEquals(t, err, ErrorStreamReset)
 	assertEquals(t, 0, n)
 }
 
@@ -783,10 +783,11 @@ func TestUnidirectionalStreamRst(t *testing.T) {
 	err = inputAll(client)
 	assertNotError(t, err, "packets should be OK")
 
+	assertEquals(t, cstream.RecvState(), RecvStreamStateResetRecvd)
 	n, err = cstream.Read(d)
-	assertEquals(t, err, io.EOF)
+	assertEquals(t, err, ErrorStreamReset)
 	assertEquals(t, n, 0)
-	assertEquals(t, cstream.RecvState(), RecvStreamStateResetRecvd)
+	assertEquals(t, cstream.RecvState(), RecvStreamStateResetRead)
 }
 
 func TestUnidirectionalStreamRstImmediate(t *testing.T) {
@@ -801,11 +802,12 @@ func TestUnidirectionalStreamRstImmediate(t *testing.T) {
 	assertNotError(t, err, "packets should be OK")
 
 	cstream := pair.client.GetRecvStream(sstream.Id())
+	assertEquals(t, cstream.RecvState(), RecvStreamStateResetRecvd)
 	var d [3]byte
 	n, err := cstream.Read(d[:])
-	assertEquals(t, err, io.EOF)
+	assertEquals(t, err, ErrorStreamReset)
 	assertEquals(t, n, 0)
-	assertEquals(t, cstream.RecvState(), RecvStreamStateResetRecvd)
+	assertEquals(t, cstream.RecvState(), RecvStreamStateResetRead)
 }
 
 func TestUnidirectionalStopSending(t *testing.T) {
@@ -839,10 +841,11 @@ func TestUnidirectionalStopSending(t *testing.T) {
 	err = inputAll(pair.server)
 	assertNotError(t, err, "packets should be OK")
 
+	assertEquals(t, sstream.RecvState(), RecvStreamStateResetRecvd)
 	n, err = sstream.Read(d)
-	assertEquals(t, err, io.EOF)
+	assertEquals(t, err, ErrorStreamReset)
 	assertEquals(t, n, 0)
-	assertEquals(t, sstream.RecvState(), RecvStreamStateResetRecvd)
+	assertEquals(t, sstream.RecvState(), RecvStreamStateResetRead)
 }
 
 func TestBidirectionalStopSending(t *testing.T) {

errors.go

@@ -86,6 +86,7 @@ var ErrorDestroyConnection = fatalError("Terminate connection")
 var ErrorReceivedVersionNegotiation = fatalError("Received a version negotiation packet advertising a different version than ours")
 var ErrorConnIsClosed = fatalError("Connection is closed")
 var ErrorConnIsClosing = nonFatalError("Connection is closing")
+var ErrorStreamReset = fatalError("Stream was reset")
 var ErrorStreamIsClosed = fatalError("Stream is closed")
 var ErrorInvalidPacket = nonFatalError("Invalid packet")
 var ErrorConnectionTimedOut = fatalError("Connection timed out")

stream.go

@@ -52,7 +52,7 @@ const (
 	RecvStreamStateDataRecvd  = RecvStreamState(2) // Not tracked
 	RecvStreamStateResetRecvd = RecvStreamState(3)
 	RecvStreamStateDataRead   = RecvStreamState(4)
-	RecvStreamStateResetRead  = RecvStreamState(5) // Not tracked
+	RecvStreamStateResetRead  = RecvStreamState(5)
 )
 
 // String produces a nice string from a RecvStreamState.
@@ -392,6 +392,12 @@ func (s *recvStreamBase) read(b []byte) (int, error) {
 	s.log(logTypeStream, "Reading len=%v read offset=%v available chunks=%v",
 		len(b), s.readOffset, len(s.chunks))
 
+	if s.state == RecvStreamStateResetRecvd {
+		s.log(logTypeStream, "Reading stopped for RST_STREAM")
+		s.setRecvState(RecvStreamStateResetRead)
+		return 0, ErrorStreamReset
+	}
+
 	read := 0
 
 	for len(b) > 0 {