fix(errortracking): address bouncer double-count, cap bypass, and async stack capture

- Rollover now delivers priorCount-1 (suppressed-only) to avoid double-
  counting the first sighting already emitted; silent rollover when no
  suppressions occurred (priorCount==1)
- Re-check cap after pruneLocked: drop new key as pass-through when all
  entries are still within the window, enforcing the memory bound
- Add SyncCapture handler wrapper that pre-captures the call stack in
  the emitting goroutine and stores it as a slog attr; Handler.Handle
  reads the attr before falling back to runtime.Callers, fixing full-
  stack dedup across an async handler boundary

Author: pducolin

pkg/util/log/errortracking/bouncer.go

@@ -16,9 +16,11 @@ import (
 // time as firstSeen. Subsequent observations of the same key inside the
 // same window are suppressed (return suppressed=true) and the count is
 // incremented. After the window elapses, the next observation returns
-// suppressed=false carrying the prior window's total count (so the
-// delivered record's Count field reflects every sighting in that
-// window), then resets the entry to a fresh count=1.
+// suppressed=false carrying the suppressed count of the prior window
+// (priorTotal-1, since the first sighting was already delivered), then
+// resets the entry to a fresh count=1. If no sightings were suppressed
+// (priorTotal==1), the rollover is silent and the next observation is
+// treated as a fresh first sighting (count=1).
 //
 // The key is an opaque uint64 — the caller is responsible for choosing
 // it. Today's caller (Handler.Handle) hashes the captured stack PCs
@@ -29,10 +31,9 @@ import (
 // (agent_telemetry.errortracking.bouncer_window_seconds) was chosen so a hot
 // bug path collapses to one record per quarter-hour — long enough to avoid
 // flooding the wire, short enough that operators see new error patterns
-// promptly. Count on the delivered record carries the total sightings of
-// the prior window so suppressed duplicates are not lost; a hot bug path
-// with N sightings per window ships one record per window with Count=N,
-// rather than N records each carrying Count=1.
+// promptly. A hot bug path with N sightings per window ships Count=1 at
+// first sighting, then Count=N-1 on rollover (the suppressed portion).
+// Summing both gives N without double-counting the first sighting.
 //
 // Bouncer is purpose-built rather than reusing the global
 // rate.Sometimes wrapper in pkg/util/log/log_limit.go — that primitive
@@ -73,19 +74,24 @@ func NewBouncer(window time.Duration, maxEntries int) *Bouncer {
 }
 
 // Observe records a sighting of the given key at now and returns
-// whether the caller should suppress the record, the running count of
-// sightings represented by the next delivered record (≥ 1), and the
-// firstSeen time of the current window.
+// whether the caller should suppress the record, the count to attach to
+// the next delivered record (≥ 1), and the firstSeen time of the
+// current window.
 //
 // The first sighting of a key in a window returns suppressed=false,
 // count=1, firstSeen=now. Subsequent sightings inside the same window
 // return suppressed=true with an incrementing count and the original
 // firstSeen. When the window elapses since firstSeen, the next
-// sighting returns suppressed=false with count=priorWindowCount (the
-// total sightings observed during the elapsed window), then resets
-// the entry to a fresh count=1. The wire payload's Count field on the
-// delivered record therefore carries the total occurrences of the
-// prior window — no suppressed sightings are lost.
+// sighting returns suppressed=false with count=priorSuppressed (the
+// number of sightings that were suppressed in the elapsed window, i.e.
+// priorTotal-1), then resets the entry to a fresh count=1. If no
+// sightings were suppressed (priorTotal==1), the rollover is silent:
+// Observe returns suppressed=false, count=1, as though it were a fresh
+// first sighting, avoiding a Count=0 delivery.
+//
+// This design ensures consumers can sum Count fields without
+// double-counting: the first delivery carries Count=1, and the rollover
+// carries the remainder.
 //
 // When window is non-positive, Observe is a pass-through: returns
 // suppressed=false, count=1, firstSeen=now.
@@ -99,25 +105,35 @@ func (b *Bouncer) Observe(key uint64, now time.Time) (suppressed bool, count uin
 
 	if e, ok := b.entries[key]; ok {
 		if now.Sub(e.firstSeen) > b.window {
-			// Window elapsed — the delivered record carries the prior
-			// window's total count so suppressed duplicates are not
-			// lost. Reset the entry to a fresh window.
+			// Window elapsed. Deliver the suppressed portion of the prior
+			// window (priorTotal-1) so callers can sum without
+			// double-counting the first sighting already delivered.
+			// When nothing was suppressed (priorTotal==1), skip the
+			// rollover and treat this as a fresh first sighting.
 			priorCount := e.count
 			e.firstSeen = now
 			e.lastSeen = now
 			e.count = 1
-			return false, priorCount, now
+			if priorCount <= 1 {
+				return false, 1, now
+			}
+			return false, priorCount - 1, now
 		}
 		e.lastSeen = now
 		e.count++
 		return true, e.count, e.firstSeen
 	}
 
-	// New key. Prune opportunistically if we're approaching the cap so
-	// the map doesn't grow unboundedly under pathological churn (many
-	// unique keys, each appearing once).
+	// New key. Prune opportunistically if at the cap so the map doesn't
+	// grow unboundedly under pathological churn (many unique keys, each
+	// appearing once). Re-check after pruning: if all entries are still
+	// within the window, pruneLocked removes nothing and the cap must be
+	// enforced by dropping the new key (pass-through, not suppressed).
 	if len(b.entries) >= b.maxEntries {
 		b.pruneLocked(now)
+		if len(b.entries) >= b.maxEntries {
+			return false, 1, now
+		}
 	}
 	b.entries[key] = &bouncerEntry{
 		firstSeen: now,

pkg/util/log/errortracking/bouncer_test.go

@@ -43,22 +43,22 @@ func TestBouncer_ResetsAfterWindow(t *testing.T) {
 	b.Observe(0xCAFE, t0.Add(time.Minute)) // count=2 (suppressed)
 
 	// Step past the window — next sighting MUST NOT be suppressed and
-	// MUST carry the prior window's total count (2) rather than
-	// resetting to 1. That preserves the suppressed-duplicate count on
-	// the delivered wire record. The entry is then reset to a fresh
+	// MUST carry the suppressed count of the prior window (1 = total-1)
+	// rather than the full total, to avoid double-counting the first
+	// sighting already delivered. The entry is then reset to a fresh
 	// window internally.
 	t2 := t0.Add(16 * time.Minute)
 	suppressed, count, firstSeen := b.Observe(0xCAFE, t2)
 	assert.False(t, suppressed, "sighting after window MUST NOT be suppressed")
-	assert.Equal(t, uint32(2), count, "post-window sighting count (want prior-window total)")
+	assert.Equal(t, uint32(1), count, "post-window count (want suppressed-only: total-1)")
 	assert.True(t, firstSeen.Equal(t2), "post-window firstSeen = %v, want %v", firstSeen, t2)
 }
 
 // TestBouncer_WindowElapseCarriesPriorCount exercises the
 // suppressed-count carry-forward contract end-to-end: a hot bug path
-// with N sightings per window must deliver one record per window with
-// Count=N (the total occurrences of the prior window), not N records
-// each carrying Count=1.
+// with N sightings per window must deliver Count=1 at first sighting
+// and Count=N-1 on rollover (the suppressed sightings), so consumers
+// summing both get N without double-counting the first sighting.
 func TestBouncer_WindowElapseCarriesPriorCount(t *testing.T) {
 	b := NewBouncer(15*time.Minute, 0)
 	var key uint64 = 0xABCDEF
@@ -79,11 +79,11 @@ func TestBouncer_WindowElapseCarriesPriorCount(t *testing.T) {
 	assert.True(t, suppressed, "sighting inside window MUST be suppressed")
 	assert.Equal(t, uint32(3), count, "third sighting count")
 
-	// Window elapses; next sighting is delivered with the prior
-	// window's total (3).
+	// Window elapses; next sighting delivers the suppressed count of the
+	// prior window (2 = total(3) - first-already-delivered(1)).
 	suppressed, count, _ = b.Observe(key, t0.Add(20*time.Minute))
 	assert.False(t, suppressed, "window elapsed; sighting must be delivered")
-	assert.Equal(t, uint32(3), count, "delivered record count (want prior-window total)")
+	assert.Equal(t, uint32(2), count, "rollover count (want suppressed-only: total-1)")
 
 	// Subsequent sighting in the new window — suppressed, count starts
 	// from the post-reset 1 and increments to 2.
@@ -152,3 +152,26 @@ func TestBouncer_PrunesNearCap(t *testing.T) {
 	defer b.mu.Unlock()
 	assert.LessOrEqual(t, len(b.entries), b.maxEntries, "entries exceeded cap after prune")
 }
+
+// TestBouncer_CapEnforced_WhenAllWithinWindow verifies that when the cap
+// is reached and pruneLocked removes nothing (all entries still within the
+// window), new keys are dropped rather than growing the map past maxEntries.
+// The dropped observation is returned as a pass-through (suppressed=false,
+// count=1) so the record still reaches the consumer without dedup tracking.
+func TestBouncer_CapEnforced_WhenAllWithinWindow(t *testing.T) {
+	b := NewBouncer(time.Hour, 4) // large window: no entries expire
+	t0 := time.Now()
+	for pc := uint64(1); pc <= 4; pc++ {
+		b.Observe(pc, t0)
+	}
+
+	// New unique key inserted while all existing entries are still within window.
+	t1 := t0.Add(time.Minute)
+	suppressed, count, _ := b.Observe(uint64(5), t1)
+	assert.False(t, suppressed, "dropped key must not be reported as suppressed")
+	assert.Equal(t, uint32(1), count, "dropped key must return count=1 (pass-through)")
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	assert.LessOrEqual(t, len(b.entries), b.maxEntries, "map must not exceed maxEntries")
+}

pkg/util/log/errortracking/handler.go

@@ -19,6 +19,13 @@ import (
 // between runtime.Callers and the user call site.
 const stackSearchBuf = MaxStackFrames + 16
 
+// stackPCsAttrKey is the slog attribute key used by SyncCapture to carry
+// stack PCs captured in the emitting goroutine across an async handler
+// boundary. Handler.Handle reads this attr before falling back to
+// runtime.Callers so full-stack dedup is preserved even when the handler
+// runs in an async worker goroutine.
+const stackPCsAttrKey = "errortracking.pcs"
+
 var _ slog.Handler = (*Handler)(nil)
 
 // Handler is an slog.Handler that captures records at level >= Error and
@@ -121,28 +128,42 @@ func (h *Handler) Handle(_ context.Context, r slog.Record) error {
 		return nil
 	}
 
-	// Capture the full goroutine stack and anchor the slice at r.PC —
-	// the call-site PC already computed by the caller (slog or the
-	// pkg/util/log Wrapper). This is correct regardless of how many
-	// wrapper frames sit between user code and Handle, whereas a fixed
-	// skip would land inside the wrapper internals when the path goes
-	// through pkg/util/log.Error → Wrapper → Handler.Handle.
-	var buf [stackSearchBuf]uintptr
-	n := runtime.Callers(1, buf[:]) // skip runtime.Callers itself
+	// Capture the full call stack anchored at r.PC.
+	//
+	// Preferred path: read PCs pre-captured by a SyncCapture wrapper that
+	// ran synchronously in the emitting goroutine before any async
+	// dispatch. Present when the wiring layer uses NewSyncCapture.
+	//
+	// Fallback path: call runtime.Callers here. Correct only when Handle
+	// is called synchronously in the emitting goroutine (i.e. the handler
+	// is NOT behind handlers.NewAsync or any other async boundary).
 	var pcs [MaxStackFrames]uintptr
 	var pcsLen int
-	for i := 0; i < n; i++ {
-		if buf[i] == r.PC {
-			pcsLen = copy(pcs[:], buf[i:n])
-			break
+	r.Attrs(func(a slog.Attr) bool {
+		if a.Key == stackPCsAttrKey {
+			if captured, ok := a.Value.Any().([]uintptr); ok {
+				pcsLen = copy(pcs[:], captured)
+			}
+			return false
+		}
+		return true
+	})
+	if pcsLen == 0 {
+		var buf [stackSearchBuf]uintptr
+		n := runtime.Callers(1, buf[:]) // skip runtime.Callers itself
+		for i := 0; i < n; i++ {
+			if buf[i] == r.PC {
+				pcsLen = copy(pcs[:], buf[i:n])
+				break
+			}
+		}
+		if pcsLen == 0 && r.PC != 0 {
+			// r.PC not found in the captured slice (async path or
+			// synthetic record): store it alone so downstream consumers
+			// always have at least a valid call-site frame.
+			pcs[0] = r.PC
+			pcsLen = 1
 		}
-	}
-	if pcsLen == 0 && r.PC != 0 {
-		// r.PC not found in the captured slice (e.g. synthetic record
-		// in tests): store it alone so downstream consumers always
-		// have at least a valid call-site frame.
-		pcs[0] = r.PC
-		pcsLen = 1
 	}
 
 	var count uint32
@@ -214,6 +235,66 @@ func hashPCs(pcs []uintptr) uint64 {
 	return h.Sum64()
 }
 
+// SyncCapture wraps any slog.Handler and pre-captures the goroutine's full
+// call stack in the emitting goroutine before forwarding the record to the
+// inner handler. This solves the async-boundary problem: when the inner
+// handler is dispatched by an async worker goroutine, runtime.Callers can
+// no longer see the original caller's frames. SyncCapture must be placed in
+// the synchronous layer of the logger chain (before handlers.NewAsync or any
+// other async wrapper) so that Handle is called in the same goroutine that
+// emitted the log record.
+//
+// Handler.Handle reads the pre-captured PCs from the stackPCsAttrKey slog
+// attribute added by SyncCapture.Handle, bypassing its own runtime.Callers
+// call when the attr is present.
+type SyncCapture struct {
+	inner slog.Handler
+}
+
+var _ slog.Handler = (*SyncCapture)(nil)
+
+// NewSyncCapture returns a SyncCapture that wraps inner. Install the returned
+// handler in the synchronous layer of the logger chain so it is always called
+// from the emitting goroutine.
+func NewSyncCapture(inner slog.Handler) *SyncCapture {
+	return &SyncCapture{inner: inner}
+}
+
+// Enabled delegates to the inner handler.
+func (s *SyncCapture) Enabled(ctx context.Context, level slog.Level) bool {
+	return s.inner.Enabled(ctx, level)
+}
+
+// Handle captures the current goroutine's call stack anchored at r.PC,
+// attaches the PCs as a stackPCsAttrKey slog attribute, then forwards to
+// the inner handler. Must be called from the goroutine that emitted r.
+func (s *SyncCapture) Handle(ctx context.Context, r slog.Record) error {
+	if r.Level < slog.LevelError {
+		return s.inner.Handle(ctx, r)
+	}
+	var buf [stackSearchBuf]uintptr
+	n := runtime.Callers(1, buf[:])
+	for i := 0; i < n; i++ {
+		if buf[i] == r.PC {
+			pcs := make([]uintptr, n-i)
+			copy(pcs, buf[i:n])
+			r.AddAttrs(slog.Any(stackPCsAttrKey, pcs))
+			break
+		}
+	}
+	return s.inner.Handle(ctx, r)
+}
+
+// WithAttrs returns a SyncCapture wrapping the inner handler's WithAttrs result.
+func (s *SyncCapture) WithAttrs(attrs []slog.Attr) slog.Handler {
+	return &SyncCapture{inner: s.inner.WithAttrs(attrs)}
+}
+
+// WithGroup returns a SyncCapture wrapping the inner handler's WithGroup result.
+func (s *SyncCapture) WithGroup(name string) slog.Handler {
+	return &SyncCapture{inner: s.inner.WithGroup(name)}
+}
+
 // WithAttrs is a required slog.Handler interface method. Attrs are not
 // shipped to the wire; this method is a required interface no-op.
 func (h *Handler) WithAttrs(_ []slog.Attr) slog.Handler {