adapter/statsclient: read ring buffers incrementally

A ring-buffer stat could only be read whole. CopyEntryData copies every
thread's entire ring into a fresh allocation, so the cost of a read is the
size of the ring rather than the number of entries produced into it - which
is the wrong way round. A ring is sized for burst headroom, so making it big
enough to survive a stall makes it too expensive to poll: an 8192-entry ring
of 128-byte records is 1 MiB copied and allocated per thread per read, and at
the 10 ms cadence a fast producer needs, on eight workers, that is 800 MiB/s
of copying and garbage to deliver a few hundred entries. A 16M-entry ring is
2 GiB per thread per read and cannot be polled at all.

adapter.RingBufferWindowStat copies only what the producers appended since
the previous refresh, into buffers it already owns. On an 8192-entry ring
delivering 256 entries:

    RingBufferStat        483 us, 1 MiB, 5 allocations
    RingBufferWindowStat  4.1 us, none

118x, and the ratio grows with the ring: ring size no longer appears in the
cost of a read.

The read cursor is consumer-side state, a field on the RingBufferWindowStat the
caller holds rather than anything in the segment, which is mapped read-only. So
a window refreshes through UpdateDir like any other prepared entry - the switch
in UpdateEntryData already dispatches on the stat's dynamic type. CopyEntryData
deliberately never produces one: windowing needs a cursor carried between calls,
and only the consumer has anywhere to keep it.

Entries are located from the sequence and not from head. The two are
congruent - both start at zero and advance together on every commit - but a
producer stores head plainly and then publishes the sequence with a release
store, so a consumer reading the pair astride a commit sees a head one slot
ahead of the sequence that explains it. Positioning off head then delivers
the slot a worker is writing and skips the oldest live entry, silently and
only under load. The sequence is the only field of the two that carries an
ordering guarantee, and it is also the one that publishes the entry bytes.

A copy is validated against the producer afterwards. The segment's optimistic
lock covers directory changes, not ring data, so a worker is free to
overwrite the slots being copied while the copy runs - most easily in the
configuration this is for, a large window on a fast producer. Re-reading the
sequence after the copy identifies the entries that were overwritten
underneath it; they are the oldest of the window, so dropping them from the
front leaves exactly the ones that are still whole, and they are counted as
lost. Delivering a record that is half one entry and half another while
reporting no loss would be worse than losing it, because loss is visible and
tearing is not.

Lost and Pending are separate. Lost counts entries the producer overwrote
before the reader reached them, which are gone; Pending counts entries still
in the ring that MaxEntries held back, which the next read delivers. They
call for opposite responses - read again now, against the reader is not
keeping up - so a single "missed" figure would tell a rate-limited reader it
was losing data.

PrepareRingBuffer exists because PrepareDir cannot express this: it populates
every entry by way of CopyEntryData, so preparing a ring copies it once even
if every later read is windowed, and that copy repeats on every epoch change.
It is reached through adapter.RingBufferAPI rather than StatsAPI, because
serving a window means carrying that cursor between refreshes and a mock or a
v1 segment has nowhere to keep one; callers type-assert for it.

The three region bounds checks are now in one helper that both the windowed
and the whole-ring path use. They are what stands between a racing header and
an out-of-bounds read of the mapped segment, so a second copy of them is a
second chance to get one subtly wrong; go vet's unsafe.Pointer warnings for
this file drop from three to none as a side effect.

Tested against the synthetic v2 segment: first-refresh positioning with and
without SkipBacklog, delivering only what is new, unwrapping across the
ring's seam, exact loss on a lap, the cap and the drain loop that empties it,
re-sync on a sequence rewind, per-thread independence, a head published ahead
of its sequence, entries overwritten while the copy runs, and that a steady
refresh allocates nothing. The last two are reachable only by racing a real
producer, so an unexported hook between the copy and its validation stands in
for the worker; it is nil in production and costs one nil check per thread
per refresh.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: otroan

adapter/stats_api.go

@@ -188,15 +188,111 @@ type RingBufferStat struct {
 	Data    [][]byte // per-thread raw ring data
 }
 
