feat: add FlightRecorder sink for deferred low-level logging (#237)

* feat: add Buffer sink for deferred low-level logging

Buffer is a Sink that holds entries below a configured level in a
fixed-size in-memory ring, forwarding entries at or above the level
immediately. Flush replays the buffered entries to the wrapped sinks.

This lets a program run its logger at LevelDebug to capture verbose
entries while only emitting them on demand (e.g. on an error or
connection failure), with a configurable buffer size. When full, the
oldest entry is evicted.

* feat: add Flusher interface for deferred sinks

Flusher (Flush(ctx)) lets callers trigger a flush without depending on
the concrete Buffer type, which is handy when the buffer is created in
one place and flushed in another. Buffer implements it.

* refactor: rename buffer.go and buffer_test.go to flightrecorder.go and
flightrecorder_test.go

* refactor: rename Buffer to FlightRecorder

* refactor: make flight recording a Logger flag with auto-flush on error

Rework the flight recorder from a standalone Buffer/Flusher sink into a
Logger capability. Logger.FlightRecorder(size) enables a rolling
in-memory history of below-level entries; the history is flushed to the
sinks automatically when an entry at LevelError or above is logged, or
manually via Logger.Flush.

Enabling flight recording is independent of the Logger level, so it is
order-independent with Leveled(). Derived loggers (With, Named, Leveled)
share the underlying ring and capture their derived fields and names.

Replace the fill-tracking branch in the ring with a monotonic written
counter: the ring is allocated at full length and writes always land at
written%size, so the write path is branchless and drain derives the
entry count and oldest position from written.

* refactor: track ring position and length instead of unbounded counter

Replace the monotonic written counter with bounded pos and length fields,
so neither grows without limit and there is no risk of an integer-overflow
index. drain derives the oldest position from length and pos.

Author: aqandrew

flightrecorder.go

@@ -0,0 +1,114 @@
+package slog
+
+import (
+	"context"
+	"sync"
+)
+
+// FlightRecorder returns a Logger that records entries below its level in a
+// fixed-size, in-memory ring buffer instead of dropping them, while still
+// forwarding entries at or above its level to the sinks as usual. The recorded
+// entries are forwarded to the sinks ("flushed") automatically whenever an entry
+// at LevelError or above is logged, and can be flushed manually with
+// Logger.Flush.
+//
+// This keeps verbose (e.g. debug) logs available for diagnosing a failure
+// without emitting them during normal operation: run the Logger at LevelInfo
+// with a flight recorder, and the debug entries leading up to an error are
+// emitted only when the error occurs. When the buffer is full, the oldest
+// recorded entry is dropped to make room for the newest.
+//
+// Enabling flight recording is independent of the Logger's level, so
+// Make(sink).Leveled(LevelInfo).FlightRecorder(n) and
+// Make(sink).FlightRecorder(n).Leveled(LevelInfo) are equivalent. If size is
+// <= 0, no entries are recorded.
+//
+// The returned Logger and any Loggers derived from it share the same underlying
+// buffer.
+func (l Logger) FlightRecorder(size int) Logger {
+	l.flightRecorder = newFlightRecorder(size)
+	return l
+}
+
+// Flush forwards the entries currently held by the Logger's flight recorder to
+// its sinks, oldest first, and empties the buffer. It is a no-op when flight
+// recording is not enabled. Flushing also happens automatically when an entry at
+// LevelError or above is logged.
+func (l Logger) Flush(ctx context.Context) {
+	if l.flightRecorder != nil {
+		l.flightRecorder.flush(ctx, l.sinks)
+	}
+}
+
+// flightRecorder keeps a rolling history of entries in a fixed-size ring buffer
+// until they are flushed. It is safe for concurrent use.
+type flightRecorder struct {
+	capacity int
+
+	mu   sync.Mutex
+	ring []SinkEntry
+	// pos is the index the next entry is written to; length is the number of
+	// entries currently held. Both stay bounded by capacity, so neither grows
+	// without limit.
+	pos    int
+	length int
+}
+
+func newFlightRecorder(size int) *flightRecorder {
+	if size < 0 {
+		size = 0
+	}
+	return &flightRecorder{
+		capacity: size,
+		ring:     make([]SinkEntry, size),
+	}
+}
+
+// record stores e in the ring, overwriting the oldest entry once full.
+func (f *flightRecorder) record(e SinkEntry) {
+	if f.capacity == 0 {
+		return
+	}
+	f.mu.Lock()
+	f.ring[f.pos] = e
+	f.pos = (f.pos + 1) % f.capacity
+	if f.length < f.capacity {
+		f.length++
+	}
+	f.mu.Unlock()
+}
+
+// flush forwards the recorded entries to sinks, oldest first, and empties the
+// buffer.
+func (f *flightRecorder) flush(ctx context.Context, sinks []Sink) {
+	f.mu.Lock()
+	entries := f.drain()
+	f.mu.Unlock()
+
+	for _, e := range entries {
+		for _, s := range sinks {
+			s.LogEntry(ctx, e)
+		}
+	}
+}
+
+// drain returns the recorded entries oldest first and resets the buffer. It must
+// be called with f.mu held.
+func (f *flightRecorder) drain() []SinkEntry {
+	if f.length == 0 {
+		return nil
+	}
+	// Once the ring is full, pos points at the oldest entry; before that the
+	// oldest entry is at index 0.
+	oldest := 0
+	if f.length == f.capacity {
+		oldest = f.pos
+	}
+	entries := make([]SinkEntry, 0, f.length)
+	for i := 0; i < f.length; i++ {
+		entries = append(entries, f.ring[(oldest+i)%f.capacity])
+	}
+	f.length = 0
+	f.pos = 0
+	return entries
+}

flightrecorder_test.go

@@ -0,0 +1,271 @@
+package slog_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"cdr.dev/slog/v3"
+	"cdr.dev/slog/v3/internal/assert"
+)
+
+func TestFlightRecorder(t *testing.T) {
+	t.Parallel()
+
+	t.Run("ForwardsAtOrAboveLevel", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(8)
+
+		log.Info(bg, "info")
+		log.Warn(bg, "warn")
+
+		assert.Len(t, "forwarded", 2, s.entries)
+		assert.Equal(t, "first", "info", s.entries[0].Message)
+		assert.Equal(t, "second", "warn", s.entries[1].Message)
+	})
+
+	t.Run("BuffersBelowLevelUntilFlush", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(8)
+
+		log.Debug(bg, "debug1")
+		log.Debug(bg, "debug2")
+		// Nothing below the level is forwarded yet.
+		assert.Len(t, "held", 0, s.entries)
+
+		log.Flush(bg)
+		assert.Len(t, "flushed", 2, s.entries)
+		assert.Equal(t, "first", "debug1", s.entries[0].Message)
+		assert.Equal(t, "second", "debug2", s.entries[1].Message)
+	})
+
+	t.Run("InterleavesImmediateAndBuffered", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(8)
+
+		log.Debug(bg, "debug")
+		log.Info(bg, "info")
+		// The info entry is forwarded immediately; the debug entry waits.
+		assert.Len(t, "immediate", 1, s.entries)
+		assert.Equal(t, "immediate", "info", s.entries[0].Message)
+
+		log.Flush(bg)
+		assert.Len(t, "after flush", 2, s.entries)
+		assert.Equal(t, "buffered", "debug", s.entries[1].Message)
+	})
+
+	t.Run("FlushesAutomaticallyBeforeError", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(8)
+
+		log.Debug(bg, "debug1")
+		log.Debug(bg, "debug2")
+		// The buffered history is still held until the error triggers a flush.
+		assert.Len(t, "held", 0, s.entries)
+
+		log.Error(bg, "boom")
+		// The buffered debug entries precede the error that triggered the flush.
+		assert.Len(t, "flushed with error", 3, s.entries)
+		assert.Equal(t, "first", "debug1", s.entries[0].Message)
+		assert.Equal(t, "second", "debug2", s.entries[1].Message)
+		assert.Equal(t, "error last", "boom", s.entries[2].Message)
+	})
+
+	t.Run("CriticalFlushesBufferedHistory", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(8)
+
+		log.Debug(bg, "debug")
+		log.Critical(bg, "critical")
+
+		assert.Len(t, "flushed", 2, s.entries)
+		assert.Equal(t, "first", "debug", s.entries[0].Message)
+		assert.Equal(t, "second", "critical", s.entries[1].Message)
+	})
+
+	t.Run("EvictsOldestWhenFull", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(2)
+
+		log.Debug(bg, "debug1")
+		log.Debug(bg, "debug2")
+		log.Debug(bg, "debug3")
+
+		log.Flush(bg)
+		// debug1 was evicted; the two newest remain in order.
+		assert.Len(t, "flushed", 2, s.entries)
+		assert.Equal(t, "first", "debug2", s.entries[0].Message)
+		assert.Equal(t, "second", "debug3", s.entries[1].Message)
+	})
+
+	t.Run("EvictsOldestAcrossMultipleWraps", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(2)
+
+		log.Debug(bg, "debug1")
+		log.Debug(bg, "debug2")
+		log.Debug(bg, "debug3")
+		log.Debug(bg, "debug4")
+		log.Debug(bg, "debug5")
+
+		log.Flush(bg)
+		// Only the two newest survive after wrapping multiple times.
+		assert.Len(t, "flushed", 2, s.entries)
+		assert.Equal(t, "first", "debug4", s.entries[0].Message)
+		assert.Equal(t, "second", "debug5", s.entries[1].Message)
+	})
+
+	t.Run("FlushIsIdempotent", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(8)
+
+		log.Debug(bg, "debug")
+		log.Flush(bg)
+		log.Flush(bg)
+
+		assert.Len(t, "flushed once", 1, s.entries)
+	})
+
+	t.Run("ZeroSizeDropsBelowLevel", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(0)
+
+		log.Debug(bg, "debug")
+		log.Info(bg, "info")
+		log.Flush(bg)
+
+		// Only the info entry is forwarded; the debug entry was dropped.
+		assert.Len(t, "forwarded", 1, s.entries)
+		assert.Equal(t, "only", "info", s.entries[0].Message)
+	})
+
+	t.Run("OrderIndependentWithLeveled", func(t *testing.T) {
+		t.Parallel()
+
+		s1 := &fakeSink{}
+		s2 := &fakeSink{}
+		leveledFirst := slog.Make(s1).Leveled(slog.LevelInfo).FlightRecorder(8)
+		recorderFirst := slog.Make(s2).FlightRecorder(8).Leveled(slog.LevelInfo)
+
+		for _, log := range []slog.Logger{leveledFirst, recorderFirst} {
+			log.Debug(bg, "debug")
+			log.Info(bg, "info")
+		}
+
+		// Both loggers hold the debug entry and forward the info entry.
+		assert.Len(t, "leveled first immediate", 1, s1.entries)
+		assert.Len(t, "recorder first immediate", 1, s2.entries)
+
+		leveledFirst.Flush(bg)
+		recorderFirst.Flush(bg)
+
+		assert.Len(t, "leveled first flushed", 2, s1.entries)
+		assert.Len(t, "recorder first flushed", 2, s2.entries)
+		assert.Equal(t, "leveled first buffered", "debug", s1.entries[1].Message)
+		assert.Equal(t, "recorder first buffered", "debug", s2.entries[1].Message)
+	})
+
+	t.Run("DerivedLoggersShareRecorder", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		base := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(8)
+		derived := base.Named("sub").With(slog.F("key", "value"))
+
+		derived.Debug(bg, "debug")
+		// Flushing through the base logger emits the entry recorded via the
+		// derived logger, and the derived fields and names are captured.
+		base.Flush(bg)
+
+		assert.Len(t, "flushed", 1, s.entries)
+		assert.Equal(t, "message", "debug", s.entries[0].Message)
+		assert.Equal(t, "names", []string{"sub"}, s.entries[0].LoggerNames)
+		assert.Equal(t, "fields", slog.M(slog.F("key", "value")), s.entries[0].Fields)
+	})
+
+	t.Run("FlushForwardsToAllSinks", func(t *testing.T) {
+		t.Parallel()
+
+		s1 := &fakeSink{}
+		s2 := &fakeSink{}
+		log := slog.Make(s1, s2).Leveled(slog.LevelInfo).FlightRecorder(8)
+
+		log.Debug(bg, "debug")
+		log.Flush(bg)
+
+		assert.Len(t, "sink1", 1, s1.entries)
+		assert.Len(t, "sink2", 1, s2.entries)
+	})
+
+	t.Run("NoRecorderDropsBelowLevel", func(t *testing.T) {
+		t.Parallel()
+
+		s := &fakeSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo)
+
+		log.Debug(bg, "debug")
+		log.Info(bg, "info")
+		// Without a flight recorder, below-level entries are dropped and Flush
+		// is a no-op.
+		log.Flush(bg)
+
+		assert.Len(t, "forwarded", 1, s.entries)
+		assert.Equal(t, "only", "info", s.entries[0].Message)
+	})
+
+	t.Run("ConcurrentRecordAndFlush", func(t *testing.T) {
+		t.Parallel()
+
+		s := &lockedSink{}
+		log := slog.Make(s).Leveled(slog.LevelInfo).FlightRecorder(64)
+
+		var wg sync.WaitGroup
+		for i := 0; i < 8; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				for j := 0; j < 100; j++ {
+					log.Debug(bg, "debug")
+				}
+				log.Flush(bg)
+			}()
+		}
+		wg.Wait()
+
+		// The exact count depends on interleaving; the test only asserts that
+		// concurrent record and flush do not race or panic.
+		log.Flush(bg)
+	})
+}
+
+// lockedSink is a concurrency-safe sink for race detection tests.
+type lockedSink struct {
+	mu      sync.Mutex
+	entries []slog.SinkEntry
+}
+
+func (s *lockedSink) LogEntry(_ context.Context, e slog.SinkEntry) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.entries = append(s.entries, e)
+}
+
+func (s *lockedSink) Sync() {}