diff --git a/receive_flow.go b/receive_flow.go index d60a1e2..0808542 100644 --- a/receive_flow.go +++ b/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 diff --git a/receive_flow_test.go b/receive_flow_test.go index 4d1813a..4869915 100644 --- a/receive_flow_test.go +++ b/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) + } + }) + } +} diff --git a/session_test.go b/session_test.go index 4d03035..d364969 100644 --- a/session_test.go +++ b/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 }