perf: make batching observable and allocation-efficient (Phase 4) - #32
Merged
Merged
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughThe batcher now freezes runtime configuration, returns configuration snapshots, adapts batch capacity, drains queues in bounded batches, preserves retained batch slices, and reports expanded work-location and terminal statistics. ChangesBatcher runtime and performance
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Publisher
participant Queue
participant Aggregator
participant Worker
participant Stats
Publisher->>Queue: push accepted item
Aggregator->>Queue: popBatch items
Aggregator->>Stats: record BatchHeld
Aggregator->>Worker: dispatch batch
Aggregator->>Stats: increment BatchesFlushed
Worker-->>Stats: record terminal outcome
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
heynemann
marked this pull request as ready for review
August 6, 2026 20:02
heynemann
force-pushed
the
phase-4-observability-allocation
branch
from
August 6, 2026 23:25
e874040 to
287855a
Compare
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/improvements/thresholds.md`:
- Around line 131-133: Correct the statement about +2% enforcement in the
alternating-case discussion: either describe BenchmarkCapacity* and
capacity_test.go only as measured evidence, or add a CI check that compares a
stored allocation baseline against the +2% threshold before claiming
enforcement.
In `@pkg/batcher/batcher.go`:
- Around line 305-315: Prevent post-start option calls from mutating the
configuration used by a running Batcher. Update the option application path in
New and the Batcher startup state so Option[T] becomes a no-op after Start,
while preserving all pre-start configuration behavior; ensure process and the
startup snapshot continue using stable configuration. Add a race-enabled
regression test covering WithProcessor invoked after Start and verify it cannot
alter processing.
In `@pkg/batcher/options_test.go`:
- Around line 13-23: Update the test around batcher.New and
b.Config().ProcessorFunc to invoke the configured processor with a test batch
and assert its observable behavior, rather than only checking its function type.
Preserve the existing WithProcessor setup and verify the result demonstrates
that the supplied processor replaced the default.
In `@pkg/batcher/stats.go`:
- Around line 71-74: The mean batch-size documentation is incorrect because
Completed excludes failed and panicked items. Update pkg/batcher/stats.go lines
71-74 and README.md lines 293-304 to document (Completed + Failed + Panicked) /
BatchesFlushed, explicitly stating that this formula applies to a terminally
drained snapshot.
In `@test/scenario/allocation_evidence_test.go`:
- Around line 49-70: In the allocation evidence loop, validate result.TimedOut
immediately after scenario.Run returns and fail the case when it is true.
Perform this assertion before calculating wastePerFlush or logging
MeanBatchSize, Batches, or AllocsPerItem, while preserving the existing
calculations for non-timed-out results.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: e5e1ba24-c9ea-4ae6-9187-c307aebffda5
📒 Files selected for processing (12)
README.mddocs/improvements/plan-perf.mddocs/improvements/thresholds.mdpkg/batcher/batcher.gopkg/batcher/capacity.gopkg/batcher/capacity_bench_test.gopkg/batcher/capacity_test.gopkg/batcher/options_test.gopkg/batcher/retention_test.gopkg/batcher/stats.gopkg/batcher/stats_test.gotest/scenario/allocation_evidence_test.go
heynemann
force-pushed
the
phase-4-observability-allocation
branch
2 times, most recently
from
August 7, 2026 13:53
e0e2e88 to
f04340a
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
docs/improvements/plan-perf.md (2)
953-956: 🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick winReconcile the atomic-operation budget.
The protocol requires five successful-path atomic RMWs, while this section reports four contended atomics and does not identify a different implementation or measurement scope. State whether the measurement excludes
gate++andgate--; otherwise, correct one count.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docs/improvements/plan-perf.md` around lines 953 - 956, Reconcile the atomic-count statement in the counter-placement section with the protocol’s five successful-path atomic RMWs. Explicitly state whether the benchmark excludes gate++ and gate--; if not, update the reported contended-atomic count and performance comparison to use the correct total.
1032-1035: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winDefine adaptive capacity for small batch sizes.
WithBatchSizeaccepts positive values below 16, so[16, BatchSize]is invalid for supported configurations. Define the lower bound asmin(16, BatchSize), or rejectBatchSize < 16during validation. Add tests for small batch sizes such as 1 and 7.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docs/improvements/plan-perf.md` around lines 1032 - 1035, Define the estimator’s lower clamp bound as min(16, BatchSize) so configurations accepted by WithBatchSize, including batch sizes 1 and 7, remain valid; alternatively reject values below 16 during WithBatchSize validation. Update the adaptive-capacity documentation and add tests covering these small batch sizes.
🧹 Nitpick comments (6)
docs/improvements/plan-perf.md (1)
164-164: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd language identifiers to fenced blocks.
markdownlintreports MD040 for these seven fences. Usetextfor protocol pseudocode and diagrams, andgowhere the block contains Go syntax.Also applies to: 208-208, 229-229, 284-284, 321-321, 375-375, 1161-1161
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docs/improvements/plan-perf.md` at line 164, Add language identifiers to all seven fenced code blocks in plan-perf.md: use text for protocol pseudocode and diagrams, and go for blocks containing Go syntax, resolving the MD040 violations.Source: Linters/SAST tools
pkg/batcher/queue_bench_test.go (1)
56-62: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDo not call
b.Fatalfrom aRunParallelgoroutine.
testingrequiresFatalto be called from the goroutine that runs the benchmark. From aRunParallelbody it only exits that goroutine, and the benchmark then blocks on<-done. An unboundedpushcannot fail today, so this is unreachable, but the failure mode is a hang rather than a report.♻️ Proposed change
b.RunParallel(func(pb *testing.PB) { for pb.Next() { if err := q.push(ctx, 1, sealCh); err != nil { - b.Fatal(err) + b.Error(err) + + return } } })🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/batcher/queue_bench_test.go` around lines 56 - 62, Update the RunParallel callback in the queue benchmark to avoid calling b.Fatal from its worker goroutine; propagate any push error to the benchmark’s coordinating goroutine and report it there, ensuring the benchmark cannot hang if push unexpectedly fails.pkg/batcher/batcher.go (2)
126-146: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winNormalize
b.configtoo, soConfig()reports the effective configuration.
Newnormalizes onlyb.runtime.b.configkeeps the raw values.WithProcessor[T](nil)leavesb.config.ProcessorFuncnil whileb.runtime.processorisNoOpProcessor[T]. A caller that readsConfig().ProcessorFuncthen gets nil and panics on invocation, whichTestWithProcessordoes today. The same divergence applies toBatchSizeandConcurrencyif a future default or option path admits a non-positive value.Normalize the config first, then copy it into the snapshot.
♻️ Proposed change
b.configFrozen.Store(true) + if b.config.BatchSize < 1 { + b.config.BatchSize = DefaultBatchSize + } + + if b.config.Concurrency < 1 { + b.config.Concurrency = 1 + } + + if b.config.ProcessorFunc == nil { + b.config.ProcessorFunc = NoOpProcessor[T] + } + 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] - }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/batcher/batcher.go` around lines 126 - 146, Update New to normalize b.config.BatchSize, b.config.Concurrency, and b.config.ProcessorFunc before constructing b.runtime, applying the existing defaults including NoOpProcessor[T]. Then copy the normalized configuration into the runtime snapshot so Config() reports the effective values and its ProcessorFunc is always callable.
478-499: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winClear
drainedafter the transfer, so it does not retain items.
drainedkeeps up tobatchSizecopies ofTalive until the next drain overwrites them. An idle aggregator holds them indefinitely.queue.popBatchclears its own transferred slots for exactly this reason, so the reuse buffer should match that memory bound.♻️ Proposed change
for _, item := range drained { b.counters.received(1) take(item) } + + // Do not keep transferred items alive in the reuse buffer. + clear(drained) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/batcher/batcher.go` around lines 478 - 499, Clear the reused drained buffer after each transfer in the drainReady closure, once all items have been passed to take, so it no longer retains references while the aggregator is idle. Preserve the existing buffer reuse and popBatch flow.pkg/batcher/queue.go (1)
254-260: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider an atomic depth counter for
length().
Stats().Queuedcallslength(), which takesq.mu. That is the same mutex the profiling work identified as the push-path bottleneck: 86% of mutex delay, 61% of CPU. A metrics scraper pollingStats()at high frequency now contends directly with every producer.Maintain the depth in an
atomic.Int64updated inside the existing locked sections, and read it lock-free here. The value stays exactly as consistent as it is today, becauseStats()is already documented as a snapshot.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/batcher/queue.go` around lines 254 - 260, Replace the mutex-based length calculation in queue.length with a lock-free atomic depth read. Add an atomic.Int64 depth field to queue, update it within the existing locked enqueue and dequeue/removal sections, and have length return that value while preserving the current Stats snapshot semantics.pkg/batcher/runtime_snapshot_test.go (1)
25-35: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winMake
processedatomic.The processor writes
processedfrom a worker goroutine. The main goroutine reads it at line 65. The write is ordered before the read only through the atomic counters thatJoinobserves. That edge is real today, but it is an implicit dependency onJoininternals in a test whose whole purpose is to run under-race. Useatomic.Int64and the dependency disappears.♻️ Proposed change
- var processed int + var processed atomic.Int64 b := New( WithBatchSize[int](1), WithBatchInterval[int](time.Millisecond), WithProcessor(func(items []int) error { - processed += len(items) + processed.Add(int64(len(items))) return nil }), )- require.Positive(t, processed, + require.Positive(t, processed.Load(), "the original processor must have run; a swapped-in one would not count here")Add
"sync/atomic"to the imports.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/batcher/runtime_snapshot_test.go` around lines 25 - 35, Change the processed counter in the batcher runtime snapshot test to use sync/atomic.Int64, update the processor to atomically add the item count, and atomically load the value where the main goroutine reads it. Add the sync/atomic import and preserve the existing assertions and test flow.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@README.md`:
- Around line 299-301: Update the README documentation for s.Pending to describe
it as a conservative drain obligation that may include publishers still inside
the admission gate; state that it is exact for accepted-but-unfinished work only
when PublishersInGate == 0, and preserve the existing s.Accepted description.
In `@test/scenario/allocation_evidence_test.go`:
- Around line 12-29: Update the TestSparseWindowAllocationEvidence documentation
and wastePerFlush calculation to describe and measure the pre-adaptive-capacity
baseline using BatchSize rather than capacities.capacity(). Clarify that the
reported estimate is the baseline used for Milestone 4.2 comparison, and remove
the claim that the test never fails because it fails when result.TimedOut is
true.
---
Outside diff comments:
In `@docs/improvements/plan-perf.md`:
- Around line 953-956: Reconcile the atomic-count statement in the
counter-placement section with the protocol’s five successful-path atomic RMWs.
Explicitly state whether the benchmark excludes gate++ and gate--; if not,
update the reported contended-atomic count and performance comparison to use the
correct total.
- Around line 1032-1035: Define the estimator’s lower clamp bound as min(16,
BatchSize) so configurations accepted by WithBatchSize, including batch sizes 1
and 7, remain valid; alternatively reject values below 16 during WithBatchSize
validation. Update the adaptive-capacity documentation and add tests covering
these small batch sizes.
---
Nitpick comments:
In `@docs/improvements/plan-perf.md`:
- Line 164: Add language identifiers to all seven fenced code blocks in
plan-perf.md: use text for protocol pseudocode and diagrams, and go for blocks
containing Go syntax, resolving the MD040 violations.
In `@pkg/batcher/batcher.go`:
- Around line 126-146: Update New to normalize b.config.BatchSize,
b.config.Concurrency, and b.config.ProcessorFunc before constructing b.runtime,
applying the existing defaults including NoOpProcessor[T]. Then copy the
normalized configuration into the runtime snapshot so Config() reports the
effective values and its ProcessorFunc is always callable.
- Around line 478-499: Clear the reused drained buffer after each transfer in
the drainReady closure, once all items have been passed to take, so it no longer
retains references while the aggregator is idle. Preserve the existing buffer
reuse and popBatch flow.
In `@pkg/batcher/queue_bench_test.go`:
- Around line 56-62: Update the RunParallel callback in the queue benchmark to
avoid calling b.Fatal from its worker goroutine; propagate any push error to the
benchmark’s coordinating goroutine and report it there, ensuring the benchmark
cannot hang if push unexpectedly fails.
In `@pkg/batcher/queue.go`:
- Around line 254-260: Replace the mutex-based length calculation in
queue.length with a lock-free atomic depth read. Add an atomic.Int64 depth field
to queue, update it within the existing locked enqueue and dequeue/removal
sections, and have length return that value while preserving the current Stats
snapshot semantics.
In `@pkg/batcher/runtime_snapshot_test.go`:
- Around line 25-35: Change the processed counter in the batcher runtime
snapshot test to use sync/atomic.Int64, update the processor to atomically add
the item count, and atomically load the value where the main goroutine reads it.
Add the sync/atomic import and preserve the existing assertions and test flow.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: d5cf8a05-3d4a-47fe-a6dd-b791ff69ba38
📒 Files selected for processing (13)
README.mddocs/improvements/plan-perf.mddocs/improvements/thresholds.mdpkg/batcher/batcher.gopkg/batcher/options.gopkg/batcher/options_freeze_test.gopkg/batcher/options_test.gopkg/batcher/queue.gopkg/batcher/queue_bench_test.gopkg/batcher/queue_test.gopkg/batcher/runtime_snapshot_test.gopkg/batcher/stats.gotest/scenario/allocation_evidence_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
- pkg/batcher/stats.go
- pkg/batcher/options_test.go
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.
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.
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.
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.
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.
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.
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.
heynemann
force-pushed
the
phase-4-observability-allocation
branch
from
August 8, 2026 00:50
5d0e5a4 to
ef36391
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Phase 4 of the batcher performance plan: observability and allocation. Both
milestones complete.
feat: complete the Stats snapshot with BatchHeld and BatchesFlushedperf: size batch slices from observed demanddocs: document the completed Stats snapshot4.1 —
BatchHeldcloses a visibility holeThis is not a nice-to-have field. An item that had left the intake queue but not yet
entered a worker was invisible:
Queuedno longer counted it andInFlightdidnot yet. That is exactly where a partial timer batch sits, and where a flushed batch
sits while blocked on the unbuffered worker dispatch.
So the field that would tell an operator "batches are ready but every worker is
busy" did not exist.
Queued,BatchHeld, andInFlightare now disjoint, andwhich one is growing is the diagnosis.
Two ordering decisions matter:
received()is a transfer, not a decrement — intake down,BatchHeldup.dispatched()releasesBatchHeldafter the send to a worker completes.Releasing before would hide the saturation case the field exists to expose.
BatchesFlushedcounts batches, giving mean batch size when paired withCompleted— the coalescing signal for tuning
BatchInterval.Both counters live only on the aggregation/worker path. Verified
publish()still touches nothing new, and benchmarked against Phase 3: no enqueue regression
(geomean -3%, four of five cases statistically indistinguishable).
4.2 — Evidence first, then implementation
The plan gated this on measurement, so I measured before writing the estimator:
BatchSize=1000BatchSize=10000Far above the 2 KB/flush justification threshold, so the milestone was triggered
rather than skipped.
Results versus the full-capacity strategy:
Worst case is inside the +2% budget. Recent-max rather than a mean is the entire
point: the rejected EWMA estimator allocated more than doing nothing on alternating
traffic, because the mean sits between the modes and every large batch grows from a
too-small start. That pattern is now a pinned test.
Two corrections were needed to actually pass the gate, and I want them on record
rather than buried:
batch repeatedly — +2.23%, over budget, for a workload this was never meant to
help. The estimator now starts pessimistic at
BatchSizeand adapts down onfirst evidence.
control but broke decay — a formerly busy batcher would hold an oversized
reservation forever. Removed.
A data race this work exposed
Config()returned the live*Config, so a caller could mutate a running batcher'sbatch size or interval from another goroutine while the aggregator read those fields
per batch. Reading
BatchSizeat goroutine start made it observable under-racewhere it had been hiding —
TestWithBatchSizepassed at 4.1 and failed at 4.2.I verified it was pre-existing rather than newly introduced (stashing 4.2 made the
test pass again), then fixed the cause rather than the symptom:
run()snapshots its configuration once at start.Config()returns a value copy.Newinstead of mutating a running batcher.This was already listed as a Phase 5.1 API break. Leaving a known race in place until
a later phase isn't defensible, so it's fixed here.
Slice ownership is unchanged
Adaptive capacity changes how large a batch slice starts, never its ownership.
retention_test.goasserts a processor may keep its slice and that no two retainedbatches share backing storage — the property that ruled out pooling during planning.
Validation
go test -race ./...andgo vet ./...clean.BatchHeldownership transfer, or the dispatchaccounting, each makes its test fail.
Stats()allocation gate added (TestStatsIsAllocationFree) — metrics scrapingmust not generate garbage.
Dependency context
Stacked on #31 (Phase 3) → #29 → #28. Review bottom-up.
Phase 5 is next, and it answers the original question: the API compatibility
inventory, operating guidance, and the evidence-backed decision on whether
DefaultBatchIntervalshould change from 1s to the 5–10ms range.Stack created with GitHub Stacks CLI • Give Feedback 💬
Summary by CodeRabbit
New Features
BatchHeldandBatchesFlushedstatistics for improved batching visibility.Config()now returns a stable configuration snapshot.Performance
Documentation