-func (ScalarStat) isStat()          {}
-func (ErrorStat) isStat()           {}
-func (SimpleCounterStat) isStat()   {}
-func (CombinedCounterStat) isStat() {}
-func (NameStat) isStat()            {}
-func (EmptyStat) isStat()           {}
-func (GaugeStat) isStat()           {}
-func (HistogramLog2Stat) isStat()   {}
-func (RingBufferStat) isStat()      {}
+// RingBufferAPI is implemented by adapters that can read a ring-buffer stat
+// incrementally, and is separate from StatsAPI because not every adapter can:
+// serving a window means carrying a read cursor between refreshes. That cursor
+// is consumer-side state - the segment itself is mapped read-only - and a mock
+// or a v1 segment has nowhere to keep one. Callers type-assert for it.
+type RingBufferAPI interface {
+	// PrepareRingBuffer resolves one ring-buffer stat by name and returns a
+	// StatDir holding an incremental reader for it, to be refreshed with
+	// UpdateDir. maxEntries bounds entries delivered per thread per refresh, zero
+	// meaning the ring size; skipBacklog starts at the producer's head rather
+	// than at the oldest entry the ring still holds.
+	PrepareRingBuffer(name string, maxEntries uint32, skipBacklog bool) (*StatDir, error)
+}
+
+// RingBufferWindow is the run of entries one producer thread appended since a
+// consumer last read it.
+type RingBufferWindow struct {
+	// Entries holds Count entries of RingBufferConfig.EntrySize bytes each,
+	// oldest first and already unwrapped, so a consumer indexes it without
+	// knowing where the ring's seam fell.
+	//
+	// It aliases a buffer the stat owns and reuses, so it is only valid until
+	// the next refresh. Copy anything that must outlive that.
+	//
+	// The slice always starts at the buffer's origin and is capped to the entries
+	// this refresh delivered, so cap(Entries) is the room a refresh has to fill
+	// and there is no second buffer field to keep in step with it.
+	Entries []byte
+	Count   uint32
+
+	// FirstSeq is the producer sequence of Entries[0], and NextSeq the sequence
+	// the following read will start from. Both are absolute counts of entries
+	// this thread has ever written, so they stay meaningful across wraps.
+	FirstSeq uint64
+	NextSeq  uint64
+
+	// Lost counts entries the producer overwrote before this read reached them:
+	// data that is gone. Pending counts entries still in the ring that this read
+	// did not return because MaxEntries capped it: data that the next read will
+	// deliver.
+	//
+	// They are separate because they call for opposite responses. Pending means
+	// read again immediately; Lost means the reader is not keeping up and the
+	// gap is unrecoverable. A single "missed" figure would conflate a reader
+	// that is behind with one that is merely rate-limited.
+	Lost    uint64
+	Pending uint64
+}
+
+// RingBufferWindowStat reads a ring buffer incrementally: every refresh copies
+// only what the producers appended since the previous one, into buffers the stat
+// already owns.
+//
+// This is the difference between a cost proportional to entries produced and one
+// proportional to ring size. RingBufferStat copies the whole ring, for every
+// thread, into a fresh allocation on every read - so a ring sized for burst
+// headroom rather than for poll latency becomes unreadable long before it
+// becomes useful. A 16M-entry ring of 128-byte records is 2 GiB per thread per
+// read as a RingBufferStat, and the entries actually produced as this.
+//
+// Put one in a prepared StatDir entry's Data and refresh it with
+// StatsClient.UpdateDir, or let StatsClient.PrepareRingBuffer build both.
+// CopyEntryData never produces one: windowing needs a read cursor, and only the
+// consumer has it.
+//
+// The zero value is valid and self-initialising. The first refresh reads the
+// geometry, allocates the per-thread buffers, and positions the cursor - at the
+// oldest entry the ring still holds, or at the producer's head if SkipBacklog is
+// set - and returns no entries. SkipBacklog only decides where that first
+// refresh starts and is ignored afterwards; MaxEntries is honoured on every
+// refresh.
+type RingBufferWindowStat struct {
+	// MaxEntries bounds how many entries one refresh delivers per thread, and so
+	// bounds both the buffer this stat allocates and the work one refresh does.
+	// Zero means the ring size, which is the largest window that can ever be
+	// available. A consumer draining a fast producer wants this small enough to
+	// bound a single read and to loop while Pending is non-zero.
+	//
+	// It is read on every refresh, so raising or lowering it between refreshes
+	// takes effect on the next one; the buffers grow to match and are not shrunk.
+	MaxEntries uint32
+
+	// SkipBacklog starts the first read at the producer's head rather than at the
+	// oldest entry still in the ring, so a consumer that wants only what happens
+	// from now on does not first have to read and discard a ring of history.
+	SkipBacklog bool
+
+	Config  RingBufferConfig
+	Threads []RingBufferThreadMeta
+	Schema  []byte
+
+	// Windows holds one window per producer thread, in thread order.
+	Windows []RingBufferWindow
+}
+
+func (ScalarStat) isStat()            {}
+func (ErrorStat) isStat()             {}
+func (SimpleCounterStat) isStat()     {}
+func (CombinedCounterStat) isStat()   {}
+func (NameStat) isStat()              {}
+func (EmptyStat) isStat()             {}
+func (GaugeStat) isStat()             {}
+func (HistogramLog2Stat) isStat()     {}
+func (RingBufferStat) isStat()        {}
+func (*RingBufferWindowStat) isStat() {}
 
 func (s ScalarStat) IsZero() bool {
 	return s == 0
@@ -352,6 +448,25 @@ func (s RingBufferStat) Type() StatType {
 	return RingBuffer
 }
 
+func (s *RingBufferWindowStat) IsZero() bool {
+	return s.Config.NThreads == 0 || s.Config.EntrySize == 0
+}
+
+func (s *RingBufferWindowStat) Type() StatType {
+	return RingBuffer
+}
+
+func (s *RingBufferWindowStat) String() string {
+	var b strings.Builder
+	fmt.Fprintf(&b, "\n  config: entry_size=%d, ring_size=%d, threads=%d, schema_version=%d, schema_size=%d",
+		s.Config.EntrySize, s.Config.RingSize, s.Config.NThreads, s.Config.SchemaVersion, s.Config.SchemaSize)
+	for i, w := range s.Windows {
+		fmt.Fprintf(&b, "\n  thread[%d]: entries=%d first_seq=%d next_seq=%d lost=%d pending=%d",
+			i, w.Count, w.FirstSeq, w.NextSeq, w.Lost, w.Pending)
+	}
+	return b.String()
+}
+
 func (s RingBufferStat) String() string {
 	var b strings.Builder
 	fmt.Fprintf(&b, "\n  config: entry_size=%d, ring_size=%d, threads=%d, schema_version=%d, schema_size=%d",

adapter/statsclient/statsclient.go

@@ -769,3 +769,77 @@ func symlinkItem(full adapter.Stat, item uint32, dst *adapter.Stat) bool {
 	}
 	return false
 }
+
+var _ adapter.RingBufferAPI = (*StatsClient)(nil)
+
+// PrepareRingBuffer resolves one ring-buffer stat by name and returns a StatDir
+// holding an incremental reader for it, to be refreshed with UpdateDir.
+//
+// It exists because PrepareDir cannot express this: PrepareDir populates every
+// entry it prepares by way of CopyEntryData, which for a ring buffer copies the
+// whole ring. On a ring sized for burst headroom that one copy at prepare time is
+// the largest allocation the process makes, and it is repeated on every epoch
+// change. Here nothing is read but the directory entry itself; the first
+// UpdateDir reads the geometry and sizes the reader's buffers from maxEntries.
+//
+// maxEntries bounds entries per thread per refresh, and so bounds both the
+// buffers allocated and the work one refresh does; zero means the ring size. A
+// consumer draining a fast producer should set it and loop while any window
+// reports Pending. skipBacklog starts at the producer's head rather than at the
+// oldest entry still in the ring.
+//
+// name is matched exactly, not as a pattern: this returns one entry or an error,
+// because a caller reading a specific producer's records has nothing sensible to
+// do with a second ring that happened to match.
+func (sc *StatsClient) PrepareRingBuffer(name string, maxEntries uint32, skipBacklog bool) (*adapter.StatDir, error) {
+	sc.accessLock.RLock()
+	defer sc.accessLock.RUnlock()
+
+	if !sc.isConnected() {
+		return nil, adapter.ErrStatsDisconnected
+	}
+
+	accessEpoch := sc.accessStart()
+	if accessEpoch == 0 {
+		return nil, adapter.ErrStatsAccessFailed
+	}
+
+	vector := sc.GetDirectoryVector()
+	if vector == nil {
+		return nil, fmt.Errorf("failed to prepare ring buffer: directory vector is nil")
+	}
+
+	want := []byte(name)
+	var entry *adapter.StatEntry
+	vecLen := *(*uint32)(vectorLen(vector))
+	for i := uint32(0); i < vecLen; i++ {
+		// Compared in place: GetStatDirOnIndex clones every name it walks past,
+		// and on a real directory that is thousands of allocations to find one
+		// entry.
+		_, dirType, ok := sc.StatDirOnIndexMatches(vector, i, want)
+		if !ok {
+			continue
+		}
+		if dirType != adapter.RingBuffer {
+			return nil, fmt.Errorf("stat %q is %v, not a ring buffer", name, dirType)
+		}
+		entry = &adapter.StatEntry{
+			StatIdentifier: adapter.StatIdentifier{Index: i, Name: want},
+			Type:           adapter.RingBuffer,
+			Data: &adapter.RingBufferWindowStat{
+				MaxEntries:  maxEntries,
+				SkipBacklog: skipBacklog,
+			},
+		}
+		break
+	}
+	if entry == nil {
+		return nil, fmt.Errorf("ring buffer stat %q not found", name)
+	}
+
+	if !sc.accessEnd(accessEpoch) {
+		return nil, adapter.ErrStatsDataBusy
+	}
+
+	return &adapter.StatDir{Epoch: accessEpoch, Entries: []adapter.StatEntry{*entry}}, nil
+}