From 044f2cc448c407f7b34531b96246842aaabccce9 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Thu, 6 Aug 2026 15:25:45 -0300 Subject: [PATCH 1/7] feat: complete the Stats snapshot with BatchHeld and BatchesFlushed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Milestone 4.1. Fills the two remaining gaps in the observability contract and pins that contract with tests. BatchHeld closes a visibility hole rather than adding a nice-to-have. An item that has left the intake queue but not yet entered a worker was previously invisible: Queued no longer counted it and InFlight did not yet. That is exactly the state a partial timer batch sits in, and the state a flushed batch sits in while blocked on the unbuffered worker dispatch. So the field that would have told an operator "batches are waiting on saturated workers" did not exist. received() is therefore a transfer, not a decrement: intakePending down, batchHeld up. dispatched() releases batchHeld only after the send to a worker completes, because while that send is blocked the batch really is still aggregator-held. Releasing before the send would hide the saturation case the field exists to show. BatchesFlushed counts batches rather than items, giving mean batch size when paired with Completed. That is the coalescing signal for tuning BatchInterval: a mean far below BatchSize means windows are closing on the timer, not on size. Both counters live exclusively on the aggregation and worker paths. Verified that publish() still touches only reserve/accept/rollback, and benchmarked against Phase 3: no enqueue regression (geomean -3%, four of five cases statistically indistinguishable). Tests pin ownership transitions across all three flush paths, mutual exclusion of terminal outcomes, and allocation-free reads. They also state the consistency boundary honestly: a live snapshot is a valid ownership observation, but only a terminally drained one is a valid accounting assertion. Both new mechanisms are sabotage-verified — removing the ownership transfer or the dispatch accounting makes the corresponding test fail. --- docs/improvements/thresholds.md | 7 +- pkg/batcher/batcher.go | 8 ++ pkg/batcher/stats.go | 52 +++++-- pkg/batcher/stats_test.go | 232 ++++++++++++++++++++++++++++++++ 4 files changed, 287 insertions(+), 12 deletions(-) create mode 100644 pkg/batcher/stats_test.go diff --git a/docs/improvements/thresholds.md b/docs/improvements/thresholds.md index 61b4332..a39a6d8 100644 --- a/docs/improvements/thresholds.md +++ b/docs/improvements/thresholds.md @@ -41,9 +41,12 @@ the blocking signal rather than timings. | Recovery wrapper allocations, non-panic | exactly 0 | `TestRecoveredPanicAddsNoSteadyStateAllocations` | | Scenario recorder allocations per item | ≤ 1 total, and must not grow with run length | `TestHarnessRecorderDoesNotAllocatePerItem` (`test/scenario`) | | Goroutines per running batcher, `n=1` | exactly 2 | `TestGoroutineBudgetPerRunningBatcher` | +| `Stats()` allocations | exactly 0 | `TestStatsIsAllocationFree` | -`Stats()` is a fixed set of atomic loads returning a value type, so it has no -allocation gate of its own; the `Add` gate covers the hot path that matters. +`Stats()` returns a value type built from atomic loads plus one queue length check +under the queue's existing mutex. It now carries its own allocation gate, because +metrics scraping runs continuously in production and generating garbage per scrape +would be a real cost. The recorder threshold is not "exactly 0" because `AllocsPerItem` measures the whole pipeline, including Batcher's own per-batch allocations, not just the diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go index 4c04cc7..fb2bafa 100644 --- a/pkg/batcher/batcher.go +++ b/pkg/batcher/batcher.go @@ -223,11 +223,13 @@ func (b *Batcher[T]) Stats() Stats { IntakePending: b.counters.intakePending.Load(), PublishersInGate: b.gate.inGate(), Queued: int64(b.input.length()), + BatchHeld: b.counters.batchHeld.Load(), InFlight: b.counters.inFlight.Load(), Accepted: b.counters.accepted.Load(), Completed: b.counters.completed.Load(), Failed: b.counters.failed.Load(), Panicked: b.counters.panicked.Load(), + BatchesFlushed: b.counters.batchesFlushed.Load(), Rejected: b.counters.rejected.Load(), DroppedErrors: b.counters.droppedErrors.Load(), } @@ -356,7 +358,13 @@ func (b *Batcher[T]) run() { stopTimer() + // The send blocks until a worker is free. Until it returns, the batch is + // still aggregator-held, which is the state BatchHeld exists to expose: + // releasing the counter before the send would hide a batch waiting on + // saturated workers. batches <- items + + b.counters.dispatched(len(items)) } take := func(item T) { diff --git a/pkg/batcher/stats.go b/pkg/batcher/stats.go index a346ab9..af68b9c 100644 --- a/pkg/batcher/stats.go +++ b/pkg/batcher/stats.go @@ -41,6 +41,12 @@ type Stats struct { // the aggregator. This is the queue depth to alert on. Queued int64 + // BatchHeld is items owned by the aggregator: they have left the intake queue + // but have not yet entered a worker. This includes items accumulating in the + // current partial batch and a flushed batch blocked on the unbuffered worker + // dispatch channel. It is distinct from Queued and InFlight. + BatchHeld int64 + // InFlight is items currently inside a processor call. It is updated by workers // on batch boundaries, not on the enqueue path, so observability does not add // another contended producer atomic. @@ -62,6 +68,12 @@ type Stats struct { // expired. Rejected uint64 + // BatchesFlushed counts batches emitted by the aggregator, in batches rather + // than items. Together with Completed it gives the mean batch size, which is + // the coalescing figure to watch when tuning BatchInterval: a mean far below + // BatchSize means the window is closing on the timer rather than on size. + BatchesFlushed uint64 + // DroppedErrors counts diagnostics discarded because the Errors() buffer was // full. A non-zero value means diagnostics are being lost, not that batches // failed: it is a signal that Errors() is not being drained fast enough. @@ -76,15 +88,17 @@ type Stats struct { // contended atomics per Add measured up to 4x the cost of one, so nothing is added // to the hot path without a reason. type counters struct { - pending atomic.Int64 - intakePending atomic.Int64 - accepted atomic.Uint64 - inFlight atomic.Int64 - completed atomic.Uint64 - failed atomic.Uint64 - panicked atomic.Uint64 - rejected atomic.Uint64 - droppedErrors atomic.Uint64 + pending atomic.Int64 + intakePending atomic.Int64 + accepted atomic.Uint64 + batchHeld atomic.Int64 + inFlight atomic.Int64 + completed atomic.Uint64 + failed atomic.Uint64 + panicked atomic.Uint64 + batchesFlushed atomic.Uint64 + rejected atomic.Uint64 + droppedErrors atomic.Uint64 } // reserve claims a drain obligation before publication. @@ -112,9 +126,27 @@ func (c *counters) accept() { c.accepted.Add(1) } -// received records the aggregator taking an item out of the queue. +// received records the aggregator taking an item out of the queue and into the +// batch it is currently accumulating. +// +// This is a transfer, not a decrement: the item stops being queued and starts +// being aggregator-held, so both counters move together. Decrementing intake +// alone would make the item invisible in every field between leaving the queue +// and entering a worker. func (c *counters) received(n int) { c.intakePending.Add(int64(-n)) + c.batchHeld.Add(int64(n)) +} + +// dispatched records a batch leaving the aggregator for a worker. +// +// batchHeld is released here rather than when the processor returns, because +// inFlight takes ownership at that point; counting both would double-count the +// batch. It runs once per batch on the aggregation path, so it adds no per-item +// producer cost. +func (c *counters) dispatched(n int) { + c.batchHeld.Add(int64(-n)) + c.batchesFlushed.Add(1) } // terminal records exactly one outcome for a finished batch and releases its diff --git a/pkg/batcher/stats_test.go b/pkg/batcher/stats_test.go new file mode 100644 index 0000000..e3c198c --- /dev/null +++ b/pkg/batcher/stats_test.go @@ -0,0 +1,232 @@ +package batcher_test + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/NSXBet/batcher/pkg/batcher" + "github.com/stretchr/testify/require" +) + +// Stats snapshot contract tests. +// +// Stats is intentionally an eventually consistent diagnostic interface: each +// field is read independently, so callers cannot use one live snapshot as a +// transactional accounting assertion. These tests pin the meaningful contract: +// ownership transitions are represented, terminal accounting converges after the +// drain, and observing the snapshot is allocation-free and cannot perturb Add. + +// TestStatsShowsAggregatorHeldPartialBatch distinguishes the three ownership +// states. A partial batch has left the intake queue but cannot yet be in flight, +// because no worker receives it until the timer/size/shutdown flush. That item must +// be visible as BatchHeld rather than disappearing between Queued and InFlight. +func TestStatsShowsAggregatorHeldPartialBatch(t *testing.T) { + t.Parallel() + + b := batcher.New( + batcher.WithBatchSize[int](100), + batcher.WithBatchInterval[int](time.Hour), // only shutdown can flush it + batcher.WithProcessor(batcher.NoOpProcessor[int]), + ) + + b.Add(1) + + require.Eventually(t, func() bool { + stats := b.Stats() + + return stats.BatchHeld == 1 && stats.Queued == 0 && stats.InFlight == 0 + }, 5*time.Second, time.Millisecond, + "an item received by the aggregator but not dispatched must be BatchHeld") + + before := b.Stats() + + require.Equal(t, int64(1), before.Pending) + require.Equal(t, int64(1), before.BatchHeld) + require.Equal(t, uint64(0), before.BatchesFlushed) + + require.NoError(t, b.Close()) + + after := b.Stats() + + require.Zero(t, after.Pending) + require.Zero(t, after.BatchHeld) + require.Zero(t, after.InFlight) + require.Equal(t, uint64(1), after.BatchesFlushed, + "the shutdown partial batch is still a flushed batch") + require.Equal(t, uint64(1), after.Completed) +} + +// TestStatsCountsBatchesFlushedAcrossEveryFlushPath pins that BatchesFlushed is a +// batch counter, independent of whether a flush happened by size, timer, or final +// shutdown. This is the coalescing signal operators need when choosing a window. +func TestStatsCountsBatchesFlushedAcrossEveryFlushPath(t *testing.T) { + t.Parallel() + + const ( + batchSize = 3 + interval = 30 * time.Millisecond + ) + + b := batcher.New( + batcher.WithBatchSize[int](batchSize), + batcher.WithBatchInterval[int](interval), + batcher.WithProcessor(batcher.NoOpProcessor[int]), + ) + + // Size flush. + for i := range batchSize { + b.Add(i) + } + + require.Eventually(t, func() bool { + return b.Stats().BatchesFlushed >= 1 + }, 5*time.Second, time.Millisecond) + + // Timer flush. + b.Add(100) + b.Add(101) + + require.Eventually(t, func() bool { + return b.Stats().BatchesFlushed >= 2 + }, 5*time.Second, time.Millisecond) + + // Shutdown partial flush. + b.Add(200) + + require.NoError(t, b.Close()) + + stats := b.Stats() + + require.Equal(t, uint64(3), stats.BatchesFlushed) + require.Equal(t, uint64(batchSize+2+1), stats.Completed) + require.Zero(t, stats.Pending) + require.Zero(t, stats.BatchHeld) + require.Zero(t, stats.InFlight) +} + +// TestStatsTerminalOutcomesAreMutuallyExclusive pins accounting at the only point +// where a multi-field equality is meaningful: after terminal drain, when no +// publisher or worker can still be moving an item between fields. +func TestStatsTerminalOutcomesAreMutuallyExclusive(t *testing.T) { + t.Parallel() + + var calls atomic.Int64 + failure := errors.New("processor failed") + + b := batcher.New( + batcher.WithBatchSize[int](1), + batcher.WithBatchInterval[int](time.Millisecond), + batcher.WithProcessor(func([]int) error { + switch calls.Add(1) { + case 1: + return nil + case 2: + return failure + default: + panic("poison batch") + } + }), + ) + + // Do not let diagnostics block the scenario: outcomes, not diagnostics, + // are under test here. + go func() { + for range b.Errors() { + } + }() + + for i := range 3 { + b.Add(i) + } + + require.NoError(t, b.Shutdown(context.Background())) + + stats := b.Stats() + + require.Equal(t, uint64(3), stats.Accepted) + require.Equal(t, uint64(1), stats.Completed) + require.Equal(t, uint64(1), stats.Failed) + require.Equal(t, uint64(1), stats.Panicked) + require.Equal(t, stats.Accepted, + stats.Completed+stats.Failed+stats.Panicked, + "terminal categories are mutually exclusive and exhaustive after drain") + require.Zero(t, stats.Pending) + require.Zero(t, stats.IntakePending) + require.Zero(t, stats.BatchHeld) + require.Zero(t, stats.InFlight) + require.Zero(t, stats.PublishersInGate) +} + +// TestStatsIsAllocationFree pins the read-side performance contract. Queued takes +// the queue mutex for a length check, but neither that lock nor the atomic loads +// may allocate. Metrics scraping must never generate garbage. +func TestStatsIsAllocationFree(t *testing.T) { + // Deliberately not parallel: AllocsPerRun requires exclusive GOMAXPROCS. + b := batcher.New( + batcher.WithBatchSize[int](100), + batcher.WithBatchInterval[int](time.Hour), + batcher.WithProcessor(batcher.NoOpProcessor[int]), + ) + + defer func() { require.NoError(t, b.Close()) }() + + for range 100 { + b.Add(1) + } + + allocs := testing.AllocsPerRun(2_000, func() { + _ = b.Stats() + }) + + require.Zero(t, allocs, "Stats must not allocate") +} + +// TestStatsIsEventuallyConsistent, rather than pretending live counters are a +// transaction, makes the boundary explicit. While the processor is blocked, the +// snapshot has a stable meaningful ownership state. Once it is released and +// terminal drain completes, the conservation equality holds. Callers must not +// infer that arbitrary intermediate snapshots satisfy the latter equality. +func TestStatsIsEventuallyConsistent(t *testing.T) { + t.Parallel() + + entered := make(chan struct{}) + release := make(chan struct{}) + + b := batcher.New( + batcher.WithBatchSize[int](1), + batcher.WithBatchInterval[int](time.Millisecond), + batcher.WithProcessor(func([]int) error { + close(entered) + <-release + + return nil + }), + ) + + b.Add(1) + + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("processor did not start") + } + + live := b.Stats() + + require.Equal(t, int64(1), live.InFlight, + "a stable live snapshot exposes the item in its current ownership state") + require.Equal(t, int64(1), live.Pending) + + close(release) + require.NoError(t, b.Shutdown(context.Background())) + + settled := b.Stats() + + require.Zero(t, settled.Pending) + require.Equal(t, settled.Accepted, + settled.Completed+settled.Failed+settled.Panicked, + "only a terminally drained snapshot is a valid accounting assertion") +} From 1268906a4902986c465404ba70cc4e7affff8f90 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Thu, 6 Aug 2026 16:20:01 -0300 Subject: [PATCH 2/7] perf: size batch slices from observed demand MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Milestone 4.2. The plan gated this on evidence, so the evidence came first: at a 1ms window with BatchSize=1000 the aggregator reserved 55,944 B per flush to hold a single item, and 559,944 B at BatchSize=10000. That is far above the 2 KB/flush threshold for justifying the work, so the milestone was triggered rather than skipped. Capacity now follows a recent-max estimator rounded to a power of two and clamped to BatchSize. Recent-max rather than a mean is the whole point: an EWMA of the mean was rejected during planning because on traffic alternating between tiny and full batches it allocated MORE than doing nothing, since the mean sits between the modes and every large batch grows from a too-small start. capacity_test.go pins that pattern explicitly so the mistake cannot return. Two corrections were needed to satisfy the +2% no-regression budget: - Starting at the floor made a size-triggered workload grow its first batch repeatedly, measured as +2.23% against full capacity — over budget for a workload the optimisation was never meant to help. The estimator now starts pessimistic at BatchSize and adapts down on the first observed batch, so full-batch workloads are exactly neutral while sparse ones still improve ~98%. - An earlier attempt pinned capacity permanently once a batch filled. That fixed the control but broke decay: a batcher that had been busy would hold an oversized reservation forever. Removed, so capacity falls again after a window of sparse traffic. Final results versus full capacity: sparse -97.9%, small batches -93.3%, full batches -0.00%, alternating +1.12%, burst-after-idle -72.2%, bimodal -3.45%. Worst case is inside the +2% gate. Also fixes a data race this work exposed. Config() returned the live *Config, so a caller could mutate a running batcher's batch size or interval from another goroutine; the aggregator read those fields per batch. Reading BatchSize at goroutine start made the race observable under -race where it previously hid. run() now snapshots its configuration once at start, and Config() returns a value copy. Options are construction-time configuration, so mutating them afterwards was never coherent — the option tests that did so are updated to configure at New. This was already listed as a Phase 5.1 API break; it is fixed here because leaving a known race in place until a later phase is not defensible. Slice ownership is unchanged: retention_test.go asserts a processor may keep its batch slice and that no two retained batches share backing storage, which is what ruled out pooling. --- docs/improvements/plan-perf.md | 6 +- docs/improvements/thresholds.md | 28 +++- pkg/batcher/batcher.go | 53 ++++-- pkg/batcher/capacity.go | 165 +++++++++++++++++++ pkg/batcher/capacity_bench_test.go | 112 +++++++++++++ pkg/batcher/capacity_test.go | 187 ++++++++++++++++++++++ pkg/batcher/options_test.go | 57 ++----- pkg/batcher/retention_test.go | 150 +++++++++++++++++ test/scenario/allocation_evidence_test.go | 72 +++++++++ 9 files changed, 772 insertions(+), 58 deletions(-) create mode 100644 pkg/batcher/capacity.go create mode 100644 pkg/batcher/capacity_bench_test.go create mode 100644 pkg/batcher/capacity_test.go create mode 100644 pkg/batcher/retention_test.go create mode 100644 test/scenario/allocation_evidence_test.go diff --git a/docs/improvements/plan-perf.md b/docs/improvements/plan-perf.md index eb96bc8..000ed2d 100644 --- a/docs/improvements/plan-perf.md +++ b/docs/improvements/plan-perf.md @@ -1080,8 +1080,10 @@ rather than intuition. sealing instead of panicking; `Close()` no longer abandoning the drain at its deadline; **`Len()` now counting accepted-but-unfinished work including in-flight batches, so it can be non-zero with an empty queue**; `Join`'s - clarified quiescence-snapshot contract; the removal of the `rill` and `chann` - dependencies; **the module's minimum Go version moving from 1.22.4 to 1.25.0** + clarified quiescence-snapshot contract; **`Config()` returning a value snapshot + instead of a live mutable pointer**, which removes the ability to reconfigure a + running batcher and the data race that came with it; the removal of the `rill` and + `chann` dependencies; **the module's minimum Go version moving from 1.22.4 to 1.25.0** (required by `golang.org/x/sys` v0.46.0, pulled in when dependencies were modernised in 2.1); and any `ProvideBatcherInFX` signature change. - **`Config()` must stop exposing live mutable state.** It currently returns the diff --git a/docs/improvements/thresholds.md b/docs/improvements/thresholds.md index a39a6d8..0866035 100644 --- a/docs/improvements/thresholds.md +++ b/docs/improvements/thresholds.md @@ -107,10 +107,30 @@ zero leaked. ## Conditional gates (Milestone 4.2 only) -| Gate | Threshold | -| ------------------------------------------ | ---------------------------- | -| Sparse-window allocated bytes/flush | > 2 KB/flush to justify work | -| Allocation regression ceiling, any scenario | ≤ +2% allocated bytes | +| Gate | Threshold | Measured | +| ------------------------------------------- | ---------------------------- | --------------- | +| Sparse-window allocated bytes/flush | > 2 KB/flush to justify work | 55,944 B/flush | +| Allocation regression ceiling, any scenario | ≤ +2% allocated bytes | +1.12% (worst) | + +Milestone 4.2 was **triggered and implemented**. Sparse-window waste measured far +above the justification threshold: at a 1ms window with `BatchSize=1000` the +aggregator reserved 55,944 B/flush to hold a single item, and 559,944 B/flush at +`BatchSize=10000`. + +Results per workload, adaptive versus the previous full-capacity strategy: + +| Workload | Change | +| ----------------------- | ------- | +| steady sparse | -97.9% | +| small batches | -93.3% | +| full batches (control) | -0.00% | +| alternating sparse/full | +1.12% | +| burst after idle | -72.2% | +| bimodal | -3.45% | + +The alternating case is the one that killed the rejected EWMA estimator, which +allocated *more* than doing nothing there. Recent-max keeps it inside the +2% +budget. Enforced by `BenchmarkCapacity*` and `capacity_test.go`. The second gate must hold for alternating sparse/full, burst-after-idle, and bimodal workloads, not just the sparse case the optimisation targets. An diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go index fb2bafa..ee34c42 100644 --- a/pkg/batcher/batcher.go +++ b/pkg/batcher/batcher.go @@ -138,9 +138,17 @@ func (b *Batcher[T]) Start() { }) } -// Config returns the batcher's configuration. -func (b *Batcher[T]) Config() *Config[T] { - return b.config +// Config returns a copy of the batcher's configuration. +// +// It is a snapshot, not a handle. Returning the live pointer let a caller mutate a +// running batcher's batch size, interval or processor from another goroutine, which +// is a data race against the aggregation loop and could change batching semantics +// mid-batch. Options are meant to be applied at construction; mutating them +// afterwards was never coherent, so this closes the hole rather than documenting it. +// +// Callers that need to change configuration should construct a new Batcher. +func (b *Batcher[T]) Config() Config[T] { + return *b.config } // Add enqueues an item. It is the compatibility fast path and returns no error. @@ -296,13 +304,28 @@ func (b *Batcher[T]) Errors() <-chan error { func (b *Batcher[T]) run() { defer close(b.stopped) - batches := make(chan []T) + // Snapshot the configuration once, at start, and use only the snapshot from here + // on. Config is a plain struct the caller still holds a pointer to, so reading it + // per batch races with any option applied after Start. Freezing it also matches + // the semantics callers already rely on: batch size and interval are fixed for + // the lifetime of a running batcher, so re-reading them could never have taken + // effect coherently mid-batch anyway. + var ( + batchSize = b.config.BatchSize + batchInterval = b.config.BatchInterval + workers = b.config.Concurrency + ) + + if batchSize < 1 { + batchSize = DefaultBatchSize + } - workers := b.config.Concurrency if workers < 1 { workers = 1 } + batches := make(chan []T) + var processing sync.WaitGroup processing.Add(workers) @@ -326,9 +349,10 @@ func (b *Batcher[T]) run() { }() var ( - batch []T - timer *time.Timer - timerC <-chan time.Time + batch []T + timer *time.Timer + timerC <-chan time.Time + capacities = newCapacityEstimator(batchSize) ) stopTimer := func() { @@ -365,16 +389,21 @@ func (b *Batcher[T]) run() { batches <- items b.counters.dispatched(len(items)) + capacities.observe(len(items)) } take := func(item T) { if len(batch) == 0 { - batch = make([]T, 0, b.config.BatchSize) + // The capacity estimator is owned by this aggregation goroutine. It + // reduces timer-flush allocation pressure without pooling or reusing a + // slice after the processor receives it, so callers may still retain the + // batch slice exactly as before. + batch = make([]T, 0, capacities.capacity()) if timer == nil { - timer = time.NewTimer(b.config.BatchInterval) + timer = time.NewTimer(batchInterval) } else { - timer.Reset(b.config.BatchInterval) + timer.Reset(batchInterval) } timerC = timer.C @@ -382,7 +411,7 @@ func (b *Batcher[T]) run() { batch = append(batch, item) - if len(batch) >= b.config.BatchSize { + if len(batch) >= batchSize { flush() } } diff --git a/pkg/batcher/capacity.go b/pkg/batcher/capacity.go new file mode 100644 index 0000000..5a68aee --- /dev/null +++ b/pkg/batcher/capacity.go @@ -0,0 +1,165 @@ +package batcher + +// capacityEstimator chooses the initial capacity for the next batch slice. +// +// The aggregator allocates one slice per batch. Reserving BatchSize every time is +// correct but wasteful whenever batches close on the timer rather than on size: a +// 1ms window with BatchSize=1000 was measured reserving ~56KB per flush to hold a +// single item, and ~560KB at BatchSize=10000. +// +// The estimator tracks the maximum batch size seen in a recent window and rounds it +// up to a power of two, so capacity follows observed demand instead of the +// configured ceiling. +// +// Why recent-max rather than a mean: +// +// An EWMA of the mean was evaluated during planning and rejected. On traffic +// alternating between tiny and full batches it allocated *more* than doing nothing, +// because the mean sits between the two modes: every large batch then grows from a +// too-small start, and each growth reallocates and copies. Tracking the maximum +// means a workload containing large batches keeps capacity sized for them, so +// alternating traffic behaves like the full-capacity strategy rather than worse. +// +// Concurrency: this is owned exclusively by the aggregator goroutine and updated +// once per flush. It deliberately has no locks and no atomics; nothing else may +// touch it. +type capacityEstimator struct { + // batchSize is the configured ceiling. Capacity never exceeds it, so a + // processor that retains its slice never holds more than today's worst case. + batchSize int + + // ceiling is the current power-of-two capacity, derived from recentMax. It + // starts at batchSize for the first batch so a full-batch workload never pays + // append growth before the estimator has any evidence. The first observed + // sparse batch then adapts it downward. + ceiling int + + // initialized records whether the first observed batch has set a data-driven + // ceiling. It distinguishes the initial full-capacity safety value from a + // ceiling reached because recent demand genuinely saturated BatchSize. + initialized bool + + // recentMax is the largest batch observed in the current evaluation window. + recentMax int + + // seen counts observations in the current window. + seen int +} + +const ( + // capacityFloor keeps very sparse traffic from reallocating on a batch that + // happens to be slightly larger than the last. 16 items is small enough that + // the floor itself is not meaningful waste. + capacityFloor = 16 + + // capacityWindow is how many flushes are observed before the ceiling is + // re-derived. Long enough that a single small batch cannot collapse capacity + // for a busy batcher, short enough that a batcher which quiets down releases + // its oversized reservation promptly. + capacityWindow = 32 + + // capacityHeadroom scales the observed maximum before rounding, so a workload + // sitting just above a power of two does not reallocate on every batch. + capacityHeadroom = 5 + capacityDivisor = 4 // headroom/divisor == 1.25x +) + +func newCapacityEstimator(batchSize int) *capacityEstimator { + if batchSize < 1 { + batchSize = 1 + } + + return &capacityEstimator{ + batchSize: batchSize, + // Start pessimistic: reserve the configured ceiling until there is evidence + // that batches are smaller. Starting small instead made a size-triggered + // workload pay repeated append growth on its very first batch, measured as a + // 2.2% allocation regression against the full-capacity strategy. + ceiling: batchSize, + } +} + +// capacity returns the initial capacity to allocate for the next batch. +func (e *capacityEstimator) capacity() int { + return e.ceiling +} + +// observe records the size of a completed batch. +// +// The ceiling rises immediately when a batch exceeds it, because under-allocating a +// large batch costs a reallocation and copy right away. It falls only at window +// boundaries, so capacity decays gradually rather than oscillating. +func (e *capacityEstimator) observe(size int) { + if size > e.recentMax { + e.recentMax = size + } + + if !e.initialized { + // First evidence: replace the pessimistic starting ceiling outright, so a + // sparse workload stops reserving BatchSize after a single batch rather than + // waiting for a full evaluation window. + e.initialized = true + e.ceiling = clampCapacity(roundUpPow2(scaleHeadroom(size)), e.batchSize) + } else if size > e.ceiling { + e.ceiling = clampCapacity(roundUpPow2(scaleHeadroom(size)), e.batchSize) + } + + e.seen++ + + if e.seen < capacityWindow { + return + } + + e.ceiling = clampCapacity(roundUpPow2(scaleHeadroom(e.recentMax)), e.batchSize) + e.recentMax = 0 + e.seen = 0 +} + +// scaleHeadroom applies the headroom factor with integer arithmetic. +func scaleHeadroom(n int) int { + if n <= 0 { + return 0 + } + + return n * capacityHeadroom / capacityDivisor +} + +// roundUpPow2 returns the smallest power of two at or above n. +func roundUpPow2(n int) int { + if n <= capacityFloor { + return capacityFloor + } + + p := capacityFloor + for p < n { + next := p << 1 + // Stop before overflow; the caller clamps to batchSize anyway. + if next <= p { + return p + } + + p = next + } + + return p +} + +// clampCapacity keeps capacity within [capacityFloor, batchSize]. +// +// The upper clamp is a contract, not an optimisation: it bounds what a processor +// that retains its batch slice can hold. +func clampCapacity(n, batchSize int) int { + if n > batchSize { + return batchSize + } + + if n < capacityFloor { + if batchSize < capacityFloor { + return batchSize + } + + return capacityFloor + } + + return n +} diff --git a/pkg/batcher/capacity_bench_test.go b/pkg/batcher/capacity_bench_test.go new file mode 100644 index 0000000..85c141c --- /dev/null +++ b/pkg/batcher/capacity_bench_test.go @@ -0,0 +1,112 @@ +package batcher + +import "testing" + +// Capacity strategy comparison for Milestone 4.2. +// +// The aggregator reserves make([]T, 0, BatchSize) for every batch. When a batch +// closes on the timer holding far fewer items, the difference is pure waste. These +// benchmarks compare the current strategy against the recent-max estimator across +// the workloads the plan requires, including the ones that must NOT regress. +// +// EWMA-of-mean was rejected during planning because it regressed alternating +// traffic; recentMaxCapacity is benchmarked here against that same pattern to +// confirm the replacement does not repeat the mistake. + +type benchItem struct { + _ [64]byte +} + +const benchBatchSize = 1_000 + +var benchSink []benchItem + +// fullCapacity is today's strategy: always reserve BatchSize. +func fullCapacity(sizes []int) { + for _, n := range sizes { + batch := make([]benchItem, 0, benchBatchSize) + + for range n { + batch = append(batch, benchItem{}) + } + + benchSink = batch + } +} + +// adaptive uses the recent-max estimator under test. +func adaptive(sizes []int) { + est := newCapacityEstimator(benchBatchSize) + + for _, n := range sizes { + batch := make([]benchItem, 0, est.capacity()) + + for range n { + batch = append(batch, benchItem{}) + } + + est.observe(len(batch)) + + benchSink = batch + } +} + +func repeat(pattern []int, times int) []int { + out := make([]int, 0, len(pattern)*times) + for range times { + out = append(out, pattern...) + } + + return out +} + +func benchBoth(b *testing.B, sizes []int) { + b.Run("full", func(b *testing.B) { + b.ReportAllocs() + + for range b.N { + fullCapacity(sizes) + } + }) + + b.Run("adaptive", func(b *testing.B) { + b.ReportAllocs() + + for range b.N { + adaptive(sizes) + } + }) +} + +// BenchmarkCapacitySteadySparse is the target workload: every batch holds one item. +func BenchmarkCapacitySteadySparse(b *testing.B) { + benchBoth(b, repeat([]int{1}, 200)) +} + +// BenchmarkCapacitySmallBatches models a moderate rate closing on the timer. +func BenchmarkCapacitySmallBatches(b *testing.B) { + benchBoth(b, repeat([]int{50}, 200)) +} + +// BenchmarkCapacityFullBatches is the control: reserved capacity is fully used, so +// adaptive must not do worse. +func BenchmarkCapacityFullBatches(b *testing.B) { + benchBoth(b, repeat([]int{benchBatchSize}, 100)) +} + +// BenchmarkCapacityAlternating is the pattern that killed EWMA-of-mean. +func BenchmarkCapacityAlternating(b *testing.B) { + benchBoth(b, repeat([]int{1, benchBatchSize}, 100)) +} + +// BenchmarkCapacityBurstAfterIdle models a long sparse period then a burst. +func BenchmarkCapacityBurstAfterIdle(b *testing.B) { + sizes := append(repeat([]int{1}, 150), repeat([]int{benchBatchSize}, 50)...) + benchBoth(b, sizes) +} + +// BenchmarkCapacityBimodal models mostly small batches with regular large ones. +func BenchmarkCapacityBimodal(b *testing.B) { + pattern := []int{20, 20, 20, 20, 20, 20, 20, 20, 20, 800} + benchBoth(b, repeat(pattern, 20)) +} diff --git a/pkg/batcher/capacity_test.go b/pkg/batcher/capacity_test.go new file mode 100644 index 0000000..a8c11a4 --- /dev/null +++ b/pkg/batcher/capacity_test.go @@ -0,0 +1,187 @@ +package batcher + +import "testing" + +// Capacity estimator unit tests. +// +// These pin the properties the aggregator relies on, independently of timing: the +// bound that protects a retained slice, adaptation in both directions, and the +// behaviour on the two workloads that made the rejected EWMA estimator worse than +// doing nothing. + +// TestCapacityNeverExceedsBatchSize pins the retention bound. A processor may keep +// its batch slice, so capacity must never exceed the configured ceiling; otherwise +// a retained slice could hold more than today's worst case. +func TestCapacityNeverExceedsBatchSize(t *testing.T) { + t.Parallel() + + for _, batchSize := range []int{1, 7, 16, 100, 1_000} { + est := newCapacityEstimator(batchSize) + + if got := est.capacity(); got > batchSize { + t.Fatalf("batchSize=%d: initial capacity %d exceeds ceiling", batchSize, got) + } + + // Feed sizes at and beyond the ceiling, including absurd ones. + for _, size := range []int{0, 1, batchSize, batchSize * 4, batchSize * 1000} { + est.observe(size) + + if got := est.capacity(); got > batchSize { + t.Fatalf("batchSize=%d: capacity %d exceeds ceiling after observing %d", + batchSize, got, size) + } + } + } +} + +// TestCapacityStartsAtBatchSize pins the pessimistic start. +// +// Starting small measured as a 2.2% allocation regression on size-triggered +// workloads, because the first full batch grew repeatedly before any evidence +// existed. The first batch therefore reserves the ceiling. +func TestCapacityStartsAtBatchSize(t *testing.T) { + t.Parallel() + + est := newCapacityEstimator(1_000) + + if got := est.capacity(); got != 1_000 { + t.Fatalf("initial capacity = %d, want 1000 (no evidence yet)", got) + } +} + +// TestCapacityAdaptsDownAfterFirstSparseBatch pins that a sparse workload stops +// reserving the ceiling immediately, rather than waiting a full evaluation window. +func TestCapacityAdaptsDownAfterFirstSparseBatch(t *testing.T) { + t.Parallel() + + est := newCapacityEstimator(1_000) + + est.observe(1) + + got := est.capacity() + + if got >= 1_000 { + t.Fatalf("capacity = %d after a 1-item batch, want well below 1000", got) + } + + if got < capacityFloor { + t.Fatalf("capacity = %d, want at least the floor %d", got, capacityFloor) + } +} + +// TestCapacityRisesImmediatelyOnLargerBatch pins that growth is not deferred to a +// window boundary. Under-allocating a large batch costs a reallocation right away, +// so the ceiling must react on the spot. +func TestCapacityRisesImmediatelyOnLargerBatch(t *testing.T) { + t.Parallel() + + est := newCapacityEstimator(1_000) + + est.observe(1) + + small := est.capacity() + + est.observe(500) + + grown := est.capacity() + + if grown <= small { + t.Fatalf("capacity did not rise: %d -> %d after observing 500", small, grown) + } + + if grown < 500 { + t.Fatalf("capacity = %d, want at least the observed 500", grown) + } +} + +// TestCapacityFullWorkloadStaysAtBatchSize pins the no-regression control. +// Repeated full batches keep the recent maximum at the configured ceiling, so the +// estimator must continue allocating BatchSize and never pay append growth. +func TestCapacityFullWorkloadStaysAtBatchSize(t *testing.T) { + t.Parallel() + + est := newCapacityEstimator(1_000) + + for range capacityWindow * 2 { + est.observe(1_000) + + if got := est.capacity(); got != 1_000 { + t.Fatalf("capacity = %d after a full batch, want 1000", got) + } + } +} + +// TestCapacityDecaysAfterWindow pins that a batcher which quiets down releases its +// oversized reservation, so a formerly busy batcher does not hold capacity forever. +func TestCapacityDecaysAfterWindow(t *testing.T) { + t.Parallel() + + est := newCapacityEstimator(1_000) + + // Reach the ceiling under a busy workload. + est.observe(1_000) + + busy := est.capacity() + + if busy != 1_000 { + t.Fatalf("capacity = %d, want 1000 after a full batch", busy) + } + + // Two full windows of sparse traffic. + for range capacityWindow * 2 { + est.observe(1) + } + + quiet := est.capacity() + + if quiet >= busy { + t.Fatalf("capacity did not decay: %d -> %d after sustained sparse traffic", + busy, quiet) + } +} + +// TestCapacityHandlesAlternatingTraffic is the regression guard for the estimator +// that was rejected during planning. +// +// An EWMA of the mean allocated MORE than doing nothing on this pattern, because the +// mean sits between the two modes and every large batch grew from a too-small start. +// Tracking the maximum must keep capacity sized for the large mode. +func TestCapacityHandlesAlternatingTraffic(t *testing.T) { + t.Parallel() + + est := newCapacityEstimator(1_000) + + for range 50 { + est.observe(1) + est.observe(1_000) + } + + if got := est.capacity(); got != 1_000 { + t.Fatalf("capacity = %d on alternating traffic, want 1000: a mean-based "+ + "estimator would sit between the modes and reallocate every large batch", + got) + } +} + +// TestRoundUpPow2 pins the rounding helper, including its floor and idempotence. +func TestRoundUpPow2(t *testing.T) { + t.Parallel() + + cases := map[int]int{ + -5: capacityFloor, + 0: capacityFloor, + 1: capacityFloor, + 16: capacityFloor, + 17: 32, + 32: 32, + 33: 64, + 63: 64, + 65: 128, + } + + for input, want := range cases { + if got := roundUpPow2(input); got != want { + t.Errorf("roundUpPow2(%d) = %d, want %d", input, got, want) + } + } +} diff --git a/pkg/batcher/options_test.go b/pkg/batcher/options_test.go index 844b35f..aa3376d 100644 --- a/pkg/batcher/options_test.go +++ b/pkg/batcher/options_test.go @@ -10,67 +10,44 @@ import ( ) func TestWithProcessor(t *testing.T) { - // ARRANGE - b := batcher.New[test.BatchItem]() - - var ( - pr batcher.Processor[test.BatchItem] - processor batcher.Processor[test.BatchItem] - ) - - processor = func(_ []test.BatchItem) error { + processor := batcher.Processor[test.BatchItem](func(_ []test.BatchItem) error { return nil - } + }) - // ACT - batcher.WithProcessor(processor)(b) + // Options are construction-time configuration. Applying one after New starts + // the batcher races its aggregation goroutine and was never a coherent runtime + // reconfiguration API. + b := batcher.New(batcher.WithProcessor(processor)) + defer func() { _ = b.Close() }() - // ASSERT - pr = b.Config().ProcessorFunc - require.IsType(t, processor, pr) + require.IsType(t, processor, b.Config().ProcessorFunc) } func TestWithBatchSize(t *testing.T) { - // ARRANGE - b := batcher.New[test.BatchItem]() - - // ACT - batcher.WithBatchSize[test.BatchItem](1000)(b) + b := batcher.New(batcher.WithBatchSize[test.BatchItem](1000)) + defer func() { _ = b.Close() }() - // ASSERT require.Equal(t, 1000, b.Config().BatchSize) t.Run("WithBatchSize - zero size", func(t *testing.T) { - // ARRANGE - b := batcher.New[test.BatchItem]() + b := batcher.New(batcher.WithBatchSize[test.BatchItem](0)) + defer func() { _ = b.Close() }() - // ACT - batcher.WithBatchSize[test.BatchItem](0)(b) - - // ASSERT require.Equal(t, batcher.DefaultBatchSize, b.Config().BatchSize) }) } func TestWithBatchInterval(t *testing.T) { - // ARRANGE - b := batcher.New[test.BatchItem]() - duration := 1 * time.Second - - // ACT - batcher.WithBatchInterval[test.BatchItem](duration)(b) + duration := time.Second + b := batcher.New(batcher.WithBatchInterval[test.BatchItem](duration)) + defer func() { _ = b.Close() }() - // ASSERT require.Equal(t, duration, b.Config().BatchInterval) t.Run("WithBatchInterval - zero duration", func(t *testing.T) { - // ARRANGE - b := batcher.New[test.BatchItem]() - - // ACT - batcher.WithBatchInterval[test.BatchItem](0)(b) + b := batcher.New(batcher.WithBatchInterval[test.BatchItem](0)) + defer func() { _ = b.Close() }() - // ASSERT require.Equal(t, batcher.DefaultBatchInterval, b.Config().BatchInterval) }) } diff --git a/pkg/batcher/retention_test.go b/pkg/batcher/retention_test.go new file mode 100644 index 0000000..e213da9 --- /dev/null +++ b/pkg/batcher/retention_test.go @@ -0,0 +1,150 @@ +package batcher_test + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/NSXBet/batcher/pkg/batcher" + "github.com/stretchr/testify/require" +) + +// TestProcessorMayRetainBatchSlices is the contract test for adaptive batch +// capacity. +// +// The capacity estimator changes how large each batch slice starts, but it must not +// change slice *ownership*: the processor receives a slice it may keep, and no later +// batch may write through it. This is the property that ruled out pooling during +// planning, so it is asserted directly rather than assumed. +// +// The test retains every batch, mutates the retained copies after the fact, and then +// verifies that nothing the batcher did afterwards corrupted them. +func TestProcessorMayRetainBatchSlices(t *testing.T) { + t.Parallel() + + var ( + mu sync.Mutex + retained [][]int + ) + + b := batcher.New( + // Small batches with a short interval: many flushes, so the estimator adapts + // repeatedly while batches are being retained. + batcher.WithBatchSize[int](8), + batcher.WithBatchInterval[int](2*time.Millisecond), + batcher.WithProcessor(func(items []int) error { + // Deliberately keep the slice the batcher handed us, without copying. + mu.Lock() + retained = append(retained, items) + mu.Unlock() + + return nil + }), + ) + + const total = 500 + + for i := range total { + b.Add(i) + + // Vary arrival pacing so batches close on both size and timer, driving the + // estimator up and down during the run. + if i%37 == 0 { + time.Sleep(3 * time.Millisecond) + } + } + + require.NoError(t, b.Shutdown(context.Background())) + + mu.Lock() + defer mu.Unlock() + + // Every accepted item must appear exactly once across the retained slices, and + // each slice must still hold what it held when the processor saw it. + seen := make(map[int]int, total) + count := 0 + + for i, batch := range retained { + require.NotEmpty(t, batch, "batch %d: no empty batch may be emitted", i) + require.LessOrEqual(t, len(batch), 8, + "batch %d: len must never exceed BatchSize", i) + + for _, item := range batch { + seen[item]++ + count++ + } + } + + require.Equal(t, total, count, + "retained slices must still contain every accepted item exactly once; "+ + "a reused backing array would show duplicates or missing values") + + for i := range total { + require.Equal(t, 1, seen[i], "item %d appeared %d times", i, seen[i]) + } +} + +// TestRetainedBatchSlicesAreNotAliased pins the stronger property: two retained +// batches must not share backing storage. If they did, mutating one would corrupt +// the other, which is exactly the failure mode a pooled or reused slice introduces. +func TestRetainedBatchSlicesAreNotAliased(t *testing.T) { + t.Parallel() + + var ( + mu sync.Mutex + retained [][]int + ) + + b := batcher.New( + batcher.WithBatchSize[int](4), + batcher.WithBatchInterval[int](2*time.Millisecond), + batcher.WithProcessor(func(items []int) error { + mu.Lock() + retained = append(retained, items) + mu.Unlock() + + return nil + }), + ) + + for i := range 100 { + b.Add(i) + } + + require.NoError(t, b.Shutdown(context.Background())) + + mu.Lock() + defer mu.Unlock() + + require.Greater(t, len(retained), 1, "need several batches to compare") + + // Snapshot the retained contents, then overwrite every retained slice. If any + // two shared storage, the overwrite of a later batch would change an earlier one. + snapshots := make([][]int, len(retained)) + for i, batch := range retained { + snapshots[i] = append([]int(nil), batch...) + } + + for i, batch := range retained { + for j := range batch { + batch[j] = -(i + 1) + } + } + + for i, batch := range retained { + for j := range batch { + require.Equal(t, -(i + 1), batch[j], + "batch %d slot %d was changed by a write to another batch, so the "+ + "batches share backing storage", i, j) + } + } + + // And the snapshots prove the original contents were distinct per batch. + total := 0 + for _, snap := range snapshots { + total += len(snap) + } + + require.Equal(t, 100, total, "every accepted item must have been delivered once") +} diff --git a/test/scenario/allocation_evidence_test.go b/test/scenario/allocation_evidence_test.go new file mode 100644 index 0000000..a51e9b4 --- /dev/null +++ b/test/scenario/allocation_evidence_test.go @@ -0,0 +1,72 @@ +package scenario_test + +import ( + "testing" + "time" + "unsafe" + + "github.com/NSXBet/batcher/test/scenario" +) + +// TestSparseWindowAllocationEvidence measures allocated bytes per flush for the +// workload Milestone 4.2 targets: a small window with a large BatchSize, where +// every batch closes on the timer holding far fewer items than it reserved +// capacity for. +// +// This exists to decide whether 4.2 should be implemented at all. The plan gates it +// on sparse-window allocation pressure above 2 KB/flush; the aggregator allocates +// make([]T, 0, BatchSize) per batch, so the theoretical waste is +// (BatchSize - actual) * sizeof(T) per flush. +// +// Report-only: it prints evidence and never fails, because it is an input to a +// decision rather than a gate. +func TestSparseWindowAllocationEvidence(t *testing.T) { + t.Parallel() + + if testing.Short() { + t.Skip("measurement only; runs in the full suite") + } + + cases := []struct { + name string + batchSize int + window time.Duration + rate int + }{ + // The worst realistic case: 1ms window, 1000-item BatchSize, sparse arrivals. + {"sparse/1ms/batch=1000", 1_000, time.Millisecond, 200}, + {"sparse/5ms/batch=1000", 1_000, 5 * time.Millisecond, 200}, + {"sparse/1ms/batch=10000", 10_000, time.Millisecond, 200}, + // Moderate rate: batches still close on the timer well below BatchSize. + {"steady/5ms/batch=1000", 1_000, 5 * time.Millisecond, 10_000}, + // Control: batches fill, so reserved capacity is actually used. + {"full/100ms/batch=1000", 1_000, 100 * time.Millisecond, 50_000}, + } + + t.Logf("%-26s %-11s %-9s %-12s %s", + "scenario", "mean_batch", "batches", "allocs/item", "bytes/flush (est)") + + for _, c := range cases { + result := scenario.Run(scenario.Config{ + Name: c.name, + BatchSize: c.batchSize, + BatchInterval: c.window, + Arrival: scenario.Steady(c.rate, time.Second), + Processor: scenario.NoOpProcessor(), + LatenessBudget: time.Second, + }) + + // scenario.Item is the harness payload; the aggregator reserves + // BatchSize slots of it per flush regardless of how many arrive. + itemSize := float64(unsafe.Sizeof(scenario.Item{})) + + wastePerFlush := float64(c.batchSize-int(result.MeanBatchSize)) * itemSize + if wastePerFlush < 0 { + wastePerFlush = 0 + } + + t.Logf("%-26s %-11.0f %-9d %-12.3f %.0f B", + c.name, result.MeanBatchSize, result.Batches, + result.AllocsPerItem, wastePerFlush) + } +} From b1c9713ca3544caadaa2c3f5084e718ea575c4b7 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Thu, 6 Aug 2026 16:23:04 -0300 Subject: [PATCH 3/7] docs: document the completed Stats snapshot Documents the three disjoint ownership fields (Queued, BatchHeld, InFlight) and what each one tells an operator, since the useful signal is which of the three is growing rather than any single number. Calls out the two diagnoses the fields now support: a mean batch size well below BatchSize means the interval is costing latency without buying batching, and a rising BatchHeld with InFlight at its ceiling means raise WithConcurrency rather than shrink the window. States the eventual-consistency boundary so callers do not treat a live snapshot as an accounting identity. --- README.md | 29 +++++++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 5102450..c688a51 100644 --- a/README.md +++ b/README.md @@ -291,12 +291,33 @@ and after shutdown it is a counted no-op rather than a panic. ```go s := b.Stats() -// s.Queued -> queue depth; the value to alert on -// s.Pending -> accepted work not yet finished, including in-flight batches -// s.Rejected -> refused enqueues -// s.DroppedErrors -> diagnostics lost because Errors() was not drained +// Where work currently is. These three are disjoint, so together they show +// whether a backlog is waiting on the queue, on batching, or on the processor: +// s.Queued -> published, not yet taken by the aggregator (alert on this) +// s.BatchHeld -> held by the aggregator: filling, or waiting for a free worker +// s.InFlight -> inside a processor call + +// Totals. +// s.Pending -> accepted work not yet finished, including in-flight batches +// s.Accepted -> successful enqueues +// s.Rejected -> refused enqueues +// s.Completed / s.Failed / s.Panicked -> mutually exclusive terminal outcomes +// s.BatchesFlushed -> batches emitted; Completed/BatchesFlushed is mean batch size +// s.DroppedErrors -> diagnostics lost because Errors() was not drained ``` +`BatchesFlushed` is the coalescing signal when tuning `WithBatchInterval`: a mean +batch size well below `BatchSize` means windows are closing on the timer rather than +filling, so the interval is costing latency without buying batching. + +A rising `BatchHeld` with `InFlight` at its ceiling means batches are ready but every +worker is busy — that is the signal to raise `WithConcurrency`, not to shrink the +window. + +The snapshot is eventually consistent, not transactional: each field is read +independently, so use it to observe where work sits, not as an accounting identity +while load is in flight. + ### Handling Errors Whenever the processor function returns an error, the batcher will send the error in the `Errors()` channel: From 6f63b4dcd9812b57ed5248511cd564283e6ec2a4 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Thu, 6 Aug 2026 22:57:13 -0300 Subject: [PATCH 4/7] fix: freeze options after construction and correct Phase 4 claims MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five review findings on Phase 4, one of which was a real data race. Option[T] is a callable function, so a caller could retain one and invoke it after New. The earlier Config()-returns-a-value fix closed the external pointer route but not this one: process() still read b.config.ProcessorFunc live. Reproduced under -race: WARNING: DATA RACE pkg/batcher/batcher.go:492 process reads ProcessorFunc pkg/batcher/options.go:12 post-start WithProcessor writes it Fixed for the whole class rather than just WithProcessor. New freezes configuration after applying options and validating, before any goroutine starts, and every option becomes a no-op afterwards. Configuration was never coherent at runtime — run() already snapshots batch size, interval and concurrency at start — so this makes the construction-time contract explicit instead of leaving one field mutable. options_freeze_test.go hammers five options from four goroutines during active processing and asserts the original config and processor are still in use. Sabotage-verified: removing the guard for WithProcessor makes it report DATA RACE again. The mean batch-size formula was wrong in two places. Completed excludes failed and panicked items, so Completed/BatchesFlushed undercounts whenever the processor errors. Both stats.go and the README now document (Completed + Failed + Panicked) / BatchesFlushed after a terminal drain, and the README states why: a failed batch was still flushed. thresholds.md claimed the +2% allocation ceiling was "Enforced by BenchmarkCapacity* and capacity_test.go". It is not: the benchmarks report allocated bytes without comparing them to the limit, and capacity_test.go asserts estimator behaviour rather than allocation deltas. Relabelled as measured evidence, with the command to re-check and a note that a real gate needs a stored baseline for the reference runner. TestWithProcessor asserted only require.IsType on the configured processor. Every Processor[T] shares that type, including the default no-op, so the assertion passed even if WithProcessor had done nothing. It now invokes the processor and asserts the observable effect. The allocation evidence scenario now rejects a timed-out run before logging batch sizes or allocation counts, with a regression test that deliberately times out a run and asserts TimedOut is reported. Evidence from an unfinished run is not evidence. --- README.md | 7 +- docs/improvements/thresholds.md | 16 +++- pkg/batcher/batcher.go | 12 +++ pkg/batcher/options.go | 36 +++++++++ pkg/batcher/options_freeze_test.go | 90 +++++++++++++++++++++++ pkg/batcher/options_test.go | 10 ++- pkg/batcher/stats.go | 8 +- test/scenario/allocation_evidence_test.go | 23 ++++++ 8 files changed, 195 insertions(+), 7 deletions(-) create mode 100644 pkg/batcher/options_freeze_test.go diff --git a/README.md b/README.md index c688a51..bf2b003 100644 --- a/README.md +++ b/README.md @@ -302,13 +302,16 @@ s := b.Stats() // s.Accepted -> successful enqueues // s.Rejected -> refused enqueues // s.Completed / s.Failed / s.Panicked -> mutually exclusive terminal outcomes -// s.BatchesFlushed -> batches emitted; Completed/BatchesFlushed is mean batch size +// s.BatchesFlushed -> batches emitted. After a terminal drain, +// (Completed+Failed+Panicked)/BatchesFlushed is mean batch size // s.DroppedErrors -> diagnostics lost because Errors() was not drained ``` `BatchesFlushed` is the coalescing signal when tuning `WithBatchInterval`: a mean batch size well below `BatchSize` means windows are closing on the timer rather than -filling, so the interval is costing latency without buying batching. +filling, so the interval is costing latency without buying batching. Include every +terminal outcome in that mean — a failed or panicked batch was still flushed, so +dividing by `Completed` alone undercounts whenever the processor errors. A rising `BatchHeld` with `InFlight` at its ceiling means batches are ready but every worker is busy — that is the signal to raise `WithConcurrency`, not to shrink the diff --git a/docs/improvements/thresholds.md b/docs/improvements/thresholds.md index 0866035..551d207 100644 --- a/docs/improvements/thresholds.md +++ b/docs/improvements/thresholds.md @@ -130,7 +130,21 @@ Results per workload, adaptive versus the previous full-capacity strategy: The alternating case is the one that killed the rejected EWMA estimator, which allocated *more* than doing nothing there. Recent-max keeps it inside the +2% -budget. Enforced by `BenchmarkCapacity*` and `capacity_test.go`. +budget. + +These figures are **measured evidence, not an enforced gate.** `BenchmarkCapacity*` +reports allocated bytes for each workload but does not compare them against the +2% +limit, and `capacity_test.go` asserts estimator behaviour (bounds, adaptation, +decay) rather than allocation deltas. Re-check the numbers with: + +```sh +go test -run='^$' -bench='BenchmarkCapacity' -benchmem -count=1 ./pkg/batcher +``` + +Turning this into a gate needs a stored allocation baseline for the reference +runner, which does not exist yet for the same reason the throughput gate is still +advisory. Until then, a regression here is caught by reading the benchmark output, +not by CI. The second gate must hold for alternating sparse/full, burst-after-idle, and bimodal workloads, not just the sparse case the optimisation targets. An diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go index ee34c42..3178191 100644 --- a/pkg/batcher/batcher.go +++ b/pkg/batcher/batcher.go @@ -6,6 +6,7 @@ import ( "fmt" "runtime/debug" "sync" + "sync/atomic" "time" ) @@ -41,6 +42,12 @@ type Config[T any] struct { type Batcher[T any] struct { config *Config[T] + // configFrozen becomes true before any processing goroutine starts. Options + // are construction-time configuration: applying one to an existing batcher is + // a no-op, so a caller cannot race the aggregator or processor by mutating + // BatchSize, BatchInterval or ProcessorFunc after New returns. + configFrozen atomic.Bool + gate *admissionGate counters counters @@ -93,6 +100,11 @@ func New[T any](options ...Option[T]) *Batcher[T] { b.validateConfig() + // No option may change configuration after this point, including a batcher + // constructed with WithSkipAutoStart. SkipAutoStart controls lifecycle, not + // whether the configuration is still mutable. + b.configFrozen.Store(true) + b.input = newQueue[T](b.config.MaxQueueSize) b.errorsChan = make(chan error, b.config.ErrorBufferSize) diff --git a/pkg/batcher/options.go b/pkg/batcher/options.go index fef5324..671a2da 100644 --- a/pkg/batcher/options.go +++ b/pkg/batcher/options.go @@ -9,6 +9,10 @@ type Option[T any] func(*Batcher[T]) // WithProcessor sets the processor function to be called for each batch. func WithProcessor[T any](fn Processor[T]) Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + b.config.ProcessorFunc = fn } } @@ -16,6 +20,10 @@ func WithProcessor[T any](fn Processor[T]) Option[T] { // WithBatchSize sets the batch size. func WithBatchSize[T any](batchSize int) Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + if batchSize <= 0 { batchSize = 1000 } @@ -27,6 +35,10 @@ func WithBatchSize[T any](batchSize int) Option[T] { // WithBatchInterval sets the batch interval. func WithBatchInterval[T any](batchInterval time.Duration) Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + if batchInterval <= 0 { batchInterval = 1 * time.Second } @@ -38,6 +50,10 @@ func WithBatchInterval[T any](batchInterval time.Duration) Option[T] { // WithSkipAutoStart skips the automatic start of the batcher. func WithSkipAutoStart[T any]() Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + b.config.SkipAutoStart = true } } @@ -59,6 +75,10 @@ func WithSkipAutoStart[T any]() Option[T] { // the peak by up to two batches. func WithMaxQueueSize[T any](maxQueueSize int) Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + if maxQueueSize < 0 { maxQueueSize = 0 } @@ -71,6 +91,10 @@ func WithMaxQueueSize[T any](maxQueueSize int) Option[T] { // reporting ErrTimeout. The drain is never abandoned when the grace expires. func WithCloseGrace[T any](grace time.Duration) Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + if grace <= 0 { grace = DefaultCloseGrace } @@ -92,6 +116,10 @@ func WithCloseGrace[T any](grace time.Duration) Option[T] { // loss. func WithErrorBufferSize[T any](size int) Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + if size <= 0 { size = DefaultErrorBufferSize } @@ -122,6 +150,10 @@ func WithErrorBufferSize[T any](size int) Option[T] { // remaining processor time *plus* a full BatchInterval. func WithConcurrency[T any](concurrency int) Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + if concurrency < 1 { concurrency = 1 } @@ -145,6 +177,10 @@ func WithConcurrency[T any](concurrency int) Option[T] { // - every accepted item is processed exactly once by the pool itself. func WithoutOrderedProcessing[T any]() Option[T] { return func(b *Batcher[T]) { + if b.configFrozen.Load() { + return + } + b.config.UnorderedProcessingAcknowledged = true } } diff --git a/pkg/batcher/options_freeze_test.go b/pkg/batcher/options_freeze_test.go new file mode 100644 index 0000000..6654f30 --- /dev/null +++ b/pkg/batcher/options_freeze_test.go @@ -0,0 +1,90 @@ +package batcher_test + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/NSXBet/batcher/internal/test" + "github.com/NSXBet/batcher/pkg/batcher" + "github.com/stretchr/testify/require" +) + +// TestOptionsAreFrozenAfterNew pins the construction-time option contract. +// +// Option[T] is a callable function, so callers can retain one and invoke it after +// New returns. Before this test, that was a data race: WithProcessor wrote the live +// config while process() read ProcessorFunc, and the other options could similarly +// change batching semantics mid-run. Configuration was never coherent at runtime, +// so every post-New option application is deliberately a no-op. +// +// Run under -race: this fails if a future option bypasses configFrozen and writes a +// live field again. +func TestOptionsAreFrozenAfterNew(t *testing.T) { + t.Parallel() + + var ( + originalCalls atomic.Int64 + mutatedCalls atomic.Int64 + ) + + original := batcher.Processor[test.BatchItem](func([]test.BatchItem) error { + originalCalls.Add(1) + + return nil + }) + + mutated := batcher.Processor[test.BatchItem](func([]test.BatchItem) error { + mutatedCalls.Add(1) + + return nil + }) + + b := batcher.New( + batcher.WithBatchSize[test.BatchItem](1), + batcher.WithBatchInterval[test.BatchItem](time.Millisecond), + batcher.WithProcessor(original), + ) + + defer func() { require.NoError(t, b.Close()) }() + + // Apply every mutable option concurrently with active processing. They must all + // become no-ops after New; the original config remains the one run() snapshots. + var wg sync.WaitGroup + + for range 4 { + wg.Add(1) + + go func() { + defer wg.Done() + + for range 200 { + batcher.WithProcessor(mutated)(b) + batcher.WithBatchSize[test.BatchItem](99)(b) + batcher.WithBatchInterval[test.BatchItem](time.Hour)(b) + batcher.WithConcurrency[test.BatchItem](8)(b) + batcher.WithMaxQueueSize[test.BatchItem](1)(b) + } + }() + } + + for i := range 400 { + b.Add(test.BatchItem{Key: string(rune('a' + i%26))}) + } + + wg.Wait() + + require.NoError(t, b.Join(10*time.Second)) + + config := b.Config() + + require.Equal(t, 1, config.BatchSize) + require.Equal(t, time.Millisecond, config.BatchInterval) + require.Equal(t, 1, config.Concurrency) + require.Equal(t, 0, config.MaxQueueSize) + require.Equal(t, int64(400), originalCalls.Load(), + "processing must keep using the construction-time processor") + require.Zero(t, mutatedCalls.Load(), + "a post-start WithProcessor must not replace the original") +} diff --git a/pkg/batcher/options_test.go b/pkg/batcher/options_test.go index aa3376d..dd11baf 100644 --- a/pkg/batcher/options_test.go +++ b/pkg/batcher/options_test.go @@ -10,7 +10,11 @@ import ( ) func TestWithProcessor(t *testing.T) { + called := false + processor := batcher.Processor[test.BatchItem](func(_ []test.BatchItem) error { + called = true + return nil }) @@ -20,7 +24,11 @@ func TestWithProcessor(t *testing.T) { b := batcher.New(batcher.WithProcessor(processor)) defer func() { _ = b.Close() }() - require.IsType(t, processor, b.Config().ProcessorFunc) + // Invoke the configured processor rather than checking its type. Every + // Processor[T] has the same type, including the default no-op, so a type + // assertion would pass even if WithProcessor had done nothing. + require.NoError(t, b.Config().ProcessorFunc(nil)) + require.True(t, called, "the configured processor must replace the default") } func TestWithBatchSize(t *testing.T) { diff --git a/pkg/batcher/stats.go b/pkg/batcher/stats.go index af68b9c..52eb788 100644 --- a/pkg/batcher/stats.go +++ b/pkg/batcher/stats.go @@ -69,9 +69,11 @@ type Stats struct { Rejected uint64 // BatchesFlushed counts batches emitted by the aggregator, in batches rather - // than items. Together with Completed it gives the mean batch size, which is - // the coalescing figure to watch when tuning BatchInterval: a mean far below - // BatchSize means the window is closing on the timer rather than on size. + // than items. After a terminal drain, (Completed + Failed + Panicked) / + // BatchesFlushed gives the mean batch size. All terminal outcomes are included: + // a processor failure or recovered panic does not make its batch disappear from + // the coalescing figure. A mean far below BatchSize means the window is closing + // on the timer rather than on size. BatchesFlushed uint64 // DroppedErrors counts diagnostics discarded because the Errors() buffer was diff --git a/test/scenario/allocation_evidence_test.go b/test/scenario/allocation_evidence_test.go index a51e9b4..a0ada02 100644 --- a/test/scenario/allocation_evidence_test.go +++ b/test/scenario/allocation_evidence_test.go @@ -6,6 +6,7 @@ import ( "unsafe" "github.com/NSXBet/batcher/test/scenario" + "github.com/stretchr/testify/require" ) // TestSparseWindowAllocationEvidence measures allocated bytes per flush for the @@ -56,6 +57,9 @@ func TestSparseWindowAllocationEvidence(t *testing.T) { LatenessBudget: time.Second, }) + require.False(t, result.TimedOut, + "%s: scenario timed out; allocation evidence is invalid", c.name) + // scenario.Item is the harness payload; the aggregator reserves // BatchSize slots of it per flush regardless of how many arrive. itemSize := float64(unsafe.Sizeof(scenario.Item{})) @@ -70,3 +74,22 @@ func TestSparseWindowAllocationEvidence(t *testing.T) { result.AllocsPerItem, wastePerFlush) } } + +// TestSparseWindowAllocationEvidenceRejectsTimedOutRun pins that evidence is not +// logged when the processor did not finish. A timed-out run has incomplete batches +// and allocation counts, so treating it as capacity evidence would be misleading. +func TestSparseWindowAllocationEvidenceRejectsTimedOutRun(t *testing.T) { + t.Parallel() + + result := scenario.Run(scenario.Config{ + Name: "timed-out-evidence", + BatchSize: 1, + BatchInterval: time.Millisecond, + Arrival: scenario.Steady(1_000, 100*time.Millisecond), + Processor: scenario.FixedProcessor(100 * time.Millisecond), + CompletionDeadline: time.Millisecond, + LatenessBudget: time.Second, + }) + + require.True(t, result.TimedOut, "the deliberately blocked run must time out") +} From bb1eae99fda82334f4778ddcd39d0f2a19889305 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Fri, 7 Aug 2026 10:37:26 -0300 Subject: [PATCH 5/7] fix: read processing config only from an immutable snapshot The Phase 4 review found the data-race fix incomplete, and it was right: run() snapshotted batch size, interval and worker count but process() still dereferenced b.config.ProcessorFunc on every batch. One live field is enough to race, so the rest of that snapshot bought nothing, and run()'s own comment described a discipline the code did not follow. Reproduced by writing the Config field directly while workers ran: WARNING: DATA RACE pkg/batcher/batcher.go:330 Rather than pass a local into process and hope future reads remember, the fields the pipeline needs are now copied once into a runtimeConfig, in New, before any goroutine exists. Workers receive the processor as a parameter. No processing path reads Config at all, so the class is closed rather than narrowed -- and normalisation (batch size, worker count, nil processor) happens in one place instead of being repeated at start. TestProcessingReadsOnlyTheRuntimeSnapshot pins it by mutating Config directly, deliberately bypassing the frozen-option guard, since that guard is a separate defence and this test exists to prove the snapshot holds without it. Sabotage-verified: restoring the live read makes it fail under -race. Also documented the freeze on the Option type itself. WithBatchSize(99) applied after New silently did nothing -- no error, no panic -- so a caller could believe batch size was 99 while it was 1000. The type doc now states options are construction-time only and explains why runtime reconfiguration is not offered; WithSkipAutoStart says explicitly that it delays start without leaving configuration mutable. --- pkg/batcher/batcher.go | 68 +++++++++++++++------- pkg/batcher/options.go | 16 +++++- pkg/batcher/runtime_snapshot_test.go | 84 ++++++++++++++++++++++++++++ 3 files changed, 147 insertions(+), 21 deletions(-) create mode 100644 pkg/batcher/runtime_snapshot_test.go diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go index 3178191..6358f3a 100644 --- a/pkg/batcher/batcher.go +++ b/pkg/batcher/batcher.go @@ -48,6 +48,11 @@ type Batcher[T any] struct { // BatchSize, BatchInterval or ProcessorFunc after New returns. configFrozen atomic.Bool + // runtime is the immutable snapshot the processing goroutines use. It is taken + // in New, before any goroutine exists, so no worker ever reads the Config + // struct a caller may still hold a reference to. + runtime runtimeConfig[T] + gate *admissionGate counters counters @@ -76,6 +81,19 @@ type Batcher[T any] struct { shutdownDone chan struct{} } +// runtimeConfig is the frozen view of Config that processing goroutines read. +// +// Config is a plain struct and callers can hold a reference to one, so reading it +// from a worker is a data race by construction. Copying the fields the pipeline +// needs, once, before any goroutine starts, removes that whole class rather than +// relying on every read site to remember to use a local. +type runtimeConfig[T any] struct { + batchSize int + batchInterval time.Duration + workers int + processor Processor[T] +} + // New creates a new Batcher with the given options. func New[T any](options ...Option[T]) *Batcher[T] { b := &Batcher[T]{ @@ -105,6 +123,25 @@ func New[T any](options ...Option[T]) *Batcher[T] { // whether the configuration is still mutable. b.configFrozen.Store(true) + b.runtime = runtimeConfig[T]{ + batchSize: b.config.BatchSize, + batchInterval: b.config.BatchInterval, + workers: b.config.Concurrency, + processor: b.config.ProcessorFunc, + } + + if b.runtime.batchSize < 1 { + b.runtime.batchSize = DefaultBatchSize + } + + if b.runtime.workers < 1 { + b.runtime.workers = 1 + } + + if b.runtime.processor == nil { + b.runtime.processor = NoOpProcessor[T] + } + b.input = newQueue[T](b.config.MaxQueueSize) b.errorsChan = make(chan error, b.config.ErrorBufferSize) @@ -316,26 +353,17 @@ func (b *Batcher[T]) Errors() <-chan error { func (b *Batcher[T]) run() { defer close(b.stopped) - // Snapshot the configuration once, at start, and use only the snapshot from here - // on. Config is a plain struct the caller still holds a pointer to, so reading it - // per batch races with any option applied after Start. Freezing it also matches - // the semantics callers already rely on: batch size and interval are fixed for - // the lifetime of a running batcher, so re-reading them could never have taken - // effect coherently mid-batch anyway. + // Use the immutable runtime snapshot taken by New before any goroutine starts. + // In particular, ProcessorFunc is passed to workers rather than dereferenced + // from Config per batch. This keeps every processing read off the mutable Config + // struct, so a caller with a stale reference cannot race the pipeline. var ( - batchSize = b.config.BatchSize - batchInterval = b.config.BatchInterval - workers = b.config.Concurrency + batchSize = b.runtime.batchSize + batchInterval = b.runtime.batchInterval + workers = b.runtime.workers + processor = b.runtime.processor ) - if batchSize < 1 { - batchSize = DefaultBatchSize - } - - if workers < 1 { - workers = 1 - } - batches := make(chan []T) var processing sync.WaitGroup @@ -347,7 +375,7 @@ func (b *Batcher[T]) run() { defer processing.Done() for items := range batches { - b.process(items) + b.process(processor, items) } }() } @@ -478,7 +506,7 @@ func (b *Batcher[T]) run() { // The ordering is fixed and load-bearing: whatever happens, exactly one terminal // category is counted and the drain obligation is released exactly once. Leaving it // unreleased would inflate Pending permanently and hang shutdown forever. -func (b *Batcher[T]) process(items []T) { +func (b *Batcher[T]) process(processor Processor[T], items []T) { b.counters.inFlight.Add(int64(len(items))) outcome := outcomeCompleted @@ -503,7 +531,7 @@ func (b *Batcher[T]) process(items []T) { }) }() - if err := b.config.ProcessorFunc(items); err != nil { + if err := processor(items); err != nil { outcome = outcomeFailed b.publishError(err) diff --git a/pkg/batcher/options.go b/pkg/batcher/options.go index 671a2da..df975ff 100644 --- a/pkg/batcher/options.go +++ b/pkg/batcher/options.go @@ -4,6 +4,16 @@ import ( "time" ) +// Option configures a Batcher during New. +// +// Options are construction-time only. New applies them, validates the resulting +// configuration, and freezes it before any processing goroutine can start. Calling +// an Option on a Batcher after New returns is a deliberate no-op: runtime +// reconfiguration was never coherent (the pipeline snapshots its configuration at +// start), and allowing it would reintroduce races against workers. +// +// To change configuration, construct a new Batcher. In particular, WithSkipAutoStart +// delays lifecycle start; it does not leave the configuration mutable until Start. type Option[T any] func(*Batcher[T]) // WithProcessor sets the processor function to be called for each batch. @@ -47,7 +57,11 @@ func WithBatchInterval[T any](batchInterval time.Duration) Option[T] { } } -// WithSkipAutoStart skips the automatic start of the batcher. +// WithSkipAutoStart skips automatic lifecycle start. +// +// It does not defer configuration freeze: options are still applied only during New. +// Call Start when ready to process, or Shutdown to drain queued work without an +// explicit Start. func WithSkipAutoStart[T any]() Option[T] { return func(b *Batcher[T]) { if b.configFrozen.Load() { diff --git a/pkg/batcher/runtime_snapshot_test.go b/pkg/batcher/runtime_snapshot_test.go new file mode 100644 index 0000000..83fb5c9 --- /dev/null +++ b/pkg/batcher/runtime_snapshot_test.go @@ -0,0 +1,84 @@ +package batcher + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestProcessingReadsOnlyTheRuntimeSnapshot pins that processing goroutines never +// read the mutable Config struct. +// +// Config is a plain struct and a caller can hold a reference to one, so any read of +// it from a worker is a data race by construction. run() previously snapshotted +// batch size, interval and worker count but still dereferenced +// b.config.ProcessorFunc on every batch, which made the rest of that snapshot +// pointless: one live field is enough to race. +// +// This test writes the Config field directly, bypassing the frozen-option guard +// entirely, because the guard is a separate defence. Under -race it fails if any +// processing path reads Config again. +func TestProcessingReadsOnlyTheRuntimeSnapshot(t *testing.T) { + t.Parallel() + + var processed int + + b := New( + WithBatchSize[int](1), + WithBatchInterval[int](time.Millisecond), + WithProcessor(func(items []int) error { + processed += len(items) + + return nil + }), + ) + + t.Cleanup(func() { _ = b.Close() }) + + done := make(chan struct{}) + + go func() { + defer close(done) + + for range 300 { + b.Add(1) + } + }() + + // A hostile or stale caller mutating configuration while work is in flight. + for range 300 { + b.config.ProcessorFunc = func([]int) error { return nil } + b.config.BatchSize = 999 + b.config.BatchInterval = time.Hour + } + + <-done + + require.NoError(t, b.Join(10*time.Second)) + + // The pipeline must still be running the configuration it started with. + require.Equal(t, 1, b.runtime.batchSize) + require.Equal(t, time.Millisecond, b.runtime.batchInterval) + require.Equal(t, 1, b.runtime.workers) + require.NotNil(t, b.runtime.processor) + require.Positive(t, processed, + "the original processor must have run; a swapped-in one would not count here") +} + +// TestRuntimeSnapshotNormalisesInvalidConfig pins the defaults applied when the +// snapshot is taken, so a zero or negative value cannot produce a pipeline with no +// capacity or no processor. +func TestRuntimeSnapshotNormalisesInvalidConfig(t *testing.T) { + t.Parallel() + + b := New[int](WithSkipAutoStart[int]()) + + t.Cleanup(func() { _ = b.Close() }) + + // Defaults, not zero values. + require.Positive(t, b.runtime.batchSize) + require.GreaterOrEqual(t, b.runtime.workers, 1) + require.NotNil(t, b.runtime.processor, + "a nil processor must normalise to the no-op rather than panicking a worker") +} From 3f48bcb5865308ac2bfc140bc50e00baa4276054 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Fri, 7 Aug 2026 12:38:06 -0300 Subject: [PATCH 6/7] perf: drain the intake queue in batches instead of per item Profiling the queue in its real shape -- multiple producers, one greedy-draining aggregator -- showed the consumer, not the lock itself, was the cost. With 8 producers moving 200k items, 86% of all mutex delay and 61% of CPU samples landed on push, contending with a consumer that acquired the same mutex once per item because drainReady called pop in a loop. The aggregator is the only consumer and holds no invariant between items: it appends each one to the batch it is building. So a run can move under a single acquisition, which is what popBatch does. Measured on the same workload (darwin/arm64, Apple M4 Pro, Go 1.26.5, 200k items): 5.4x faster with one producer, 1.1x with 256, and unbounded-mode allocation down from ~8MiB to ~0.2MiB per run. Consumer-side mutex delay fell from ~86% to 8.8%; the remainder is producer-to-producer contention in push, which is inherent to a single FIFO. Two alternatives were measured rather than assumed, since a mutex here looks like a smell: - xsync/v4 UMPSCQueue, a lock-free unbounded MPSC queue, measured within noise of this change across 1-256 producers. It also blocks in Dequeue with no non-blocking or readiness variant, so the aggregator could not select across queue readiness, the flush timer and shutdown. No reason to take the dependency. - Sharding the push side across 8 stripes, which abandons FIFO, was not faster at high producer counts (16.4ms versus 14.4ms at 256 producers). Giving up ordering buys nothing here. For the record, ahrav/go-lockfree-queue measured 5-24x slower than this queue, and its constructor starts a reclamation goroutine with no stop path. bruceshao/lockfree does not build on Go 1.26 at all -- its //go:linkname references to runtime.osyield and runtime.procyield are rejected by the linker, and its own test suite fails the same way. The transfer is capped by the space remaining in the batch being built, not by batchSize. Draining a full batchSize regardless of that space parked an extra batch-sized buffer of accepted work outside both the queue and the batch, which pushed Pending past the documented N + 2*BatchSize + gate bound and failed TestAcceptedWorkBoundIncludesHeldAndInFlightBatches at 12 against 10. This lands on Phase 4 rather than Phase 2, where the queue was introduced. The drain needs a stable transfer bound, and on Phase 2 configuration is still mutable after New, so reading BatchSize in the drain loop is a data race that the per-item path never had -- TestWithBatchSize catches it. Phase 4 freezes configuration and snapshots it, so the bound is available without a new read of Config. Five popBatch tests cover FIFO across compaction, the transfer cap, buffer reuse at zero allocations, retained-storage reclamation, and the notFull signal. All are sabotage-verified. The notFull test asserts the latch directly after an earlier version passed with the signal deleted: a parked publisher can be rescheduled into free space regardless, so racing a goroutine against the drain tested scheduling rather than the wakeup. queue_bench_test.go adds the MPSC benchmark the profiles came from, with the pprof invocation in its doc comment, so the next person can reproduce the attribution instead of trusting this message. The third-party comparisons stayed in a scratch module; no benchmark-only dependency enters go.mod. --- pkg/batcher/batcher.go | 36 ++++++- pkg/batcher/queue.go | 121 ++++++++++++++++++---- pkg/batcher/queue_bench_test.go | 66 ++++++++++++ pkg/batcher/queue_test.go | 174 ++++++++++++++++++++++++++++++++ 4 files changed, 374 insertions(+), 23 deletions(-) create mode 100644 pkg/batcher/queue_bench_test.go diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go index 6358f3a..b32e11e 100644 --- a/pkg/batcher/batcher.go +++ b/pkg/batcher/batcher.go @@ -458,15 +458,43 @@ func (b *Batcher[T]) run() { // drainReady empties whatever is currently queued. The aggregator drains // greedily rather than one item per wakeup, so a burst costs one signal. + // + // It transfers a run of items under a single queue lock rather than locking per + // item. The aggregator is the only consumer and keeps no invariant between items, + // so a block transfer is indistinguishable from N pops to it, while per-item + // locking was what serialised producers: profiling attributed 86% of mutex delay + // and 61% of CPU to the push path contending with this loop. + // + // Each transfer takes only what the batch being built can still accept. That + // bounds how long the queue lock is held, and it keeps the capacity contract + // exact: transferred items move straight into batch, so no additional + // batch-sized buffer of accepted work exists outside the queue. Draining a full + // batchSize regardless of the batch's remaining space added exactly that third + // buffer and pushed accepted work past N + 2*BatchSize + gate. + // + // drained is reused across wakeups to keep steady-state draining allocation-free. + // It is deliberately separate from batch: batch is handed to a worker and may be + // retained by the processor, whereas drained never leaves this goroutine. + drained := make([]T, 0, batchSize) + drainReady := func() { for { - item, ok := b.input.pop() - if !ok { + room := batchSize - len(batch) + if room <= 0 { + // take flushes at batchSize, so this only happens if a flush could not + // complete. Fall back to one item and let take drive the flush. + room = 1 + } + + drained = b.input.popBatch(drained[:0], room) + if len(drained) == 0 { return } - b.counters.received(1) - take(item) + for _, item := range drained { + b.counters.received(1) + take(item) + } } } diff --git a/pkg/batcher/queue.go b/pkg/batcher/queue.go index 209ff9c..46a7e5e 100644 --- a/pkg/batcher/queue.go +++ b/pkg/batcher/queue.go @@ -109,7 +109,108 @@ func (q *queue[T]) tryPush(item T) bool { return true } +// popBatch moves up to max queued items onto dst and returns the extended slice. +// +// This exists because the aggregator is a single consumer that drains greedily: it +// pops until the queue is empty on every wakeup. Doing that one item at a time takes +// and releases the mutex once per item, and under concurrent producers those +// acquisitions are what serialise the queue. Profiling an 8-producer, 200k-item MPSC +// workload attributed 86% of all mutex delay to the push path waiting behind exactly +// this traffic, and 61% of CPU samples to push. +// +// Transferring a run under one acquisition is safe precisely because the consumer is +// single and holds no invariant between items: it appends them to the batch it is +// building, so one block transfer and N single pops are indistinguishable to it. +// +// max bounds the transfer so a large backlog cannot be moved in one acquisition, +// which would hold the lock for an unbounded time and starve producers. Passing +// dst[:0] reuses the caller's buffer, so steady-state draining does not allocate. +// +// Measured against per-item pop on the same workload (darwin/arm64, Apple M4 Pro, +// Go 1.26.5, 200k items): 5.4x faster with one producer, 1.1x with 256, and +// unbounded-mode allocation down from ~8MiB to ~0.2MiB per run. For reference, a +// lock-free MPSC queue (xsync/v4 UMPSCQueue) measured within noise of this on the +// same workload, so the mutex was never the cost -- per-item locking was. +func (q *queue[T]) popBatch(dst []T, max int) []T { + if max <= 0 { + return dst + } + + q.mu.Lock() + + queued := len(q.items) - q.head + if queued == 0 { + // Fully drained: reset the cursor and keep the backing array, so the next + // batch of pushes reuses capacity instead of allocating. + if q.head > 0 { + q.items = q.items[:0] + q.head = 0 + } + + q.mu.Unlock() + + return dst + } + + n := min(queued, max) + + dst = append(dst, q.items[q.head:q.head+n]...) + + // Clear the transferred slots so popped items are not kept alive by the backing + // array. + clear(q.items[q.head : q.head+n]) + + q.head += n + + remaining := len(q.items) - q.head + + q.reclaimLocked(remaining) + + q.mu.Unlock() + + if remaining > 0 { + // Keep the latch armed: the aggregator drains greedily, but re-arming means + // a missed wakeup cannot strand items. + signal(q.notEmpty) + } + + signal(q.notFull) + + return dst +} + +// reclaimLocked keeps the backing array proportional to queue depth rather than to +// total throughput. The caller must hold q.mu. +// +// Without this, push always appends and the full-drain reset only fires when the +// queue is observed completely empty. A steady producer that keeps even one item +// resident never triggers it, so the array grows without bound. Measured before this +// compaction: a queue holding a single item reached 219,136 slots after 200,000 +// pushes. +// +// Compacting when head >= remaining keeps the copy cost amortised O(1) per item, +// because each compaction halves the live region's offset and moves at most as many +// items as have been consumed since the last one. +func (q *queue[T]) reclaimLocked(remaining int) { + if remaining == 0 { + q.items = q.items[:0] + q.head = 0 + + return + } + + if q.head >= remaining { + copy(q.items, q.items[q.head:]) + clear(q.items[remaining:]) + + q.items = q.items[:remaining] + q.head = 0 + } +} + // pop removes the oldest item, reporting false when the queue is empty. +// +// popBatch is the aggregator's path; this remains for single-item callers and tests. func (q *queue[T]) pop() (T, bool) { var zero T @@ -135,25 +236,7 @@ func (q *queue[T]) pop() (T, bool) { remaining := len(q.items) - q.head - // Reclaim the consumed prefix once it dominates the slice. - // - // Without this, the backing array grows with total throughput rather than with - // queue depth: push always appends, and the full-drain reset above only fires - // when the queue is observed completely empty. A steady producer that keeps even - // one item resident never triggers it, so the array grows without bound. - // Measured before this compaction: a queue holding a single item reached 219,136 - // slots after 200,000 pushes. - // - // Compacting when head >= remaining keeps the copy cost amortised O(1) per item, - // because each compaction halves the live region's offset and moves at most as - // many items as have been consumed since the last one. - if q.head >= remaining { - copy(q.items, q.items[q.head:]) - clear(q.items[remaining:]) - - q.items = q.items[:remaining] - q.head = 0 - } + q.reclaimLocked(remaining) q.mu.Unlock() diff --git a/pkg/batcher/queue_bench_test.go b/pkg/batcher/queue_bench_test.go new file mode 100644 index 0000000..37204e8 --- /dev/null +++ b/pkg/batcher/queue_bench_test.go @@ -0,0 +1,66 @@ +package batcher + +import ( + "context" + "sync/atomic" + "testing" +) + +// BenchmarkQueueBatchDrainMPSC measures Batcher's queue in its real access shape: +// multiple publishers and exactly one greedy-draining aggregator. +// +// It deliberately exercises the queue directly instead of BenchmarkBatcherEnqueue: +// the latter measures the public producer API (admission, lifecycle counters and the +// processor) and is the right compatibility benchmark, but it cannot attribute a +// CPU or mutex profile to the intake queue itself. +// +// Capture profiles with: +// +// go test -run '^$' -bench '^BenchmarkQueueBatchDrainMPSC$' -benchtime=5s \ +// -cpuprofile cpu.out -memprofile mem.out -mutexprofile mutex.out ./pkg/batcher +// +// Inspect with `go tool pprof -http=:0 cpu.out` (and the other profiles). In +// particular, compare mutex delay in queue.push against earlier profiles before +// reintroducing any per-item locking or replacing the queue with an atomic structure. +func BenchmarkQueueBatchDrainMPSC(b *testing.B) { + q := newQueue[int](0) + sealCh := make(chan struct{}) + ctx := context.Background() + + var received atomic.Int64 + + done := make(chan struct{}) + + go func() { + defer close(done) + + buf := make([]int, 0, DefaultBatchSize) + + for received.Load() < int64(b.N) { + <-q.ready() + + for { + buf = q.popBatch(buf[:0], DefaultBatchSize) + if len(buf) == 0 { + break + } + + received.Add(int64(len(buf))) + } + } + }() + + b.ReportAllocs() + b.ResetTimer() + + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + if err := q.push(ctx, 1, sealCh); err != nil { + b.Fatal(err) + } + } + }) + + b.StopTimer() + <-done +} diff --git a/pkg/batcher/queue_test.go b/pkg/batcher/queue_test.go index 8b1817d..ef0c5c5 100644 --- a/pkg/batcher/queue_test.go +++ b/pkg/batcher/queue_test.go @@ -3,6 +3,7 @@ package batcher import ( "context" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -74,3 +75,176 @@ func TestQueueCompactionPreservesOrder(t *testing.T) { _, ok := q.pop() require.False(t, ok) } + +// TestPopBatchPreservesFIFOAcrossCompaction is the equivalence check that matters: +// batch draining must be indistinguishable from repeated pop, including across the +// prefix compaction that moves the live region. +func TestPopBatchPreservesFIFOAcrossCompaction(t *testing.T) { + t.Parallel() + + q := newQueue[int](0) + sealCh := make(chan struct{}) + + const total = 5_000 + + var ( + got []int + next int + buf []int + ) + + // Interleave pushes and partial batch drains so head advances, the suffix is + // compacted, and drains land mid-array rather than always at offset zero. + for next < total { + for range 7 { + if next >= total { + break + } + + require.NoError(t, q.push(context.Background(), next, sealCh)) + + next++ + } + + buf = q.popBatch(buf[:0], 3) + got = append(got, buf...) + } + + for { + buf = q.popBatch(buf[:0], 64) + if len(buf) == 0 { + break + } + + got = append(got, buf...) + } + + require.Len(t, got, total) + + for i, v := range got { + require.Equal(t, i, v, "batch draining must preserve FIFO order") + } +} + +// TestPopBatchRespectsMaxAndReportsEmpty pins the transfer cap, which is what bounds +// how long the queue lock is held. Without it a large backlog would move in one +// acquisition and starve producers for an unbounded time. +func TestPopBatchRespectsMaxAndReportsEmpty(t *testing.T) { + t.Parallel() + + q := newQueue[int](0) + sealCh := make(chan struct{}) + + for i := range 10 { + require.NoError(t, q.push(context.Background(), i, sealCh)) + } + + got := q.popBatch(nil, 4) + require.Equal(t, []int{0, 1, 2, 3}, got, "at most max items may be transferred") + + got = q.popBatch(got[:0], 100) + require.Equal(t, []int{4, 5, 6, 7, 8, 9}, got, + "a max above the queued count must transfer only what is queued") + + require.Empty(t, q.popBatch(got[:0], 8), "an empty queue must transfer nothing") + + // A non-positive max is a caller bug; returning dst unchanged keeps it from + // silently draining the queue or panicking. + require.NoError(t, q.push(context.Background(), 42, sealCh)) + require.Empty(t, q.popBatch(nil, 0)) + require.Equal(t, 1, q.length(), "max<=0 must not consume items") +} + +// TestPopBatchReusesCallerBuffer pins the allocation property the aggregator relies +// on: steady-state draining into a reused buffer must not allocate. +func TestPopBatchReusesCallerBuffer(t *testing.T) { + // AllocsPerRun temporarily changes process-wide GC settings, so Go rejects it + // from a parallel test. + q := newQueue[int](0) + sealCh := make(chan struct{}) + + buf := make([]int, 0, 64) + + // Warm up so slice growth and the queue's own backing array settle. + for range 64 { + require.NoError(t, q.push(context.Background(), 1, sealCh)) + } + + buf = q.popBatch(buf[:0], 64) + require.Len(t, buf, 64) + + allocs := testing.AllocsPerRun(200, func() { + for range 32 { + _ = q.push(context.Background(), 1, sealCh) + } + + buf = q.popBatch(buf[:0], 32) + }) + + require.Zero(t, allocs, + "draining into a reused buffer must not allocate; got %.1f allocs/op", allocs) +} + +// TestPopBatchDrainedQueueReclaimsStorage pins that batch draining keeps the same +// memory bound as pop: backing storage proportional to depth, not to throughput. +func TestPopBatchDrainedQueueReclaimsStorage(t *testing.T) { + t.Parallel() + + q := newQueue[int](0) + sealCh := make(chan struct{}) + + var buf []int + + require.NoError(t, q.push(context.Background(), 0, sealCh)) + + const cycles = 50_000 + + for i := range cycles { + require.NoError(t, q.push(context.Background(), i, sealCh)) + + buf = q.popBatch(buf[:0], 1) + require.Len(t, buf, 1, "cycle %d: queue unexpectedly empty", i) + } + + q.mu.Lock() + defer q.mu.Unlock() + + require.Equal(t, 1, len(q.items)-q.head, "one item must remain live") + require.LessOrEqual(t, cap(q.items), 8, + "backing capacity must stay proportional to depth; got cap=%d after %d cycles", + cap(q.items), cycles) +} + +// TestPopBatchSignalsBoundedProducers pins the actual release mechanism: a batch +// transfer from a bounded queue arms notFull. push selects on this latch while full, +// so it is the signal that wakes parked publishers. Testing the channel directly is +// stronger than racing a goroutine against the drain: a publisher can be scheduled +// to retry after space exists even if the signal is accidentally removed. +func TestPopBatchSignalsBoundedProducers(t *testing.T) { + t.Parallel() + + q := newQueue[int](4) + sealCh := make(chan struct{}) + + for i := range 4 { + require.NoError(t, q.push(context.Background(), i, sealCh)) + } + + require.False(t, q.tryPush(99), "queue must be full") + + // Ensure the signal observed below comes from this popBatch, not an earlier pop. + select { + case <-q.notFull: + default: + } + + buf := q.popBatch(nil, 4) + require.Len(t, buf, 4) + + select { + case <-q.notFull: + // The one-place latch is armed. A publisher parked in push can now retry. + case <-time.After(time.Second): + t.Fatal("popBatch did not arm notFull; bounded publishers would remain parked") + } +} From ef36391aa3c0fe06e67a652245fa7b19e55a2f8d Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Fri, 7 Aug 2026 21:45:14 -0300 Subject: [PATCH 7/7] docs: correct Pending semantics and the 4.2 allocation baseline Two review findings on Phase 4. The README described Stats().Pending as "accepted work not yet finished", which is narrower than the field is. Pending counts publishers that have reserved but not yet published, so it can exceed accepted work while load is in flight. stats.go already documents it as a conservative drain obligation; the README now matches, and states that it equals accepted-but-unfinished work only once PublishersInGate == 0. TestSparseWindowAllocationEvidence carried two claims that this milestone falsified. It said the aggregator allocates make([]T, 0, BatchSize) per batch, which 4.2 changed to capacities.capacity(), and it said the test never fails, while it fails on result.TimedOut. The wastePerFlush formula is deliberately unchanged. Recomputing it from the adaptive estimate would destroy the comparison the milestone is judged by, so the column is now labelled "bytes/flush (pre-4.2 est)" and documented as the pre-adaptive baseline: the waste a fixed reservation would have incurred, which is what adaptive capacity removed. The TimedOut assertion is described as a validity check rather than an allocation gate, since a timed-out run is not evidence of anything. --- README.md | 5 ++++- test/scenario/allocation_evidence_test.go | 24 +++++++++++++++-------- 2 files changed, 20 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index bf2b003..60722e1 100644 --- a/README.md +++ b/README.md @@ -298,7 +298,10 @@ s := b.Stats() // s.InFlight -> inside a processor call // Totals. -// s.Pending -> accepted work not yet finished, including in-flight batches +// s.Pending -> conservative drain obligation: accepted-or-reserved work, +// including in-flight batches. It counts publishers that have +// reserved but not yet published, so it equals +// accepted-but-unfinished work only once PublishersInGate == 0 // s.Accepted -> successful enqueues // s.Rejected -> refused enqueues // s.Completed / s.Failed / s.Panicked -> mutually exclusive terminal outcomes diff --git a/test/scenario/allocation_evidence_test.go b/test/scenario/allocation_evidence_test.go index a0ada02..f5d33be 100644 --- a/test/scenario/allocation_evidence_test.go +++ b/test/scenario/allocation_evidence_test.go @@ -15,12 +15,19 @@ import ( // capacity for. // // This exists to decide whether 4.2 should be implemented at all. The plan gates it -// on sparse-window allocation pressure above 2 KB/flush; the aggregator allocates -// make([]T, 0, BatchSize) per batch, so the theoretical waste is -// (BatchSize - actual) * sizeof(T) per flush. +// on sparse-window allocation pressure above 2 KB/flush. // -// Report-only: it prints evidence and never fails, because it is an input to a -// decision rather than a gate. +// The "bytes/flush (est)" column is the PRE-ADAPTIVE BASELINE, deliberately kept +// after 4.2 landed: it is (BatchSize - actual) * sizeof(T), the waste a fixed +// make([]T, 0, BatchSize) reservation would incur. The aggregator now reserves +// capacities.capacity() instead, so this column reports the waste adaptive capacity +// removed rather than waste that remains. It is the figure the milestone is measured +// against, which is why it is not recomputed from the current reservation. +// +// The measurements are report-only: no allocation threshold is asserted, because the +// numbers are an input to a decision rather than a gate. The scenario's own validity +// is still enforced -- a run that timed out is not evidence of anything, so +// result.TimedOut fails the test. func TestSparseWindowAllocationEvidence(t *testing.T) { t.Parallel() @@ -45,7 +52,7 @@ func TestSparseWindowAllocationEvidence(t *testing.T) { } t.Logf("%-26s %-11s %-9s %-12s %s", - "scenario", "mean_batch", "batches", "allocs/item", "bytes/flush (est)") + "scenario", "mean_batch", "batches", "allocs/item", "bytes/flush (pre-4.2 est)") for _, c := range cases { result := scenario.Run(scenario.Config{ @@ -60,8 +67,9 @@ func TestSparseWindowAllocationEvidence(t *testing.T) { require.False(t, result.TimedOut, "%s: scenario timed out; allocation evidence is invalid", c.name) - // scenario.Item is the harness payload; the aggregator reserves - // BatchSize slots of it per flush regardless of how many arrive. + // scenario.Item is the harness payload. BatchSize, not the adaptive estimate, + // is deliberately used here: this reproduces what a fixed reservation would + // have cost, which is the baseline 4.2 is compared against. itemSize := float64(unsafe.Sizeof(scenario.Item{})) wastePerFlush := float64(c.batchSize-int(result.MeanBatchSize)) * itemSize