From a503146b0468437e0d27d3c258fecee34928d6a7 Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Fri, 28 Aug 2026 16:27:52 +0200 Subject: [PATCH] Tidy logger usage and configuration --- options.go | 23 ++++++++++++ options_test.go | 70 +++++++++++++++++++++++++++++++++++++ receive_flow.go | 23 ++++++++---- receive_flow_buffer.go | 5 +-- receive_flow_buffer_test.go | 10 +++--- receive_flow_test.go | 66 +++++++++++++++++++++++----------- send_flow.go | 10 ++++-- session.go | 9 +++-- 8 files changed, 176 insertions(+), 40 deletions(-) diff --git a/options.go b/options.go index d6c3015..d9cd798 100644 --- a/options.go +++ b/options.go @@ -2,6 +2,7 @@ package roq import ( "fmt" + "log/slog" "github.com/quic-go/quic-go/qlogwriter" ) @@ -21,6 +22,7 @@ type Option func(*sessionConfig) error type sessionConfig struct { receiveBufferSize int unknownFlowBufferSize int + logger *slog.Logger qlogTrace qlogwriter.Trace qlogPacketData bool qlogPacketDataLimit int @@ -31,6 +33,7 @@ func newSessionConfig(opts []Option) (*sessionConfig, error) { c := &sessionConfig{ receiveBufferSize: defaultReceiveBufferSize, unknownFlowBufferSize: defaultUnknownFlowBufferSize, + logger: discardLogger(), } for _, opt := range opts { if opt == nil { @@ -104,3 +107,23 @@ func WithQlogPacketData(maxBytes int) Option { return nil } } + +// WithLogger sets the logger the session and its flows write to.The session +// discards all log records unless the application passes one in. Pass +// slog.Default() to send them to the standard logger. The logger must not be +// nil. +func WithLogger(logger *slog.Logger) Option { + return func(c *sessionConfig) error { + if logger == nil { + return fmt.Errorf("%w: logger must not be nil", errInvalidOption) + } + c.logger = logger + return nil + } +} + +// discardLogger returns a logger that drops every record. It is what a session +// uses when the application did not pass WithLogger. +func discardLogger() *slog.Logger { + return slog.New(slog.DiscardHandler) +} diff --git a/options_test.go b/options_test.go index 4078677..217089b 100644 --- a/options_test.go +++ b/options_test.go @@ -1,7 +1,11 @@ package roq import ( + "bytes" "errors" + "log" + "log/slog" + "os" "testing" "time" @@ -94,6 +98,7 @@ func TestInvalidOptions(t *testing.T) { {"zero unknown flow buffer", WithUnknownFlowBufferSize(0)}, {"negative unknown flow buffer", WithUnknownFlowBufferSize(-1)}, {"negative qlog packet data limit", WithQlogPacketData(-1)}, + {"nil logger", WithLogger(nil)}, {"nil option", nil}, } { t.Run(tc.name, func(t *testing.T) { @@ -107,6 +112,71 @@ func TestInvalidOptions(t *testing.T) { } } +func TestWithLogger(t *testing.T) { + var logs bytes.Buffer + logger := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{ + Level: slog.LevelDebug, + })) + + c := newStubConn() + s, err := NewSession(c, true, WithLogger(logger)) + if err != nil { + t.Fatal(err) + } + defer closeWithin(t, s, 5*time.Second) + + rf, err := s.NewReceiveFlow(1) + if err != nil { + t.Fatal(err) + } + if rf.logger != logger { + t.Error("receive flow does not use the logger passed with WithLogger") + } + sf, err := s.NewSendFlow(1) + if err != nil { + t.Fatal(err) + } + if sf.logger != logger { + t.Error("send flow does not use the logger passed with WithLogger") + } + + c.datagrams <- quicvarint.Append(nil, 2) + if bf := awaitBufferedFlow(t, s, 2); bf.logger != logger { + t.Error("buffered flow does not use the logger passed with WithLogger") + } +} + +func TestSessionDoesNotUseGlobalLoggerByDefault(t *testing.T) { + var logs bytes.Buffer + log.SetOutput(&logs) + defer log.SetOutput(os.Stderr) + defaultLogger := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{ + Level: slog.LevelDebug, + }))) + defer slog.SetDefault(defaultLogger) + + s, err := NewSession(newStubConn(), true) + if err != nil { + t.Fatal(err) + } + defer closeWithin(t, s, 5*time.Second) + + f, err := s.NewReceiveFlow(1) + if err != nil { + t.Fatal(err) + } + f.readStream(&stubReceiveStream{ + id: 1, + readErr: errors.New("boom"), + cancelled: make(chan struct{}), + }) + + if logs.Len() != 0 { + t.Errorf("session wrote %q to the global logger, want nothing", logs.String()) + } +} + // awaitBufferedFlow waits for the session to buffer a flow for an unclaimed // flow ID and returns it. func awaitBufferedFlow(t *testing.T, s *Session, id uint64) *ReceiveFlow { diff --git a/receive_flow.go b/receive_flow.go index cb230ee..7aa1afa 100644 --- a/receive_flow.go +++ b/receive_flow.go @@ -5,7 +5,7 @@ import ( "context" "errors" "io" - "log" + "log/slog" "os" "sync" "time" @@ -38,10 +38,14 @@ type ReceiveFlow struct { deadlineExpired bool deadlineTimer *time.Timer - qlog *qlogger + qlog *qlogger + logger *slog.Logger } -func newReceiveFlow(id uint64, receiveBufferSize int, qlog *qlogger) *ReceiveFlow { +func newReceiveFlow(id uint64, receiveBufferSize int, qlog *qlogger, logger *slog.Logger) *ReceiveFlow { + if logger == nil { + logger = discardLogger() + } ctx, cancel := context.WithCancel(context.Background()) return &ReceiveFlow{ id: id, @@ -58,6 +62,7 @@ func newReceiveFlow(id uint64, receiveBufferSize int, qlog *qlogger) *ReceiveFlo closed: false, deadlineCh: make(chan struct{}), qlog: qlog, + logger: logger, } } @@ -110,10 +115,12 @@ func (f *ReceiveFlow) readStream(rs ReceiveStream) { } var streamErr *quic.StreamError if errors.As(err, &streamErr) { - log.Printf("got stream error while reading length: %v", streamErr) + f.logger.Debug("got stream error while reading length", + "flowID", f.id, "streamID", rs.ID(), "error", streamErr) return } - log.Printf("got unexpected error while reading length: %v", err) + f.logger.Warn("got unexpected error while reading length", + "flowID", f.id, "streamID", rs.ID(), "error", err) return } r := io.LimitReader(reader, int64(length)) @@ -126,10 +133,12 @@ func (f *ReceiveFlow) readStream(rs ReceiveStream) { if err != nil { var streamErr *quic.StreamError if errors.As(err, &streamErr) { - log.Printf("got stream error after reading %v bytes of payload: %v", n, streamErr) + f.logger.Debug("got stream error while reading payload", + "flowID", f.id, "streamID", rs.ID(), "read", n, "error", streamErr) return } - log.Printf("got unexpected error after reading %v bytes of payload: %v", n, err) + f.logger.Warn("got unexpected error while reading payload", + "flowID", f.id, "streamID", rs.ID(), "read", n, "error", err) return } if f.qlog != nil { diff --git a/receive_flow_buffer.go b/receive_flow_buffer.go index d363a81..2643144 100644 --- a/receive_flow_buffer.go +++ b/receive_flow_buffer.go @@ -1,6 +1,7 @@ package roq import ( + "log/slog" "slices" "sync" ) @@ -26,7 +27,7 @@ func newReceiveFlowBuffer(maxLen int) *receiveFlowBuffer { // flow if the buffer holds none. Lookup and creation happen under a single // lock hold, so concurrent callers for the same unknown flow ID all get the // same flow. -func (b *receiveFlowBuffer) getOrCreate(id uint64, receiveBufferSize int, qlog *qlogger) *ReceiveFlow { +func (b *receiveFlowBuffer) getOrCreate(id uint64, receiveBufferSize int, qlog *qlogger, logger *slog.Logger) *ReceiveFlow { b.mutex.Lock() defer b.mutex.Unlock() if f, ok := b.buffer[id]; ok { @@ -41,7 +42,7 @@ func (b *receiveFlowBuffer) getOrCreate(id uint64, receiveBufferSize int, qlog * delete(b.buffer, front) b.queue = b.queue[1:] } - f := newReceiveFlow(id, receiveBufferSize, qlog) + f := newReceiveFlow(id, receiveBufferSize, qlog, logger) b.queue = append(b.queue, id) b.buffer[id] = f return f diff --git a/receive_flow_buffer_test.go b/receive_flow_buffer_test.go index f2d4b21..ff55810 100644 --- a/receive_flow_buffer_test.go +++ b/receive_flow_buffer_test.go @@ -7,7 +7,7 @@ import ( func TestReceiveFlowBufferPopRemovesFlow(t *testing.T) { b := newReceiveFlowBuffer(16) - f := b.getOrCreate(1, 1, nil) + f := b.getOrCreate(1, 1, nil, nil) if got := b.pop(1); got != f { t.Fatalf("pop returned %v, want %v", got, f) @@ -21,7 +21,7 @@ func TestReceiveFlowBufferPopRemovesFlow(t *testing.T) { if got := b.pop(1); got != nil { t.Errorf("second pop returned %v, want nil", got) } - if got := b.getOrCreate(1, 1, nil); got == f { + if got := b.getOrCreate(1, 1, nil, nil); got == f { t.Error("getOrCreate returned the popped flow, want a new one") } } @@ -31,7 +31,7 @@ func TestReceiveFlowBufferPopRemovesFlow(t *testing.T) { func TestReceiveFlowBufferPoppedFlowIsNotEvicted(t *testing.T) { const maxLen = 16 b := newReceiveFlowBuffer(maxLen) - f := b.getOrCreate(1, 1, nil) + f := b.getOrCreate(1, 1, nil, nil) rs := &stubReceiveStream{id: 1, cancelled: make(chan struct{})} f.streams[rs.ID()] = rs @@ -40,7 +40,7 @@ func TestReceiveFlowBufferPoppedFlowIsNotEvicted(t *testing.T) { t.Fatalf("pop returned %v, want %v", got, f) } for i := range uint64(maxLen + 1) { - b.getOrCreate(100+i, 1, nil) + b.getOrCreate(100+i, 1, nil, nil) } select { @@ -71,7 +71,7 @@ func TestReceiveFlowBufferGetOrCreateIsAtomic(t *testing.T) { go func() { defer done.Done() start.Wait() - flows[i] = b.getOrCreate(1, 1, nil) + flows[i] = b.getOrCreate(1, 1, nil, nil) }() } start.Done() diff --git a/receive_flow_test.go b/receive_flow_test.go index 6e57851..8336212 100644 --- a/receive_flow_test.go +++ b/receive_flow_test.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "log" + "log/slog" "os" "strings" "sync" @@ -20,7 +21,7 @@ import ( // readStream must unregister the stream when it returns, otherwise flows that // receive one stream per frame accumulate finished streams forever. func TestReadStreamUnregistersOnEOF(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) for i := range int64(4) { payload := []byte{0x42} @@ -45,7 +46,7 @@ func TestReadStreamUnregistersOnEOF(t *testing.T) { } func TestReadStreamUnregistersOnClose(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) rs := &stubReceiveStream{ id: 1, cancelled: make(chan struct{}), @@ -81,7 +82,7 @@ func TestReadStreamUnregistersOnClose(t *testing.T) { // Read must not silently truncate: a caller whose buffer is too small for the // next packet gets io.ErrShortBuffer instead of a mangled RTP packet. func TestReadShortBuffer(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) packet := []byte{0x80, 0x60, 0x00, 0x01, 0xde, 0xad, 0xbe, 0xef} f.push(bytes.NewBuffer(packet)) @@ -98,7 +99,7 @@ func TestReadShortBuffer(t *testing.T) { // A short read must not stall the flow: the truncated packet is dropped and the // next packet is still readable. func TestReadAfterShortBuffer(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) first := []byte{0x80, 0x60, 0x00, 0x01} second := []byte{0x42} f.push(bytes.NewBuffer(first)) @@ -120,7 +121,7 @@ func TestReadAfterShortBuffer(t *testing.T) { // A buffer of exactly the packet size is not short. func TestReadExactBuffer(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) packet := []byte{0x80, 0x60, 0x00, 0x01, 0xde, 0xad, 0xbe, 0xef} f.push(bytes.NewBuffer(packet)) @@ -137,7 +138,7 @@ func TestReadExactBuffer(t *testing.T) { // Pooled buffers must be allocated with capacity, not content: a buffer holding // 65535 bytes only works because every Get is followed by a Reset. func TestBufferPoolAllocatesEmptyBuffers(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) b := f.bufferPool.Get().(*bytes.Buffer) if b.Len() != 0 { @@ -151,7 +152,7 @@ func TestBufferPoolAllocatesEmptyBuffers(t *testing.T) { // A packet dropped because the queue is full must go back to the pool, // otherwise overload defeats the pool entirely. func TestPushReturnsDroppedBufferToPool(t *testing.T) { - f := newReceiveFlow(1, 1, nil) + f := newReceiveFlow(1, 1, nil, nil) allocated := countPoolAllocations(f) f.push(f.bufferPool.Get().(*bytes.Buffer)) @@ -195,7 +196,7 @@ func countPoolAllocations(f *ReceiveFlow) *int { // closed, instead of racing them against the cancelled context. func TestReadDrainsBufferAfterClose(t *testing.T) { const packets = 20 - f := newReceiveFlow(1, packets, nil) + f := newReceiveFlow(1, packets, nil, nil) for i := range packets { f.push(bytes.NewBuffer([]byte{byte(i)})) } @@ -222,7 +223,7 @@ func TestReadDrainsBufferAfterClose(t *testing.T) { // race must be dropped rather than queued behind the close where no Read would // ever return it. func TestPushAfterCloseIsDropped(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) f.push(bytes.NewBuffer([]byte{0x01})) if err := f.Close(); err != nil { t.Fatalf("Close: %v", err) @@ -251,7 +252,7 @@ func TestPushAfterCloseIsDropped(t *testing.T) { // Read keeps reporting the error once a closed flow ran dry, instead of // blocking or panicking on the closed buffer. func TestReadAfterCloseIsRepeatable(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) if err := f.Close(); err != nil { t.Fatalf("Close: %v", err) } @@ -266,7 +267,7 @@ func TestReadAfterCloseIsRepeatable(t *testing.T) { // Sessions close their flows on shutdown, so a flow the application also closes // itself is closed twice. That must not panic on the buffer. func TestCloseFlowTwice(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) if err := f.Close(); err != nil { t.Fatalf("Close: %v", err) } @@ -278,7 +279,7 @@ func TestCloseFlowTwice(t *testing.T) { // Closing a flow while its producers are still pushing must not lose a packet // that was queued, and must not send on the closed buffer. func TestConcurrentPushAndClose(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) var wg sync.WaitGroup wg.Add(2) go func() { @@ -335,10 +336,11 @@ func TestReadStreamWrappedErrors(t *testing.T) { }} { t.Run(tc.name, func(t *testing.T) { var logs bytes.Buffer - log.SetOutput(&logs) - defer log.SetOutput(os.Stderr) + logger := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{ + Level: slog.LevelDebug, + })) - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, logger) payload := []byte{0x42} data := quicvarint.Append(nil, uint64(len(payload))) data = append(data, payload...) @@ -374,9 +376,31 @@ func TestReadStreamWrappedErrors(t *testing.T) { } } +func TestReadStreamDoesNotUseGlobalLogger(t *testing.T) { + var logs bytes.Buffer + log.SetOutput(&logs) + defer log.SetOutput(os.Stderr) + defaultLogger := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{ + Level: slog.LevelDebug, + }))) + defer slog.SetDefault(defaultLogger) + + f := newReceiveFlow(1, 10, nil, nil) + f.readStream(&stubReceiveStream{ + id: 1, + readErr: errors.New("boom"), + cancelled: make(chan struct{}), + }) + + if logs.Len() != 0 { + t.Errorf("flow wrote %q to the global logger, want nothing", logs.String()) + } +} + // A deadline that passes while Read is blocked must unblock it. func TestReadDeadlineExpires(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) if err := f.SetReadDeadline(time.Now().Add(20 * time.Millisecond)); err != nil { t.Fatalf("SetReadDeadline: %v", err) } @@ -401,7 +425,7 @@ func TestReadDeadlineExpires(t *testing.T) { // A deadline in the past times out immediately, even with a packet buffered. func TestReadDeadlineInThePast(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) f.push(bytes.NewBuffer([]byte{0x42})) if err := f.SetReadDeadline(time.Now().Add(-time.Second)); err != nil { t.Fatalf("SetReadDeadline: %v", err) @@ -421,7 +445,7 @@ func TestReadDeadlineInThePast(t *testing.T) { // Setting a deadline must also apply to a Read that is already blocked. func TestSetReadDeadlineWhileReadBlocked(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) done := make(chan error, 1) go func() { _, err := f.Read(make([]byte, 100)) @@ -445,7 +469,7 @@ func TestSetReadDeadlineWhileReadBlocked(t *testing.T) { // Extending the deadline of a blocked reader must keep it blocked, and the // replaced deadline must not fire on the new one. func TestExtendReadDeadlineWhileReadBlocked(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) if err := f.SetReadDeadline(time.Now().Add(20 * time.Millisecond)); err != nil { t.Fatalf("SetReadDeadline: %v", err) } @@ -475,7 +499,7 @@ func TestExtendReadDeadlineWhileReadBlocked(t *testing.T) { // A deadline must not keep Read from reporting the closed flow. func TestReadDeadlineAndClose(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) if err := f.SetReadDeadline(time.Now().Add(time.Hour)); err != nil { t.Fatalf("SetReadDeadline: %v", err) } @@ -500,7 +524,7 @@ func TestReadDeadlineAndClose(t *testing.T) { // Racing SetReadDeadline against readers and expiring timers must not panic on // a double close of the deadline channel. func TestConcurrentSetReadDeadline(t *testing.T) { - f := newReceiveFlow(1, 10, nil) + f := newReceiveFlow(1, 10, nil, nil) var wg sync.WaitGroup for range 8 { wg.Add(1) diff --git a/send_flow.go b/send_flow.go index 6e97c83..2bb979f 100644 --- a/send_flow.go +++ b/send_flow.go @@ -19,9 +19,13 @@ type SendFlow struct { onClose func() closedErr error qlog *qlogger + logger *slog.Logger } -func newFlow(conn Connection, id uint64, onClose func(), qlog *qlogger) *SendFlow { +func newFlow(conn Connection, id uint64, onClose func(), qlog *qlogger, logger *slog.Logger) *SendFlow { + if logger == nil { + logger = discardLogger() + } flowID := make([]byte, 0, quicvarint.Len(id)) flowID = quicvarint.Append(flowID, id) return &SendFlow{ @@ -33,6 +37,7 @@ func newFlow(conn Connection, id uint64, onClose func(), qlog *qlogger) *SendFlo onClose: onClose, closedErr: nil, qlog: qlog, + logger: logger, } } @@ -72,7 +77,8 @@ func (f *SendFlow) NewSendStream(ctx context.Context, priority uint32, incremant } priorityStream, ok := s.(PrioritySendStream) if ok { - slog.Info("roq setting stream priority", "priority", priority, "incremental", incremantal) + f.logger.Debug("setting stream priority", + "flowID", f.id, "streamID", s.ID(), "priority", priority, "incremental", incremantal) priorityStream.SetPriority(priority) priorityStream.SetIncremental(incremantal) } diff --git a/session.go b/session.go index 593bf6f..a79c154 100644 --- a/session.go +++ b/session.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "io" + "log/slog" "sync" "github.com/mengelbart/roq/qlog" @@ -66,6 +67,7 @@ type Session struct { qlog *qlogger closeQlogOnce sync.Once + logger *slog.Logger } // NewSession creates a new roq session. QUIC connection is handled by roq. @@ -117,6 +119,7 @@ func newSession(conn Connection, acceptDatagrams bool, config *sessionConfig) *S ctx: ctx, cancelCtx: cancel, qlog: newQlogger(conn, config), + logger: config.logger, } } @@ -168,7 +171,7 @@ func (s *Session) NewSendFlow(id uint64) (*SendFlow, error) { } f := newFlow(s.conn, id, func() { s.sendFlows.delete(id) - }, s.qlog) + }, s.qlog, s.logger) if _, ok := s.sendFlows.getOrInsert(id, f); !ok { return nil, errDuplicateFlowID } @@ -186,7 +189,7 @@ func (s *Session) NewReceiveFlow(id uint64) (*ReceiveFlow, error) { } f := s.receiveFlowBuffer.pop(id) if f == nil { - f = newReceiveFlow(id, s.receiveBufferSize, s.qlog) + f = newReceiveFlow(id, s.receiveBufferSize, s.qlog, s.logger) } s.receiveFlows.set(id, f) return f, nil @@ -277,7 +280,7 @@ func (s *Session) receiveFlow(flowID uint64) *ReceiveFlow { if f, ok := s.receiveFlows.get(flowID); ok { return f } - return s.receiveFlowBuffer.getOrCreate(flowID, s.receiveBufferSize, s.qlog) + return s.receiveFlowBuffer.getOrCreate(flowID, s.receiveBufferSize, s.qlog, s.logger) } // HandleDatagram handles a datagram. If QUIC connection is handled by the