Fix receive stream error handling

receive_flow.go

@@ -3,6 +3,7 @@ package roq
 import (
 	"bytes"
 	"context"
+	"errors"
 	"io"
 	"log"
 	"sync"
@@ -95,12 +96,13 @@ func (f *ReceiveFlow) readStream(rs ReceiveStream) {
 	for {
 		length, err := quicvarint.Read(reader)
 		if err != nil {
-			if err == io.EOF {
+			if errors.Is(err, io.EOF) {
 				return
 			}
-			streamErr, ok := err.(*quic.StreamError)
-			if ok {
-				log.Printf("got stream error: %v", streamErr)
+			var streamErr *quic.StreamError
+			if errors.As(err, &streamErr) {
+				log.Printf("got stream error while reading length: %v", streamErr)
+				return
 			}
 			log.Printf("got unexpected error while reading length: %v", err)
 			return
@@ -110,9 +112,10 @@ func (f *ReceiveFlow) readStream(rs ReceiveStream) {
 		b.Reset()
 		n, err := b.ReadFrom(r)
 		if err != nil {
-			streamErr, ok := err.(*quic.StreamError)
-			if ok {
-				log.Printf("got stream error: %v", streamErr)
+			var streamErr *quic.StreamError
+			if errors.As(err, &streamErr) {
+				log.Printf("got stream error after reading %v bytes of payload: %v", n, streamErr)
+				return
 			}
 			log.Printf("got unexpected error after reading %v bytes of payload: %v", n, err)
 			return

receive_flow_test.go

@@ -4,11 +4,16 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"io"
+	"log"
+	"os"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/quic-go/quic-go"
 	"github.com/quic-go/quic-go/quicvarint"
 )
 
@@ -309,3 +314,64 @@ func TestConcurrentPushAndClose(t *testing.T) {
 		}
 	}
 }
+
+// readStream must recognise wrapped errors: a wrapped io.EOF ends the stream
+// quietly, and a wrapped stream error is reported once, not twice.
+func TestReadStreamWrappedErrors(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		err  error
+		want string
+	}{{
+		name: "wrapped EOF",
+		err:  fmt.Errorf("read failed: %w", io.EOF),
+		want: "",
+	}, {
+		name: "wrapped stream error",
+		err:  fmt.Errorf("read failed: %w", &quic.StreamError{StreamID: 1, ErrorCode: 42}),
+		want: "got stream error while reading length",
+	}, {
+		name: "other error",
+		err:  errors.New("boom"),
+		want: "got unexpected error while reading length",
+	}} {
+		t.Run(tc.name, func(t *testing.T) {
+			var logs bytes.Buffer
+			log.SetOutput(&logs)
+			defer log.SetOutput(os.Stderr)
+
+			f := newReceiveFlow(1, 10, nil)
+			payload := []byte{0x42}
+			data := quicvarint.Append(nil, uint64(len(payload)))
+			data = append(data, payload...)
+			f.readStream(&stubReceiveStream{
+				id:        1,
+				data:      data,
+				readErr:   tc.err,
+				cancelled: make(chan struct{}),
+			})
+
+			if len(f.buffer) != 1 {
+				t.Errorf("flow buffered %d packets, want 1", len(f.buffer))
+			}
+			lines := 0
+			for _, line := range strings.Split(strings.TrimSpace(logs.String()), "\n") {
+				if line != "" {
+					lines++
+				}
+			}
+			if tc.want == "" {
+				if lines != 0 {
+					t.Errorf("logged %q, want nothing", logs.String())
+				}
+				return
+			}
+			if lines != 1 {
+				t.Errorf("logged %d lines, want 1:\n%s", lines, logs.String())
+			}
+			if !strings.Contains(logs.String(), tc.want) {
+				t.Errorf("logged %q, want it to contain %q", logs.String(), tc.want)
+			}
+		})
+	}
+}

session_test.go

@@ -68,9 +68,12 @@ func (s *stubSendStream) isClosed() bool {
 // stubReceiveStream feeds a fixed payload. Once it is exhausted the stream
 // either ends (eof) or stays open until CancelRead.
 type stubReceiveStream struct {
-	id        int64
-	data      []byte
-	eof       bool
+	id   int64
+	data []byte
+	eof  bool
+	// readErr, if set, is returned once the payload is exhausted, in
+	// preference to eof and parking.
+	readErr   error
 	cancelled chan struct{}
 	once      sync.Once
 
@@ -92,6 +95,9 @@ func (s *stubReceiveStream) Read(p []byte) (int, error) {
 		s.data = s.data[n:]
 		return n, nil
 	}
+	if s.readErr != nil {
+		return 0, s.readErr
+	}
 	if s.eof {
 		return 0, io.EOF
 	}