From 0e5ce332115db8a309f5c4782fc78c0b5351a112 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Thu, 6 Aug 2026 17:08:44 -0300 Subject: [PATCH 1/8] feat: add Fx options variant and publish the v0.3.0 compatibility inventory Milestone 5.1. ProvideBatcherInFXWithOptions gives Fx applications the full option set. Bounded queues, worker concurrency, close grace and the diagnostics buffer were previously unreachable from Fx without hand-constructing the batcher and losing lifecycle integration. The existing positional ProvideBatcherInFX signature is deliberately untouched, so no current call site breaks; adding parameters there for options most callers do not use would have been the wrong trade. Both variants share one module implementation, so the lifecycle contract cannot drift between them. Fx owns the lifecycle, so WithSkipAutoStart is applied last and cannot be overridden by a caller-supplied option: the batcher must not process anything before the start hook runs. docs/improvements/compatibility.md is the migration guide the plan requires. Every behaviour change is listed with before, after, and what a caller has to do: Config() returning a value, DefaultConcurrency 3 to 1 with a functioning worker pool, Close no longer abandoning the drain, Add no longer panicking after shutdown, Len counting in-flight work, the Go floor moving to 1.25.0, and the rill and chann removals. Target version is v0.3.0. It also states the performance picture honestly rather than only the flattering half: removing the relay layers made enqueue 54% faster, while the admission gate that makes bounded queues and panic-free shutdown possible costs roughly 25-30% on the admission path itself. The net is faster, but someone benchmarking admission alone should expect the gate rather than be surprised by it. compatibility_test.go executes the guide's compatibility claims instead of asserting them in prose: legacy construction, read-only Config() access, manual start, the Errors() range loop, and the original Fx signature all still compile and pass. The one intentional break, assignment through Config(), is deliberately not tested, because it no longer compiles by design. TestConfigSnapshotCannotMutateRunningBatcher is the race proof for that break. Four goroutines hammer Config() and mutate every returned copy while the batcher aggregates and processes 400 items; the live configuration is unaffected and every item is processed. Verified separately that the old pointer form does propagate such a mutation, which is exactly the race this removes. --- docs/improvements/compatibility.md | 257 ++++++++++++++++++++++++ pkg/batcher/compatibility_test.go | 167 ++++++++++++++++ pkg/batcher/fx.go | 93 +++++++-- pkg/batcher/fx_options_test.go | 311 +++++++++++++++++++++++++++++ 4 files changed, 811 insertions(+), 17 deletions(-) create mode 100644 docs/improvements/compatibility.md create mode 100644 pkg/batcher/compatibility_test.go create mode 100644 pkg/batcher/fx_options_test.go diff --git a/docs/improvements/compatibility.md b/docs/improvements/compatibility.md new file mode 100644 index 0000000..41a8eb0 --- /dev/null +++ b/docs/improvements/compatibility.md @@ -0,0 +1,257 @@ +# API compatibility and migration — v0.3.0 + +This release completes the performance and reliability work described in +[`plan-perf.md`](./plan-perf.md). It changes public behaviour, so every change is +listed below with what it was, what it is now, and what a caller has to do. + +## Version + +**v0.2.1 → v0.3.0.** + +The module is `v0.x`, so a minor bump is the correct signal for breaking changes +under Go module semantics: no import path change is required, and callers opt in by +updating. A `v1.0.0` would be the wrong claim to make while the default batch +interval is still under review (see the [decision record](./default-window.md)). + +## Requires Go 1.25.0 + +The module's `go` directive moved **1.22.4 → 1.25.0**. + +This was not a deliberate language-feature upgrade. Modernising dependencies pulled +in `golang.org/x/sys` v0.46.0, which requires 1.25.0. Every CI job now resolves its +toolchain with `go-version-file: go.mod` rather than a hardcoded version, so the +declared floor and the tested toolchain cannot drift apart. + +**Migration:** build with Go 1.25.0 or later. There is no way to keep the older +floor while taking this release. + +## Breaking changes + +### `Config()` returns a value, not a pointer + +| | | +| --- | --- | +| **Before** | `func (b *Batcher[T]) Config() *Config[T]` | +| **After** | `func (b *Batcher[T]) Config() Config[T]` | + +The old signature handed out the live configuration struct. A caller could mutate +`BatchSize`, `BatchInterval` or `ProcessorFunc` while the aggregation goroutine was +reading them — an exported data race, and one that could change batching semantics +mid-batch. It was detected under `-race` once the aggregator began reading +`BatchSize` at start-up. + +Reconfiguring a running batcher was never coherent, so the capability is removed +rather than synchronised. + +**Migration:** + +```go +// Before: read (or, unsafely, mutate) live config. +cfg := b.Config() +size := cfg.BatchSize + +// After: identical to read, since Go auto-dereferences neither is needed. +size := b.Config().BatchSize + +// Before: mutate a running batcher (a data race, now impossible). +b.Config().BatchSize = 500 + +// After: construct a batcher with the configuration you want. +b = batcher.New(batcher.WithBatchSize[Item](500), batcher.WithProcessor(fn)) +``` + +Code that only *reads* `Config()` compiles unchanged. Code that assigned through it +will now fail to compile, which is the intended outcome: it was racing. + +### `DefaultConcurrency` changes 3 → 1, and `Concurrency` now works + +| | | +| --- | --- | +| **Before** | `DefaultConcurrency = 3`, read nowhere; observed parallelism always 1 | +| **After** | `DefaultConcurrency = 1`, honoured by a real worker pool | + +The old constant was advertised but never used, so every batcher processed one batch +at a time. Making it functional at its old value would have silently parallelised +every existing caller's processor — including processors holding unsynchronised +state. The default therefore matches the behaviour callers already had. + +Concurrency above 1 requires an explicit acknowledgement, because it gives up two +guarantees: + +```go +b := batcher.New( + batcher.WithProcessor(fn), + batcher.WithConcurrency[Item](4), + batcher.WithoutOrderedProcessing[Item](), // required; construction panics without it +) +``` + +**Migration:** no action to keep current behaviour. Callers who *want* parallelism +must add both options and ensure the processor is goroutine-safe. + +### `Close()` no longer abandons the drain + +| | | +| --- | --- | +| **Before** | On timeout, signalled done and discarded remaining batches without processing them | +| **After** | Reports an incomplete drain; accepted work keeps being processed | + +The old behaviour was silent data loss. A partial batch with an interval beyond the +internal cap was dropped: 50 items accepted, 0 processed, and `Len()` still +reporting 50. + +The timeout is now a flat, configurable 30s (`WithCloseGrace`) instead of a formula +derived from queue size and interval. + +**Migration:** handle the error rather than discarding it. + +```go +// Before: silently lost work on timeout. +defer batcher.Close() + +// After: report it, and decide deliberately. +defer func() { + if err := batcher.Close(); err != nil { + log.Printf("batcher shutdown incomplete: %v", err) + } +}() +``` + +If you need to keep waiting, use `Shutdown`, which is resumable: + +```go +if err := b.Shutdown(ctx); err != nil { + var incomplete *batcher.ShutdownIncompleteError + if errors.As(err, &incomplete) { + // Nothing was lost. Wait longer on the same drain. + err = b.Shutdown(context.Background()) + } +} +``` + +### `Add` after shutdown no longer panics + +| | | +| --- | --- | +| **Before** | Panicked with `send on closed channel`, crashing the caller's goroutine | +| **After** | A counted no-op; `Stats().Rejected` increments | + +**Migration:** none required — this only removes a crash. Callers that need to +observe rejection should use `Enqueue`, which returns `ErrClosing`. + +### `Len()` counts more than the queue + +| | | +| --- | --- | +| **Before** | Roughly "items waiting", and could transiently go negative | +| **After** | Accepted-but-unfinished work: queued **plus** aggregator-held **plus** in-flight **plus** gate reservations | + +`Len()` can therefore be non-zero while the queue is empty — for example while a +batch is inside the processor — and can briefly exceed accepted work while a +publisher is mid-publish. It can no longer be negative. + +**Migration:** if you were using `Len()` as queue depth, use `Stats().Queued`, which +is exactly that. `Len()` remains the right value for "is there outstanding work". + +### `ProvideBatcherInFX` is unchanged; new variant added + +The existing signature compiles unchanged, deliberately: no positional parameter was +added for options most callers do not use. + +```go +// Unchanged. +batcher.ProvideBatcherInFX[Item](factory, batchSize, batchInterval) + +// New: full option set. +batcher.ProvideBatcherInFXWithOptions[Item](factory, + batcher.WithBatchSize[Item](500), + batcher.WithMaxQueueSize[Item](10_000), + batcher.WithConcurrency[Item](4), + batcher.WithoutOrderedProcessing[Item](), +) +``` + +Both variants forward the Fx stop-hook context to `Shutdown`, so the application's +shutdown deadline governs the drain. Previously the hook discarded that context and +used the batcher's own timeout, so an app with a longer grace period could not use it +and one with a shorter deadline could not bound it. + +Fx stop-hook behaviour: + +| Case | Result | +| --- | --- | +| Normal drain | `nil` | +| Stop context expires | `*ShutdownIncompleteError` as an Fx lifecycle error; drain continues | +| Batcher never started but holds queued work | Drains; `Shutdown` starts a consumer when one is needed | +| Repeated stop | Idempotent; later calls observe the same terminal result | + +## Clarified contracts (no code change required) + +### `Join` is a quiescence snapshot, not a barrier + +`Join` returns when no work is pending. A concurrent `Add` can make work pending +again the instant it returns, so it is only authoritative **after admission is +sealed**. The 1ms polling implementation is unchanged: it is off the operational hot +path. Benchmarks must not use it as a completion signal, because 1ms polling cannot +resolve sub-10ms windows. + +### `Stats()` is eventually consistent + +Each field is an independent atomic load, so a live snapshot can show a state +transition partially applied. Use it to observe *where* work sits. The accounting +identity `Accepted == Completed + Failed + Panicked + Pending` holds only at +quiescence, after a terminal drain. + +`Queued`, `BatchHeld` and `InFlight` are disjoint, so which one is growing is the +diagnosis. `Stats()` takes the queue mutex for a single length check to read +`Queued`; scraping frequently is fine, scraping in a tight loop is not. + +## Removed dependencies + +`github.com/destel/rill` and `golang.design/x/chann` are both gone. Batching and +queueing are now owned in-repo. + +`chann` had to go for a correctness reason, not a preference: its unbounded queue +owns a relay goroutine whose only exit path closes its own ingress channel — the +channel publishers send to. That made "never close the input" and "leak no +goroutines" mutually exclusive, and the first is what removes the shutdown panic. + +Measured effect: goroutines per running batcher went **6 → 2** (aggregator plus +serial processor; `1 + n` with `n` workers), with none leaked after shutdown. + +## Performance notes + +Two changes pull in opposite directions, and both are worth stating plainly. + +**Enqueue got faster.** Removing the relay layers removed a channel hop and a +goroutine handoff per item: **-54% geomean sec/op** on the enqueue microbenchmarks +(`Add` 229.8 ns → 63.4 ns at small batch sizes), with 49-90% fewer bytes per +operation. + +**Bounded admission costs something.** The safety machinery — the publisher gate, +reservation-before-publish, and rollback — adds work to every `Add`. Measured in +isolation during planning, that admission path was **~25-30% slower** per call than a +bare channel send. It is paid for by removing a class of shutdown panic and by making +back-pressure possible at all. + +The net result is still faster than v0.2.1, because the relay removal outweighs the +gate. But a caller comparing a micro-benchmark of admission alone should expect the +gate cost, not be surprised by it. + +**Allocation.** Batch slices are now sized from observed demand rather than always +reserving `BatchSize`. Sparse timer-driven workloads improved ~98% in allocated +bytes; full-batch workloads are unchanged; the worst case across the tested matrix is ++1.12%. + +## What has not changed + +- `Add`, `Join`, `Errors`, `Start`, `Close`, `IsClosed` all still exist with the same + names. +- Batching rules are identical: the interval timer arms on the first item of an empty + batch, a full batch flushes immediately, empty batches are never emitted, and + shutdown flushes the partial batch. These are pinned by characterization tests + written against v0.2.x behaviour before the engine was replaced. +- The queue is still **unbounded by default**. Bounding it is opt-in via + `WithMaxQueueSize`. +- A processor may still retain the batch slice it receives. No pooling was + introduced, precisely so this stays true. diff --git a/pkg/batcher/compatibility_test.go b/pkg/batcher/compatibility_test.go new file mode 100644 index 0000000..314af88 --- /dev/null +++ b/pkg/batcher/compatibility_test.go @@ -0,0 +1,167 @@ +package batcher_test + +import ( + "testing" + "time" + + "github.com/NSXBet/batcher/pkg/batcher" + "github.com/stretchr/testify/require" + "go.uber.org/fx" + "go.uber.org/fx/fxtest" +) + +// Compile-level compatibility checks for v0.3.0. +// +// The migration guide in docs/improvements/compatibility.md claims certain v0.2.x +// call shapes still compile. These tests are that claim, executed: if a future change +// breaks one of them, this file stops building rather than the guide quietly becoming +// wrong. +// +// Deliberately NOT asserted here: assignment through Config(), which v0.2.x allowed +// and which no longer compiles. That break is intentional — it was a data race — and +// is declared in the migration guide instead. + +type legacyItem struct { + ID int +} + +type legacyProcessor struct { + seen int +} + +func newLegacyProcessor() *legacyProcessor { + return &legacyProcessor{} +} + +func (p *legacyProcessor) Process(items []*legacyItem) error { + p.seen += len(items) + + return nil +} + +// TestLegacyConstructionStillCompiles pins the constructor and option shapes from +// the v0.2.x README. +func TestLegacyConstructionStillCompiles(t *testing.T) { + t.Parallel() + + b := batcher.New[*legacyItem]( + batcher.WithBatchSize[*legacyItem](100), + batcher.WithBatchInterval[*legacyItem](time.Second), + batcher.WithProcessor(func(items []*legacyItem) error { + return nil + }), + ) + + b.Add(&legacyItem{ID: 1}) + + require.NoError(t, b.Join(10*time.Second)) + require.NoError(t, b.Close()) + require.True(t, b.IsClosed()) +} + +// TestLegacyConfigReadsStillCompile pins that read-only Config() access is +// unaffected by the pointer-to-value change. Only assignment broke. +func TestLegacyConfigReadsStillCompile(t *testing.T) { + t.Parallel() + + b := batcher.New[*legacyItem]( + batcher.WithBatchSize[*legacyItem](250), + batcher.WithBatchInterval[*legacyItem](2*time.Second), + ) + + defer func() { require.NoError(t, b.Close()) }() + + // Field access through the call, as v0.2.x code did. + require.Equal(t, 250, b.Config().BatchSize) + require.Equal(t, 2*time.Second, b.Config().BatchInterval) + require.NotNil(t, b.Config().ProcessorFunc) + + // Assigning the result to a local also still works; it is simply a copy now. + cfg := b.Config() + require.Equal(t, 250, cfg.BatchSize) +} + +// TestLegacySkipAutoStartStillCompiles pins the manual-start lifecycle shape. +func TestLegacySkipAutoStartStillCompiles(t *testing.T) { + t.Parallel() + + b := batcher.New[*legacyItem]( + batcher.WithSkipAutoStart[*legacyItem](), + batcher.WithBatchSize[*legacyItem](2), + batcher.WithBatchInterval[*legacyItem](5*time.Millisecond), + batcher.WithProcessor(func([]*legacyItem) error { return nil }), + ) + + b.Start() + + b.Add(&legacyItem{ID: 1}) + b.Add(&legacyItem{ID: 2}) + + require.NoError(t, b.Join(10*time.Second)) + require.NoError(t, b.Close()) +} + +// TestLegacyErrorsChannelStillCompiles pins the diagnostics consumption shape. +func TestLegacyErrorsChannelStillCompiles(t *testing.T) { + t.Parallel() + + b := batcher.New[*legacyItem]( + batcher.WithBatchSize[*legacyItem](1), + batcher.WithBatchInterval[*legacyItem](time.Millisecond), + batcher.WithProcessor(func([]*legacyItem) error { return nil }), + ) + + drained := make(chan struct{}) + + go func() { + defer close(drained) + + for range b.Errors() { + } + }() + + b.Add(&legacyItem{ID: 1}) + + require.NoError(t, b.Close()) + + select { + case <-drained: + case <-time.After(10 * time.Second): + t.Fatal("Errors() must still close on shutdown so a range loop terminates") + } +} + +// TestLegacyProvideBatcherInFXSignatureStillCompiles pins the Fx entry point that +// existing applications wire up. Adding options must not have required a positional +// parameter here. +func TestLegacyProvideBatcherInFXSignatureStillCompiles(t *testing.T) { + t.Parallel() + + var ( + b *batcher.Batcher[*legacyItem] + p *legacyProcessor + ) + + app := fxtest.New(t, + fx.Provide(newLegacyProcessor), + batcher.ProvideBatcherInFX[*legacyItem]( + func(processor *legacyProcessor) batcher.Processor[*legacyItem] { + return processor.Process + }, + 2, + 50*time.Millisecond, + ), + fx.Populate(&b, &p), + ) + + app.RequireStart() + + b.Add(&legacyItem{ID: 1}) + b.Add(&legacyItem{ID: 2}) + + require.NoError(t, b.Join(10*time.Second)) + + app.RequireStop() + + require.Equal(t, 2, p.seen) +} diff --git a/pkg/batcher/fx.go b/pkg/batcher/fx.go index a9fab4a..fd40ac0 100644 --- a/pkg/batcher/fx.go +++ b/pkg/batcher/fx.go @@ -7,10 +7,79 @@ import ( "go.uber.org/fx" ) +// ProvideBatcherInFX wires a Batcher into an Fx application using the three +// settings most applications configure. +// +// The signature is deliberately unchanged: every existing call site keeps +// compiling. Anything beyond batch size and interval — bounded queues, worker +// concurrency, close grace, diagnostics buffer — needs +// ProvideBatcherInFXWithOptions instead, because adding positional parameters here +// would break callers for options most of them do not use. +// +// Lifecycle behaviour is identical in both variants; see the stop-hook notes on +// provideBatcherModule. func ProvideBatcherInFX[T any]( processorFactory any, batchSize int, batchInterval time.Duration, +) fx.Option { + return provideBatcherModule[T]( + processorFactory, + func(processorFunc Processor[T]) []Option[T] { + return []Option[T]{ + WithProcessor(processorFunc), + WithBatchSize[T](batchSize), + WithBatchInterval[T](batchInterval), + } + }, + ) +} + +// ProvideBatcherInFXWithOptions wires a Batcher into an Fx application with the +// full option set. +// +// The processor comes from the injected factory, so callers must not pass +// WithProcessor: it is applied first and then overridden by anything supplied +// here, which would silently bypass dependency injection. Every other option +// behaves exactly as it does with New. +// +// WithSkipAutoStart is always applied last and cannot be overridden, because Fx +// owns the lifecycle: the batcher must not begin processing until the start hook +// runs. +func ProvideBatcherInFXWithOptions[T any]( + processorFactory any, + options ...Option[T], +) fx.Option { + return provideBatcherModule[T]( + processorFactory, + func(processorFunc Processor[T]) []Option[T] { + // The injected processor goes first so an explicit WithProcessor in + // options would win. That is a caller mistake rather than something to + // silently correct, and it is documented above. + return append([]Option[T]{WithProcessor(processorFunc)}, options...) + }, + ) +} + +// provideBatcherModule builds the Fx module shared by both entry points. +// +// Lifecycle contract: +// +// - Start hook: starts processing. The batcher is constructed with +// WithSkipAutoStart so nothing is processed before Fx says so. +// - Stop hook: forwards the hook's context to Shutdown, so the application's +// shutdown deadline governs the drain rather than the batcher's own grace +// period. A normal drain returns nil. +// - If the stop context expires, Shutdown returns *ShutdownIncompleteError and +// Fx reports it as a lifecycle error, while the drain continues in the +// background. Accepted work is never discarded to make shutdown look clean. +// - A batcher that never started but holds queued work still drains, because +// Shutdown starts the consumer when one is needed. +// - Repeated application stop is idempotent: later Shutdown calls observe the +// same terminal result. +func provideBatcherModule[T any]( + processorFactory any, + buildOptions func(Processor[T]) []Option[T], ) fx.Option { return fx.Module( "batcher", @@ -20,14 +89,13 @@ func ProvideBatcherInFX[T any]( ), fx.Provide( func(processorFunc Processor[T]) *Batcher[T] { - b := New( - WithProcessor(processorFunc), - WithBatchSize[T](batchSize), - WithBatchInterval[T](batchInterval), - WithSkipAutoStart[T](), - ) - - return b + options := buildOptions(processorFunc) + + // Fx owns the lifecycle, so auto-start is forced off last and cannot + // be re-enabled by a caller-supplied option. + options = append(options, WithSkipAutoStart[T]()) + + return New(options...) }, ), fx.Invoke( @@ -38,15 +106,6 @@ func ProvideBatcherInFX[T any]( return nil })) - // Forward the stop-hook context instead of discarding it, so the - // application's shutdown deadline governs the drain. Previously this - // used Close's own timeout, which meant an app with a longer grace - // period could not use it, and one with a shorter deadline could not - // bound it. - // - // If the hook's context expires the drain continues in the background - // and the error describes what remained, rather than silently - // discarding accepted work. lifecycle.Append(fx.StopHook(func(ctx context.Context) error { return batcher.Shutdown(ctx) })) diff --git a/pkg/batcher/fx_options_test.go b/pkg/batcher/fx_options_test.go new file mode 100644 index 0000000..1e38d5d --- /dev/null +++ b/pkg/batcher/fx_options_test.go @@ -0,0 +1,311 @@ +package batcher_test + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/NSXBet/batcher/pkg/batcher" + "github.com/stretchr/testify/require" + "go.uber.org/fx" + "go.uber.org/fx/fxtest" +) + +type optionsProcessor struct { + batches atomic.Int64 + items atomic.Int64 +} + +func newOptionsProcessor() *optionsProcessor { + return &optionsProcessor{} +} + +// optionsProcessorFunc is the factory Fx injects: it depends on the processor +// struct and returns the Processor[T] the batcher needs. +func optionsProcessorFunc(p *optionsProcessor) batcher.Processor[string] { + return p.Process +} + +func (p *optionsProcessor) Process(items []string) error { + p.batches.Add(1) + p.items.Add(int64(len(items))) + + return nil +} + +// TestProvideBatcherInFXWithOptionsAppliesEveryOption pins that the options +// variant is a full replacement for the positional one: anything New accepts, +// Fx accepts. +// +// The positional ProvideBatcherInFX can only express batch size and interval, so +// bounded queues and worker concurrency were previously unreachable from Fx +// without constructing the batcher by hand and losing lifecycle integration. +func TestProvideBatcherInFXWithOptionsAppliesEveryOption(t *testing.T) { + t.Parallel() + + var b *batcher.Batcher[string] + + app := fxtest.New(t, + fx.Provide(newOptionsProcessor), + batcher.ProvideBatcherInFXWithOptions[string]( + optionsProcessorFunc, + batcher.WithBatchSize[string](7), + batcher.WithBatchInterval[string](11*time.Millisecond), + batcher.WithMaxQueueSize[string](64), + batcher.WithCloseGrace[string](3*time.Second), + batcher.WithErrorBufferSize[string](16), + batcher.WithConcurrency[string](3), + batcher.WithoutOrderedProcessing[string](), + ), + fx.Populate(&b), + ) + + app.RequireStart() + + config := b.Config() + + require.Equal(t, 7, config.BatchSize) + require.Equal(t, 11*time.Millisecond, config.BatchInterval) + require.Equal(t, 64, config.MaxQueueSize) + require.Equal(t, 3*time.Second, config.CloseGrace) + require.Equal(t, 16, config.ErrorBufferSize) + require.Equal(t, 3, config.Concurrency) + require.True(t, config.UnorderedProcessingAcknowledged) + + app.RequireStop() +} + +// TestProvideBatcherInFXWithOptionsUsesInjectedProcessor pins that the processor +// still comes from dependency injection rather than from the option list. +func TestProvideBatcherInFXWithOptionsUsesInjectedProcessor(t *testing.T) { + t.Parallel() + + var ( + b *batcher.Batcher[string] + processor *optionsProcessor + ) + + app := fxtest.New(t, + fx.Provide(newOptionsProcessor), + batcher.ProvideBatcherInFXWithOptions[string]( + optionsProcessorFunc, + batcher.WithBatchSize[string](2), + batcher.WithBatchInterval[string](5*time.Millisecond), + ), + fx.Populate(&b, &processor), + ) + + app.RequireStart() + + for i := range 6 { + b.Add(string(rune('a' + i))) + } + + require.NoError(t, b.Join(10*time.Second)) + + app.RequireStop() + + require.Equal(t, int64(6), processor.items.Load(), + "the injected processor must receive every item") + require.Positive(t, processor.batches.Load()) +} + +// TestFxStopHookDrainsQueuedWork pins the stop-hook contract: work accepted before +// shutdown is drained, not discarded, and a clean drain reports no error. +func TestFxStopHookDrainsQueuedWork(t *testing.T) { + t.Parallel() + + var ( + b *batcher.Batcher[string] + processor *optionsProcessor + ) + + app := fxtest.New(t, + fx.Provide(newOptionsProcessor), + batcher.ProvideBatcherInFXWithOptions[string]( + optionsProcessorFunc, + // Large batch and long interval: only the stop hook can flush this. + batcher.WithBatchSize[string](1_000), + batcher.WithBatchInterval[string](time.Hour), + ), + fx.Populate(&b, &processor), + ) + + app.RequireStart() + + for i := range 5 { + b.Add(string(rune('a' + i))) + } + + app.RequireStop() + + require.Equal(t, int64(5), processor.items.Load(), + "the stop hook must flush the partial batch rather than discard it") + require.True(t, b.IsClosed()) +} + +// TestFxStopHookReportsIncompleteDrain pins that a stop context which expires is +// reported rather than hidden, and that the drain continues afterwards. +// +// This is the honest-failure case: Fx surfaces a lifecycle error, and the batcher +// stays in draining rather than claiming to be closed. +func TestFxStopHookReportsIncompleteDrain(t *testing.T) { + t.Parallel() + + release := make(chan struct{}) + defer close(release) + + blocking := func() batcher.Processor[string] { + return func([]string) error { + <-release + + return nil + } + } + + var b *batcher.Batcher[string] + + app := fx.New( + batcher.ProvideBatcherInFXWithOptions[string]( + blocking, + batcher.WithBatchSize[string](1), + batcher.WithBatchInterval[string](time.Millisecond), + ), + fx.Populate(&b), + fx.NopLogger, + ) + + startCtx, cancelStart := context.WithTimeout(context.Background(), 10*time.Second) + defer cancelStart() + + require.NoError(t, app.Start(startCtx)) + + b.Add("blocked") + + require.Eventually(t, func() bool { + return b.Stats().InFlight > 0 + }, 5*time.Second, time.Millisecond) + + stopCtx, cancelStop := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancelStop() + + err := app.Stop(stopCtx) + + require.Error(t, err, + "an expired stop context must be reported, not silently treated as a clean stop") + require.True(t, b.IsClosing(), "admission must be sealed") + require.False(t, b.IsClosed(), + "a batcher whose processor is still running has not finished draining") +} + +// TestFxRepeatedStopIsIdempotent pins that stopping twice is safe, since a +// supervisor or test harness may do so. +func TestFxRepeatedStopIsIdempotent(t *testing.T) { + t.Parallel() + + var b *batcher.Batcher[string] + + app := fx.New( + fx.Provide(newOptionsProcessor), + batcher.ProvideBatcherInFXWithOptions[string]( + optionsProcessorFunc, + batcher.WithBatchSize[string](4), + batcher.WithBatchInterval[string](2*time.Millisecond), + ), + fx.Populate(&b), + fx.NopLogger, + ) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + require.NoError(t, app.Start(ctx)) + + b.Add("one") + + require.NoError(t, app.Stop(ctx)) + require.True(t, b.IsClosed()) + + // Fx itself will not run the stop hooks twice, so call Shutdown directly to + // prove the underlying operation is idempotent. + require.NoError(t, b.Shutdown(ctx)) + require.True(t, b.IsClosed()) +} + +// TestConfigSnapshotCannotMutateRunningBatcher is the race proof required for the +// Config() change. +// +// Config() previously returned the live *Config, so a caller could change batch +// size, interval or the processor while the aggregation goroutine was reading them. +// It now returns a value copy: mutating the returned struct must be inert, and +// doing so concurrently with a running batcher must not be a data race. +// +// Run under -race, this fails if Config() ever returns shared state again. +func TestConfigSnapshotCannotMutateRunningBatcher(t *testing.T) { + t.Parallel() + + var processed atomic.Int64 + + b := batcher.New( + batcher.WithBatchSize[int](8), + batcher.WithBatchInterval[int](2*time.Millisecond), + batcher.WithProcessor(func(items []int) error { + processed.Add(int64(len(items))) + + return nil + }), + ) + + original := b.Config() + + var wg sync.WaitGroup + + // Hammer Config() and mutate every returned copy while the batcher is actively + // aggregating and processing. + for range 4 { + wg.Add(1) + + go func() { + defer wg.Done() + + for range 200 { + snapshot := b.Config() + + snapshot.BatchSize = 1 + snapshot.BatchInterval = time.Hour + snapshot.Concurrency = 99 + snapshot.MaxQueueSize = 7 + snapshot.ProcessorFunc = nil + } + }() + } + + wg.Add(1) + + go func() { + defer wg.Done() + + for i := range 400 { + b.Add(i) + } + }() + + wg.Wait() + + require.NoError(t, b.Shutdown(context.Background())) + + // The live configuration is untouched by any of those writes. + current := b.Config() + + require.Equal(t, original.BatchSize, current.BatchSize) + require.Equal(t, original.BatchInterval, current.BatchInterval) + require.Equal(t, original.Concurrency, current.Concurrency) + require.Equal(t, original.MaxQueueSize, current.MaxQueueSize) + require.NotNil(t, current.ProcessorFunc, + "nilling the snapshot's processor must not disarm the running batcher") + + require.Equal(t, int64(400), processed.Load(), + "every item must still be processed with the original configuration") +} From a336b75eb71e1fa1f4fa1124330da2381b66c137 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Thu, 6 Aug 2026 17:24:21 -0300 Subject: [PATCH 2/8] feat: set the default batch interval to 10ms on measured evidence Milestones 5.2 and 5.3. DefaultBatchInterval changes 1s -> 10ms, with the full matrix and reasoning published in docs/improvements/default-window.md. The decision rests on four measurements, not on the intuition that smaller is better: - Latency and coalescing across 1ms..1s at 1k, 10k and 50k items/s. At 1k/s a 1s default measured p99 981ms; 10ms measured p99 12ms while still coalescing ~11 items per batch. At 10k/s, p99 99ms -> 10ms with ~101 items per batch. - The sizing rule rate x interval predicts batch size: predicted 100, measured 100 at 10k/s with a 10ms interval. Asserted by a test, because the tuning guidance is only trustworthy if that relationship is real. - Sustained overload. Every interval from 10ms to 1s behaved identically under a saturated downstream: same queue peak, same heap, same ~199 downstream calls/s, because BatchSize binds before the timer does. This is the check that stopped the decision resting on latency alone. - Bursts. 10ms halved p99 against 100ms while only doubling downstream calls. Two findings shaped the choice against going lower. 1ms *increases* downstream load under saturation (495 calls/s versus 199) because it flushes before batches fill, and at 1k/s it coalesces only 2 items. 10ms is where latency is bounded by the window while coalescing stays meaningful at every rate tested. Notably, the old 1s default was already inert above ~10k/s: 1s and 100ms produced identical results because BatchSize=1000 filled first. Its only real effect was on sparse traffic, where it did the most latency damage and bought the least. The record states plainly where 10ms is the wrong choice: a low-rate service with an expensive downstream should configure a larger interval, no interval provides overload protection, and a slow processor at Concurrency=1 bounds the effective interval, so a 5ms window behind a 50ms processor measured p50 120ms versus 100ms for a 100ms window. Also fixes two option fallbacks that hardcoded literals instead of the constants, so WithBatchInterval(0) no longer silently reverts to 1s now that the default has moved. README gains a "Choosing a batch interval" section with the coalescing table, the overload disclaimer, the slow-processor inversion, the complete option list, and the new Fx options variant. Corrects the stale coverage claim to what CI actually enforces. --- README.md | 98 ++++++++++- docs/improvements/default-window.md | 195 ++++++++++++++++++++++ docs/improvements/plan-perf.md | 3 + pkg/batcher/constants.go | 14 +- pkg/batcher/options.go | 16 +- test/scenario/decision_test.go | 249 ++++++++++++++++++++++++++++ 6 files changed, 562 insertions(+), 13 deletions(-) create mode 100644 docs/improvements/default-window.md create mode 100644 test/scenario/decision_test.go diff --git a/README.md b/README.md index 60722e1..7e2bd70 100644 --- a/README.md +++ b/README.md @@ -28,8 +28,8 @@ type BatchItem struct { func main() { // create a batcher batcher := batcher.New[*BatchItem]( - batcher.WithBatchSize[*BatchItem](100), // will batch each 100 items. - batcher.WithBatchInterval[*BatchItem](1*time.Second), // or each second. + batcher.WithBatchSize[*BatchItem](100), // flush at 100 items, + batcher.WithBatchInterval[*BatchItem](10*time.Millisecond), // or when the oldest item is 10ms old. // then run this processor with each batch batcher.WithProcessor(func(items []*BatchItem) error { fmt.Printf("processing batch with %d items...\n", len(items)) @@ -83,8 +83,8 @@ To create a batcher you can use the `New` function: ```go batcher := batcher.New[*BatchItem]( - batcher.WithBatchSize[*BatchItem](100), // will batch each 100 items. - batcher.WithBatchInterval[*BatchItem](1*time.Second), // or each second. + batcher.WithBatchSize[*BatchItem](100), // flush at 100 items, + batcher.WithBatchInterval[*BatchItem](10*time.Millisecond), // or when the oldest item is 10ms old. // then run this processor with each batch batcher.WithProcessor(func(items []*BatchItem) error { fmt.Printf("processing batch with %d items...\n", len(items)) @@ -344,9 +344,60 @@ fmt.Printf("batcher has %d items\n", batcher.Len()) ## Available Options to configure batcher -- `WithBatchSize[*BatchItem](size int)`: sets the batch size. -- `WithBatchInterval[*BatchItem](interval time.Duration)`: sets the batch interval. -- `WithProcessor(func(items []*BatchItem) error)`: sets the processor function. +| Option | Default | Purpose | +| --- | --- | --- | +| `WithProcessor(fn)` | no-op | The function called with each batch. | +| `WithBatchSize[T](n)` | 1000 | Flush as soon as a batch holds `n` items. | +| `WithBatchInterval[T](d)` | **10ms** | Maximum age of a partial batch. See [Choosing a batch interval](#choosing-a-batch-interval). | +| `WithMaxQueueSize[T](n)` | unbounded | Bound queued items to get back-pressure instead of unbounded growth. | +| `WithConcurrency[T](n)` | 1 | Process `n` batches at once. Requires `WithoutOrderedProcessing`. | +| `WithoutOrderedProcessing[T]()` | off | Acknowledges that `n > 1` gives up cross-batch ordering and processor mutual exclusion. | +| `WithCloseGrace[T](d)` | 30s | How long `Close` waits before reporting an incomplete drain. | +| `WithErrorBufferSize[T](n)` | 1024 | How many diagnostics `Errors()` buffers before dropping newer ones. | +| `WithSkipAutoStart[T]()` | off | Do not start processing in `New`; call `Start` yourself. | + +## Choosing a batch interval + +The interval is the **maximum age of a partial batch**, not a periodic flush tick: +the timer starts when the first item enters an empty batch. Sparse traffic therefore +waits the full interval per item, and a request crossing several batching services +pays up to one interval per hop. + +Expected batch size is approximately **arrival rate x interval**. That rule is +measured, not estimated: at 10,000 items/s with a 10ms interval it predicts 100 items +per batch, and the measured mean is 100. + +Use it to pick an interval from the coalescing you actually need: + +| Arrival rate | 10ms (default) | 100ms | +| --- | --- | --- | +| 1,000/s | ~11 items/batch, ~94 calls/s, p99 ~12ms | ~100 items/batch, ~10 calls/s, p99 ~100ms | +| 10,000/s | ~101 items/batch, ~99 calls/s, p99 ~10ms | ~1000 items/batch, ~10 calls/s, p99 ~99ms | +| 50,000/s | ~500 items/batch, ~100 calls/s, p99 ~10ms | ~1000 items/batch, ~50 calls/s, p99 ~20ms | + +Raise the interval when your downstream is expensive enough that the call rate at +10ms is unacceptable — most relevant below a few thousand items/s. The full matrix +and the reasoning behind the default are in +[`docs/improvements/default-window.md`](./docs/improvements/default-window.md). + +**Shrinking the interval is not overload protection.** Measured with a saturated +downstream, every interval from 10ms to 1s accepted the same offered load while +completing the same amount, with the queue absorbing the difference. Use +`WithMaxQueueSize` and `Enqueue` for that, or upstream flow control. + +**A slow processor bounds the effective interval.** At the default `Concurrency` of 1, +one batch must finish before the next starts, so the effective interval is +`max(BatchInterval, processor duration)`. Measured with a 50ms processor at 10k +items/s: + +| Configured interval | Effective behaviour | p50 latency | +| --- | --- | --- | +| 5ms | bounded by the 50ms processor | **120ms** | +| 100ms | bounded by the interval | **100ms** | + +The smaller interval is *worse*, because queueing dominates. If your processor is +slow, raise `WithConcurrency` (with `WithoutOrderedProcessing`) rather than lowering +the interval — with 8 workers the same scenario measured p50 4ms. ## FX Integration @@ -431,6 +482,39 @@ func main() { } ``` +### Configuring the Fx batcher beyond size and interval + +`ProvideBatcherInFX` takes batch size and interval positionally, which covers most +applications. For anything else — bounded queues, worker concurrency, close grace, +diagnostics buffer — use `ProvideBatcherInFXWithOptions`, which accepts the same +options as `New`: + +```go +batcher.ProvideBatcherInFXWithOptions[*BatchItem]( + func(processor *Processor) batcher.Processor[*BatchItem] { + return processor.Process + }, + batcher.WithBatchSize[*BatchItem](500), + batcher.WithBatchInterval[*BatchItem](10*time.Millisecond), + batcher.WithMaxQueueSize[*BatchItem](10_000), + batcher.WithConcurrency[*BatchItem](4), + batcher.WithoutOrderedProcessing[*BatchItem](), +) +``` + +Do not pass `WithProcessor` here: the processor comes from the injected factory, and +supplying it as an option would bypass dependency injection. + +Both variants forward the Fx stop-hook context to `Shutdown`, so your application's +shutdown deadline governs the drain: + +| Case | Result | +| --- | --- | +| Normal drain | `nil` | +| Stop context expires | `*ShutdownIncompleteError` as an Fx lifecycle error; the drain continues in the background | +| Batcher holds queued work but never started | Drains anyway | +| Repeated stop | Idempotent | + The key part of the FX integration is the processor function passed to `ProvideBatcherInFX`. This function is an FX resolver that: 1. Takes any dependencies you need (like the `Processor` struct) as parameters diff --git a/docs/improvements/default-window.md b/docs/improvements/default-window.md new file mode 100644 index 0000000..0d8a6b7 --- /dev/null +++ b/docs/improvements/default-window.md @@ -0,0 +1,195 @@ +# Decision record: default batch interval + +**Decision:** `DefaultBatchInterval` changes from **1s to 10ms** in v0.3.0. + +**Status:** accepted, with the caveat in [Where 10ms is the wrong +choice](#where-10ms-is-the-wrong-choice). + +This record exists because the plan required the default to follow measurement +rather than intuition. If the evidence had been ambiguous, the correct outcome was to +keep 1s and publish a configuration recommendation instead. It was not ambiguous. + +## The problem being solved + +The batch interval is a maximum *partial-batch age*, not a periodic flush tick: the +timer arms when the first item enters an empty batch. A request that crosses several +batching services therefore pays up to one interval **per hop**, and the cost is +worst for sparse traffic, which is exactly the traffic that gains least from +batching. + +With a 1s default, a five-hop sparse path could accumulate several seconds before any +downstream work happened. + +## Environment + +``` +go=go1.26.5 os=darwin arch=arm64 gomaxprocs=12 cpus=12 +``` + +Absolute microseconds are host-specific. The **shape** of the relationship — and the +order of magnitude of every comparison below — is what the decision rests on. All +runs are open-loop with schedule-lateness validation; every row reported here passed +that validity check, so none of these numbers describe a run where the load generator +itself fell behind. + +Reproduce with: + +```sh +SCENARIO_MATRIX=1 go test -run TestDefaultWindow -timeout 30m -v ./test/scenario/ +``` + +## Evidence 1: latency and coalescing versus window + +`BatchSize=1000`, no-op processor, so this isolates batching cost from downstream +cost. + +### 1,000 items/s + +| Window | p50 | p99 | p99.9 | mean batch | calls/s | allocs/item | +| --- | --- | --- | --- | --- | --- | --- | +| 1ms | 1.21ms | 1.53ms | 2.25ms | 2 | 562 | 0.668 | +| 5ms | 3.21ms | 7.59ms | 13.1ms | 6 | 174 | 0.211 | +| **10ms** | **5.29ms** | **12.1ms** | **21.7ms** | **11** | **94** | **0.134** | +| 100ms | 51.2ms | 100ms | 100ms | 100 | 10 | 0.039 | +| 1s (old default) | 450ms | **981ms** | 997ms | 1000 | 1 | 0.033 | + +### 10,000 items/s + +| Window | p50 | p99 | p99.9 | mean batch | calls/s | allocs/item | +| --- | --- | --- | --- | --- | --- | --- | +| 1ms | 533µs | 1.05ms | 1.15ms | 11 | 929 | 0.097 | +| 5ms | 2.54ms | 5.03ms | 5.08ms | 51 | 197 | 0.023 | +| **10ms** | **5.03ms** | **10.0ms** | **10.1ms** | **101** | **99** | **0.013** | +| 100ms | 50.0ms | 99.0ms | 99.9ms | 1000 | 10 | 0.004 | +| 1s (old default) | 50.0ms | 99.0ms | 99.9ms | 1000 | 10 | 0.004 | + +### 50,000 items/s + +| Window | p50 | p99 | mean batch | calls/s | +| --- | --- | --- | --- | --- | +| 1ms | 516µs | 1.02ms | 51 | 981 | +| 5ms | 2.52ms | 4.99ms | 251 | 199 | +| **10ms** | **5.04ms** | **9.95ms** | **500** | **100** | +| 100ms | 10.0ms | 19.8ms | 1000 | 50 | +| 1s (old default) | 10.0ms | 19.8ms | 1000 | 50 | + +Two things stand out. + +**The 1s default is already ineffective above ~10k/s.** At 10k/s and 50k/s, 1s and +100ms produce *identical* results, because `BatchSize=1000` fills long before the +timer fires. The old default's only real effect was on sparse traffic — where it +did the most latency damage and bought the least. + +**Latency tracks the window almost exactly.** p99 ≈ the window, plus scheduling +overhead. That is the expected consequence of a partial-batch-age timer and confirms +the sizing rule below. + +## Evidence 2: the sizing rule holds + +Expected batch size ≈ `arrival rate × window`. Measured at 10k/s with a 10ms window: +**predicted 100, measured 100** (100 batches, 99 calls/s). Asserted in +`TestDefaultWindowCoalescingHoldsAtRate`, because the tuning guidance is only +trustworthy if this relationship is real. + +## Evidence 3: a smaller window does not worsen overload + +Offered 200,000/s against a deliberately saturated 2ms processor: + +| Window | accepted/s | completed/s | peak queued | peak heap | calls/s | mean batch | +| --- | --- | --- | --- | --- | --- | --- | +| 1ms | 199,999 | 197,972 | 1,018 | 12MB | 495 | 400 | +| 5ms | 199,997 | 197,130 | 1,415 | 12MB | 199 | 990 | +| **10ms** | **199,998** | **199,004** | **1,405** | **12MB** | **199** | **1000** | +| 100ms | 199,997 | 198,987 | 1,408 | 12MB | 199 | 1000 | +| 1s | 199,998 | 198,997 | 1,406 | 11MB | 199 | 1000 | + +At 10ms and above the behaviour is indistinguishable: under saturation `BatchSize` +is reached before the timer matters, so queue depth, heap, and downstream call rate +are the same. This is the check that prevented the decision resting on latency alone +— a smaller default would have been wrong if it traded latency for stability. + +Only 1ms differs, and it is *worse* on downstream load (495 calls/s versus 199) +because it flushes before batches fill. That is one reason the default is 10ms rather +than the lowest measured value. + +## Evidence 4: bursts + +50,000/s for 100ms, idle 100ms, five cycles — the shape where a small window looks +worst on paper, since each burst coalesces from a standing start: + +| Window | p50 | p99 | mean batch | calls/s | +| --- | --- | --- | --- | --- | +| 1ms | 518µs | 1.02ms | 51 | 545 | +| 5ms | 2.57ms | 5.02ms | 250 | 111 | +| **10ms** | **5.09ms** | **9.99ms** | **500** | **56** | +| 100ms | 10.0ms | 19.8ms | 1000 | 28 | +| 1s | 10.0ms | 19.8ms | 1000 | 28 | + +10ms halves p99 against 100ms while only doubling downstream calls — not the 10x a +timer-only model would predict, because batches still fill during the burst. + +## Why 10ms and not 5ms or 1ms + +| Candidate | Rejected because | +| --- | --- | +| **1ms** | Increases downstream calls under saturation (495/s versus 199/s) and coalesces only 2 items at 1k/s. It optimises latency by nearly removing batching. | +| **2ms** | At 1k/s, 3 items per batch and 373 calls/s. Too little coalescing for a default. | +| **5ms** | Defensible, and better latency. But at 1k/s it yields 6 items per batch versus 11 at 10ms, and p99.9 measured *worse* than 10ms at that rate (13.1ms versus 21.7ms is within noise at 1k/s, but the coalescing gap is not). 10ms is the more conservative default. | +| **20-100ms** | Better coalescing at low rates, but reintroduces the per-hop latency this work exists to remove, and provides no additional protection above 10ms under saturation. | + +10ms is the point where latency is bounded at roughly the window while coalescing +remains meaningful at every rate tested (11x at 1k/s, 101x at 10k/s, 500x at 50k/s). + +## Where 10ms is the wrong choice + +**A service at ~1k items/s with an expensive downstream.** 10ms gives ~11x +coalescing and ~94 downstream calls/s. If a downstream call is costly enough that 94 +calls/s is unacceptable, configure a larger interval deliberately: + +```go +b := batcher.New( + batcher.WithProcessor(fn), + batcher.WithBatchInterval[Item](100 * time.Millisecond), // ~100x coalescing at 1k/s +) +``` + +The default cannot know your downstream cost. What it can avoid is silently adding +up to a second of latency per hop, which the old default did. + +**A service that needs overload protection.** No interval provides it. Under +saturation every window from 10ms to 1s accepted ~200k/s while completing ~199k/s, +with the queue absorbing the difference. Protection requires `WithMaxQueueSize` and +`Enqueue`, or upstream flow control. + +**A service with a slow processor and `Concurrency=1`.** The effective interval is +`max(BatchInterval, processor duration)`. With a 50ms processor, a 10ms window +behaves like a 50ms one — and measured p50 *worse* than a 100ms window (120ms versus +100ms), because queueing dominates. Raise `WithConcurrency` (with +`WithoutOrderedProcessing`) before lowering the interval. + +## Migration + +This is a **behaviour change for every caller who did not set `WithBatchInterval`**. + +| Situation | Effect | Action | +| --- | --- | --- | +| Sets `WithBatchInterval` explicitly | None | None | +| Relies on the 1s default, traffic ≥ 10k/s | None measurable — `BatchSize` was already the binding constraint | None | +| Relies on the 1s default, sparse traffic | Latency drops sharply; downstream call rate rises | Verify the new call rate is acceptable; set a larger interval if not | +| Depends on ~1000-item batches at low traffic | Batches become much smaller | Set `WithBatchInterval` explicitly, or raise `BatchSize` | + +`Stats().BatchesFlushed` with `Completed` gives mean batch size, which is the +fastest way to confirm what the change did to your own coalescing: + +```go +s := b.Stats() +meanBatch := float64(s.Completed) / float64(s.BatchesFlushed) +``` + +## What would reverse this decision + +- Measurement on the CI reference runner (`ubuntu-latest`/amd64) showing a materially + different latency/coalescing relationship. The stored baseline is currently + darwin/arm64 only. +- A reported workload where 10ms coalescing is insufficient *and* the sizing rule + fails to predict it, which would mean the guidance is wrong rather than the default. diff --git a/docs/improvements/plan-perf.md b/docs/improvements/plan-perf.md index 000ed2d..24d808e 100644 --- a/docs/improvements/plan-perf.md +++ b/docs/improvements/plan-perf.md @@ -1206,6 +1206,9 @@ valid stopping point: are 2 with none leaked. - Stopping after **3.x**: small windows are honoured under slow processors. - Stopping after **4.x**: operators can see overload developing. +- Stopping after **5.x**: the compatibility story is published, the tuning guidance + is measured rather than asserted, and `DefaultBatchInterval` is 10ms on the + evidence recorded in `default-window.md`. ## Conclusion diff --git a/pkg/batcher/constants.go b/pkg/batcher/constants.go index f898b99..4281dcc 100644 --- a/pkg/batcher/constants.go +++ b/pkg/batcher/constants.go @@ -6,8 +6,18 @@ const ( // DefaultBatchSize is the default batch size. DefaultBatchSize = 1000 - // DefaultBatchInterval is the default batch interval. - DefaultBatchInterval = 1 * time.Second + // DefaultBatchInterval is the maximum age of a partial batch. + // + // 10ms is the evidence-backed latency/coalescing default. The decision matrix + // measured a 1s default adding up to ~1s p99 latency at 1k items/s, while 10ms + // reduced it to ~22ms and still coalesced ~11 items per batch. At 10k items/s it + // coalesced ~101 items per batch while reducing p99 ~100ms to ~10ms. See + // docs/improvements/default-window.md for the complete matrix and caveats. + // + // This is a maximum partial-batch age, not overload protection. Services that + // need larger batches at low traffic must set WithBatchInterval deliberately; + // services that need process protection must use WithMaxQueueSize and Enqueue. + DefaultBatchInterval = 10 * time.Millisecond // DefaultConcurrency is how many batches are processed at once by default. // diff --git a/pkg/batcher/options.go b/pkg/batcher/options.go index df975ff..995cfdd 100644 --- a/pkg/batcher/options.go +++ b/pkg/batcher/options.go @@ -27,7 +27,9 @@ func WithProcessor[T any](fn Processor[T]) Option[T] { } } -// WithBatchSize sets the batch size. +// WithBatchSize sets how many items a batch holds before it is flushed immediately. +// +// A non-positive value falls back to DefaultBatchSize. func WithBatchSize[T any](batchSize int) Option[T] { return func(b *Batcher[T]) { if b.configFrozen.Load() { @@ -35,14 +37,20 @@ func WithBatchSize[T any](batchSize int) Option[T] { } if batchSize <= 0 { - batchSize = 1000 + batchSize = DefaultBatchSize } b.config.BatchSize = batchSize } } -// WithBatchInterval sets the batch interval. +// WithBatchInterval sets the maximum age of a partial batch. +// +// The timer starts when the first item enters an empty batch, so this is a per-item +// worst-case wait for sparse traffic rather than a periodic flush tick. Expected +// batch size is approximately arrival rate x interval. +// +// A non-positive value falls back to DefaultBatchInterval. func WithBatchInterval[T any](batchInterval time.Duration) Option[T] { return func(b *Batcher[T]) { if b.configFrozen.Load() { @@ -50,7 +58,7 @@ func WithBatchInterval[T any](batchInterval time.Duration) Option[T] { } if batchInterval <= 0 { - batchInterval = 1 * time.Second + batchInterval = DefaultBatchInterval } b.config.BatchInterval = batchInterval diff --git a/test/scenario/decision_test.go b/test/scenario/decision_test.go new file mode 100644 index 0000000..3653c09 --- /dev/null +++ b/test/scenario/decision_test.go @@ -0,0 +1,249 @@ +package scenario_test + +import ( + "os" + "runtime" + "strconv" + "testing" + "time" + + "github.com/NSXBet/batcher/test/scenario" + "github.com/stretchr/testify/require" +) + +// TestDefaultWindowDecisionMatrix produces the evidence for Milestone 5.3: whether +// DefaultBatchInterval should move from 1s toward the 5-10ms range. +// +// It reports the four things the decision needs, per window: +// +// - latency the caller experiences (p50/p99/p99.9/max); +// - coalescing actually achieved (mean batch size, downstream calls/s); +// - allocation cost per item; +// - whether the measurement is even valid (schedule lateness). +// +// A window is only worth recommending if it improves latency AND still coalesces at +// the arrival rates a shared service sees. A window that reduces latency by turning +// every batch into a single item has not helped: it has just removed batching. +// +// Report-only. Latency on a shared machine is not a pass/fail gate, and this exists +// to inform a decision rather than to police one. +func TestDefaultWindowDecisionMatrix(t *testing.T) { + if os.Getenv("SCENARIO_MATRIX") == "" { + t.Skip("set SCENARIO_MATRIX=1 to run the decision matrix") + } + + windows := []time.Duration{ + time.Millisecond, + 2 * time.Millisecond, + 5 * time.Millisecond, + 10 * time.Millisecond, + 20 * time.Millisecond, + 50 * time.Millisecond, + 100 * time.Millisecond, + time.Second, // the current default + } + + // Rates chosen to bracket the interesting region: below, at, and well above the + // point where a small window still fills a useful batch. + rates := []int{1_000, 10_000, 50_000} + + const duration = 2 * time.Second + + t.Logf("environment: %s", scenario.CurrentEnvironment()) + t.Logf("processor: no-op (isolates batching cost from downstream cost)") + t.Logf("") + t.Logf("%-8s %-8s %-10s %-10s %-10s %-9s %-10s %-9s %s", + "rate", "window", "p50", "p99", "p99.9", "batch", "calls/s", "alloc/item", "valid") + + for _, rate := range rates { + for _, window := range windows { + result := scenario.Run(scenario.Config{ + Name: "decision", + BatchSize: 1_000, + BatchInterval: window, + Arrival: scenario.Steady(rate, duration), + Processor: scenario.NoOpProcessor(), + Producers: runtime.NumCPU(), + Warmup: 200 * time.Millisecond, + Seed: 1, + LatenessBudget: 5 * time.Millisecond, + }) + + valid := "yes" + if !result.LatenessValid { + valid = "LATE" + } + + if result.TimedOut { + valid = "TIMEOUT" + } + + t.Logf("%-8d %-8s %-10s %-10s %-10s %-9.0f %-10.0f %-9.3f %s", + rate, window, + result.EndToEnd.P50.Round(time.Microsecond), + result.EndToEnd.P99.Round(time.Microsecond), + result.EndToEnd.P999.Round(time.Microsecond), + result.MeanBatchSize, + result.DownstreamPerSec, + result.AllocsPerItem, + valid) + } + + t.Logf("") + } +} + +// TestDefaultWindowUnderSustainedOverload records what each window does when the +// downstream cannot keep up. +// +// This is the check that stops the decision being made on latency alone. A smaller +// window must not make overload worse: if it increased queue depth or downstream +// call rate under saturation, it would trade a latency win for a stability loss. +func TestDefaultWindowUnderSustainedOverload(t *testing.T) { + if os.Getenv("SCENARIO_MATRIX") == "" { + t.Skip("set SCENARIO_MATRIX=1 to run the decision matrix") + } + + const ( + // A 2ms processor retires far less than this, so every window is saturated. + rate = 200_000 + serviceTime = 2 * time.Millisecond + duration = 500 * time.Millisecond + ) + + t.Logf("environment: %s", scenario.CurrentEnvironment()) + t.Logf("offered=%d/s processor=%s (deliberately saturated)", rate, serviceTime) + t.Logf("") + t.Logf("%-8s %-12s %-12s %-12s %-11s %-10s %s", + "window", "accepted/s", "completed/s", "pending_peak", "heap_peak", "calls/s", "batch") + + for _, window := range []time.Duration{ + time.Millisecond, + 5 * time.Millisecond, + 10 * time.Millisecond, + 100 * time.Millisecond, + time.Second, + } { + result := scenario.Run(scenario.Config{ + Name: "overload", + BatchSize: 1_000, + BatchInterval: window, + Arrival: scenario.Steady(rate, duration), + Processor: scenario.FixedProcessor(serviceTime), + Producers: runtime.NumCPU(), + LatenessBudget: time.Second, + }) + + t.Logf("%-8s %-12.0f %-12.0f %-12d %-11s %-10.0f %.0f", + window, + result.AcceptedRate, + result.CompletedRate, + result.PendingPeak, + formatMB(result.HeapHighWater), + result.DownstreamPerSec, + result.MeanBatchSize) + } +} + +// TestDefaultWindowUnderBursts records burst absorption and recovery, which a steady +// arrival rate cannot show. +// +// A service that is idle between spikes is exactly where a small window looks worst +// on paper: each burst has to be coalesced from a standing start. +func TestDefaultWindowUnderBursts(t *testing.T) { + if os.Getenv("SCENARIO_MATRIX") == "" { + t.Skip("set SCENARIO_MATRIX=1 to run the decision matrix") + } + + t.Logf("environment: %s", scenario.CurrentEnvironment()) + t.Logf("pattern: 50k/s for 100ms, idle 100ms, x5") + t.Logf("") + t.Logf("%-8s %-10s %-10s %-9s %-10s %s", + "window", "p50", "p99", "batch", "calls/s", "valid") + + for _, window := range []time.Duration{ + time.Millisecond, + 5 * time.Millisecond, + 10 * time.Millisecond, + 100 * time.Millisecond, + time.Second, + } { + result := scenario.Run(scenario.Config{ + Name: "burst", + BatchSize: 1_000, + BatchInterval: window, + Arrival: scenario.Burst(50_000, 100*time.Millisecond, 100*time.Millisecond, 5), + Processor: scenario.NoOpProcessor(), + Producers: runtime.NumCPU(), + LatenessBudget: 5 * time.Millisecond, + }) + + valid := "yes" + if !result.LatenessValid { + valid = "LATE" + } + + t.Logf("%-8s %-10s %-10s %-9.0f %-10.0f %s", + window, + result.EndToEnd.P50.Round(time.Microsecond), + result.EndToEnd.P99.Round(time.Microsecond), + result.MeanBatchSize, + result.DownstreamPerSec, + valid) + } +} + +// TestDefaultWindowCoalescingHoldsAtRate is the one assertion in this file. +// +// The plan's sizing rule is that expected batch size is approximately +// arrival rate x window. If that relationship does not hold, the recommendation in +// the tuning guide is wrong and the decision record should not be trusted. +// +// Asserted loosely: the point is that the rule predicts the right order of +// magnitude, not that scheduling is exact on a shared machine. +func TestDefaultWindowCoalescingHoldsAtRate(t *testing.T) { + t.Parallel() + + if testing.Short() { + t.Skip("timing-sensitive; runs in the full suite only") + } + + const ( + rate = 10_000 + window = 10 * time.Millisecond + duration = time.Second + ) + + result := scenario.Run(scenario.Config{ + Name: "coalescing-rule", + BatchSize: 10_000, // never size-triggered: isolate the timer + BatchInterval: window, + Arrival: scenario.Steady(rate, duration), + Processor: scenario.NoOpProcessor(), + Producers: runtime.NumCPU(), + LatenessBudget: 5 * time.Millisecond, + }) + + // rate x window = 10_000/s x 10ms = 100 items. + predicted := float64(rate) * window.Seconds() + + t.Logf("predicted mean batch %.0f, measured %.0f (batches=%d, calls/s=%.0f)", + predicted, result.MeanBatchSize, result.Batches, result.DownstreamPerSec) + + require.False(t, result.TimedOut, "the run must complete") + + // Half to double the prediction. Scheduling jitter on a shared machine moves + // this around; an order-of-magnitude miss would mean the rule is wrong. + require.Greater(t, result.MeanBatchSize, predicted*0.5, + "measured batch size far below rate x window: the sizing rule in the tuning "+ + "guide would be misleading") + require.Less(t, result.MeanBatchSize, predicted*2.0, + "measured batch size far above rate x window: the sizing rule in the tuning "+ + "guide would be misleading") +} + +func formatMB(bytes uint64) string { + const mb = 1 << 20 + + return strconv.FormatUint(bytes/mb, 10) + "MB" +} From af338736ec10d6a2c501e96d3034312d82b6b063 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Thu, 6 Aug 2026 23:10:31 -0300 Subject: [PATCH 3/8] fix: enforce Fx injection and correct Phase 5 guidance Ten review findings on Phase 5. The substantive bug: ProvideBatcherInFXWithOptions documented that callers must not pass WithProcessor, but applied the injected processor first and caller options afterwards. A caller-supplied WithProcessor therefore silently replaced the injected factory result. Reproduced: injected=0, rogue=1. The injected processor is now applied last, so dependency injection wins (injected=1, rogue=0), and a regression test passes a competing processor and asserts it never runs. The interval documentation now states the actual steady-state formula: min(BatchSize, arrival rate x interval). The old rate-times-interval form overstates at high rates because size flushing happens first; at the default size of 1000, 50k/s fills in about 20ms, so a 100ms interval does not create a 5000-item batch. The README and WithBatchInterval godoc both say this now. The compatibility guide claimed every public behaviour change was listed while omitting the largest one: DefaultBatchInterval 1s -> 10ms. It now has a complete before/after/migration section, including callers with positive explicit intervals, callers relying on zero/negative fallback, sparse traffic call-rate changes, and the fact that neither a small nor large interval is overload protection. It also fixes the Config wording and correctly documents per-call rather than cached Shutdown results. Measurement correctness fixes: - scenario concurrency runs use runtime.NumCPU producers at 10k/s, so a single producer's timer precision does not collapse offered load toward service rate; - report sweeps reject timed-out runs instead of including invalid output; - procRNG is mutex-protected around the RNG read only, not the simulated sleep. This fixes a real race from JitteredProcessor/SlowOutlierProcessor combined with WithConcurrency, reproduced in math/rand.(*rngSource).Uint64; - allocation evidence now rejects TimedOut before deriving mean batch size or waste; - the +2% capacity rule is correctly labelled measured evidence, not CI enforcement. Finally, worker tests release blocked processors via t.Cleanup and sync.Once, so an assertion failure cannot turn into a package timeout, and all auto-started batchers in batcher_test.go are closed so they cannot perturb goroutine-budget baselines. All suites pass under -race, vet, gofmt, and actionlint. --- README.md | 17 ++++++---- docs/improvements/compatibility.md | 52 +++++++++++++++++++++++++++-- docs/improvements/default-window.md | 18 +++++++--- pkg/batcher/fx.go | 15 +++++---- pkg/batcher/fx_options_test.go | 47 ++++++++++++++++++++++++++ pkg/batcher/options.go | 8 +++-- 6 files changed, 135 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 7e2bd70..a8b95d5 100644 --- a/README.md +++ b/README.md @@ -363,17 +363,22 @@ the timer starts when the first item enters an empty batch. Sparse traffic there waits the full interval per item, and a request crossing several batching services pays up to one interval per hop. -Expected batch size is approximately **arrival rate x interval**. That rule is -measured, not estimated: at 10,000 items/s with a 10ms interval it predicts 100 items -per batch, and the measured mean is 100. +Expected batch size is approximately **min(arrival rate x interval, BatchSize)**. +The arrival-rate term is measured, not estimated: at 10,000 items/s with a 10ms +interval it predicts 100 items per batch, and the measured mean is 100. -Use it to pick an interval from the coalescing you actually need: +`BatchSize` is the binding constraint whenever it fills first, and the interval then +never fires. With the default `BatchSize` of 1,000, a service at 50,000 items/s +reaches 1,000 items after about 20ms, so configuring a 100ms interval changes nothing +— which is exactly what the 50,000/s row below shows. + +Use the formula to pick an interval from the coalescing you actually need: | Arrival rate | 10ms (default) | 100ms | | --- | --- | --- | | 1,000/s | ~11 items/batch, ~94 calls/s, p99 ~12ms | ~100 items/batch, ~10 calls/s, p99 ~100ms | | 10,000/s | ~101 items/batch, ~99 calls/s, p99 ~10ms | ~1000 items/batch, ~10 calls/s, p99 ~99ms | -| 50,000/s | ~500 items/batch, ~100 calls/s, p99 ~10ms | ~1000 items/batch, ~50 calls/s, p99 ~20ms | +| 50,000/s | ~500 items/batch, ~100 calls/s, p99 ~10ms | ~1000 items/batch (capped by `BatchSize`), ~50 calls/s, p99 ~20ms | Raise the interval when your downstream is expensive enough that the call rate at 10ms is unacceptable — most relevant below a few thousand items/s. The full matrix @@ -397,7 +402,7 @@ items/s: The smaller interval is *worse*, because queueing dominates. If your processor is slow, raise `WithConcurrency` (with `WithoutOrderedProcessing`) rather than lowering -the interval — with 8 workers the same scenario measured p50 4ms. +the interval. With 8 workers, the same scenario measured a p50 latency of 4ms. ## FX Integration diff --git a/docs/improvements/compatibility.md b/docs/improvements/compatibility.md index 41a8eb0..d9d5482 100644 --- a/docs/improvements/compatibility.md +++ b/docs/improvements/compatibility.md @@ -50,7 +50,7 @@ rather than synchronised. cfg := b.Config() size := cfg.BatchSize -// After: identical to read, since Go auto-dereferences neither is needed. +// After: Config() returns a value; read its fields directly, no pointer needed. size := b.Config().BatchSize // Before: mutate a running batcher (a data race, now impossible). @@ -63,6 +63,54 @@ b = batcher.New(batcher.WithBatchSize[Item](500), batcher.WithProcessor(fn)) Code that only *reads* `Config()` compiles unchanged. Code that assigned through it will now fail to compile, which is the intended outcome: it was racing. +### `DefaultBatchInterval` changes 1s → 10ms + +| | | +| --- | --- | +| **Before** | `DefaultBatchInterval = 1 * time.Second` | +| **After** | `DefaultBatchInterval = 10 * time.Millisecond` | + +This affects every caller that did not set `WithBatchInterval`, and it is a +behaviour change rather than a bug fix: batches close sooner, so latency drops and +the downstream call rate rises whenever `BatchSize` is not reached first. + +The decision and its full measurement matrix are in +[`default-window.md`](./default-window.md). In short: at 1,000 items/s the old +default measured p99 981ms; 10ms measured p99 12ms while still coalescing ~11 items +per batch. + +**Who is affected:** + +| Situation | Effect | Action | +| --- | --- | --- | +| Sets a **positive** `WithBatchInterval` | None | None | +| Passes `WithBatchInterval(0)` or a negative value | Falls back to the default, so it moves to 10ms too | Pass an explicit positive interval to keep 1s | +| Relies on the default, traffic ≥ 10k/s | None measurable — `BatchSize` was already binding | None | +| Relies on the default, sparse traffic | Latency drops sharply; more downstream calls | Confirm the new call rate is acceptable | +| Depends on ~1000-item batches at low traffic | Batches become much smaller | Set `WithBatchInterval` explicitly, or raise `BatchSize` | + +**Migration:** + +```go +// Keep the previous behaviour explicitly. +b := batcher.New( + batcher.WithProcessor(fn), + batcher.WithBatchInterval[Item](time.Second), +) +``` + +Note that shrinking the interval is **not** overload protection, and neither is +keeping it large: under a saturated downstream every interval from 10ms to 1s +accepted the same load. If the concern is memory rather than latency, bound the +queue instead: + +```go +b := batcher.New( + batcher.WithProcessor(fn), + batcher.WithMaxQueueSize[Item](10_000), // back-pressure, reported via Enqueue +) +``` + ### `DefaultConcurrency` changes 3 → 1, and `Concurrency` now works | | | @@ -183,7 +231,7 @@ Fx stop-hook behaviour: | Normal drain | `nil` | | Stop context expires | `*ShutdownIncompleteError` as an Fx lifecycle error; drain continues | | Batcher never started but holds queued work | Drains; `Shutdown` starts a consumer when one is needed | -| Repeated stop | Idempotent; later calls observe the same terminal result | +| Repeated stop | Idempotent: later calls wait on the same drain and return their own wait result, so a caller that waits long enough gets `nil` even if an earlier caller timed out | ## Clarified contracts (no code change required) diff --git a/docs/improvements/default-window.md b/docs/improvements/default-window.md index 0d8a6b7..cd7d5be 100644 --- a/docs/improvements/default-window.md +++ b/docs/improvements/default-window.md @@ -22,7 +22,7 @@ downstream work happened. ## Environment -``` +```text go=go1.26.5 os=darwin arch=arm64 gomaxprocs=12 cpus=12 ``` @@ -173,17 +173,25 @@ This is a **behaviour change for every caller who did not set `WithBatchInterval | Situation | Effect | Action | | --- | --- | --- | -| Sets `WithBatchInterval` explicitly | None | None | +| Sets `WithBatchInterval` to a **positive** value | None | None | +| Passes `WithBatchInterval(0)` or a negative duration | Falls back to `DefaultBatchInterval`, so it moves 1s → 10ms with everyone else | Pass an explicit positive interval if you relied on the old 1s fallback | | Relies on the 1s default, traffic ≥ 10k/s | None measurable — `BatchSize` was already the binding constraint | None | | Relies on the 1s default, sparse traffic | Latency drops sharply; downstream call rate rises | Verify the new call rate is acceptable; set a larger interval if not | | Depends on ~1000-item batches at low traffic | Batches become much smaller | Set `WithBatchInterval` explicitly, or raise `BatchSize` | -`Stats().BatchesFlushed` with `Completed` gives mean batch size, which is the -fastest way to confirm what the change did to your own coalescing: +`Stats().BatchesFlushed` with the terminal counters gives mean batch size after a +drain, which is the fastest way to confirm what the change did to your own +coalescing: ```go s := b.Stats() -meanBatch := float64(s.Completed) / float64(s.BatchesFlushed) + +// Include every terminal outcome: a failed or panicked batch was still flushed, so +// dividing by Completed alone undercounts whenever the processor errors. +if s.BatchesFlushed > 0 { + meanBatch := float64(s.Completed+s.Failed+s.Panicked) / float64(s.BatchesFlushed) + log.Printf("mean batch size: %.1f", meanBatch) +} ``` ## What would reverse this decision diff --git a/pkg/batcher/fx.go b/pkg/batcher/fx.go index fd40ac0..89c31a9 100644 --- a/pkg/batcher/fx.go +++ b/pkg/batcher/fx.go @@ -38,9 +38,9 @@ func ProvideBatcherInFX[T any]( // ProvideBatcherInFXWithOptions wires a Batcher into an Fx application with the // full option set. // -// The processor comes from the injected factory, so callers must not pass -// WithProcessor: it is applied first and then overridden by anything supplied -// here, which would silently bypass dependency injection. Every other option +// The processor always comes from the injected factory. A caller-supplied +// WithProcessor is ignored rather than honoured: the injected processor is applied +// last, so dependency injection cannot be bypassed by accident. Every other option // behaves exactly as it does with New. // // WithSkipAutoStart is always applied last and cannot be overridden, because Fx @@ -53,10 +53,11 @@ func ProvideBatcherInFXWithOptions[T any]( return provideBatcherModule[T]( processorFactory, func(processorFunc Processor[T]) []Option[T] { - // The injected processor goes first so an explicit WithProcessor in - // options would win. That is a caller mistake rather than something to - // silently correct, and it is documented above. - return append([]Option[T]{WithProcessor(processorFunc)}, options...) + // The injected processor goes LAST so it cannot be overridden. Placing it + // first meant a caller-supplied WithProcessor silently replaced it and + // bypassed dependency injection entirely — the documented restriction was + // not enforced, so the failure was invisible rather than loud. + return append(append([]Option[T]{}, options...), WithProcessor(processorFunc)) }, ) } diff --git a/pkg/batcher/fx_options_test.go b/pkg/batcher/fx_options_test.go index 1e38d5d..d3ac747 100644 --- a/pkg/batcher/fx_options_test.go +++ b/pkg/batcher/fx_options_test.go @@ -309,3 +309,50 @@ func TestConfigSnapshotCannotMutateRunningBatcher(t *testing.T) { require.Equal(t, int64(400), processed.Load(), "every item must still be processed with the original configuration") } + +// TestProvideBatcherInFXWithOptionsIgnoresCallerProcessor pins that dependency +// injection cannot be bypassed. +// +// The injected processor is applied last, so a caller-supplied WithProcessor is +// ignored. Before this, options were applied after the injected processor and a +// caller-supplied one silently replaced it: the documented restriction existed but +// nothing enforced it, so the mistake produced a working app wired to the wrong +// processor rather than an error. +func TestProvideBatcherInFXWithOptionsIgnoresCallerProcessor(t *testing.T) { + t.Parallel() + + var ( + b *batcher.Batcher[string] + processor *optionsProcessor + rogue atomic.Int64 + ) + + app := fxtest.New(t, + fx.Provide(newOptionsProcessor), + batcher.ProvideBatcherInFXWithOptions[string]( + optionsProcessorFunc, + batcher.WithBatchSize[string](1), + batcher.WithBatchInterval[string](time.Millisecond), + // Deliberately competing with the injected processor. + batcher.WithProcessor(batcher.Processor[string](func(items []string) error { + rogue.Add(int64(len(items))) + + return nil + })), + ), + fx.Populate(&b, &processor), + ) + + app.RequireStart() + + b.Add("routed-by-injection") + + require.NoError(t, b.Join(10*time.Second)) + + app.RequireStop() + + require.Equal(t, int64(1), processor.items.Load(), + "the injected processor must receive the item") + require.Zero(t, rogue.Load(), + "a caller-supplied WithProcessor must not override dependency injection") +} diff --git a/pkg/batcher/options.go b/pkg/batcher/options.go index 995cfdd..7e088c9 100644 --- a/pkg/batcher/options.go +++ b/pkg/batcher/options.go @@ -47,8 +47,12 @@ func WithBatchSize[T any](batchSize int) Option[T] { // WithBatchInterval sets the maximum age of a partial batch. // // The timer starts when the first item enters an empty batch, so this is a per-item -// worst-case wait for sparse traffic rather than a periodic flush tick. Expected -// batch size is approximately arrival rate x interval. +// worst-case wait for sparse traffic rather than a periodic flush tick. +// +// Under steady traffic, expected batch size is approximately +// min(BatchSize, arrival rate x interval): the interval bounds how long a partial +// batch waits, but BatchSize can flush it sooner. At high arrival rates the size +// limit is the binding constraint and the interval never fires. // // A non-positive value falls back to DefaultBatchInterval. func WithBatchInterval[T any](batchInterval time.Duration) Option[T] { From 1df01a26733a7e2f8eb48fd9f0a92cc79430dab2 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Fri, 7 Aug 2026 00:11:05 -0300 Subject: [PATCH 4/8] test: pin lifecycle stability and queue edge-case correctness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the tests that were missing rather than the ones that were easy. Each covers behaviour that is currently correct but pinned by nothing, so a refactor could break it silently. Stability, all written as repeated trials because every one is an ordering race and a single pass proves little: - Blocking Add released by shutdown. Enqueue's parked case was already covered, but Add is the compatibility fast path with no error return, so if shutdown failed to release it the caller would hang with no way to observe why. Reachable only in bounded mode, which is why it was missed. - Add racing Close, 200 trials, asserting every accepted item reaches exactly one terminal outcome. This is the interleaving a request handler produces accidentally during process shutdown. - Panic in the final shutdown batch with a concurrent Errors() consumer, 150 trials. This is the narrowest window in the protocol: a diagnostic published while the coordinator closes the diagnostics channel. Get the single-closer rule wrong and it panics inside the library's own goroutine, where the caller cannot recover it. - Eight concurrent Shutdown callers, asserting one drain, no deadlock, and no false error for work that completed. - A processor that always errors with a 2-entry diagnostics buffer, proving a failed batch still releases its drain obligation. If it did not, Pending would never reach zero and shutdown would wait forever on work that already ran. - Start racing Shutdown from five goroutines, asserting exactly one lifecycle. Correctness, covering paths only reachable at a boundary: - tryPush rejection at exactly capacity, and that the rejection is transient. - Unbounded mode never rejecting, since that is the documented default. - Negative capacity normalising to unbounded rather than rejecting everything. - The full-drain cursor reset, distinct from the partial-drain compaction test. - push honouring a cancelled context, and being released by seal. - Capacity clamping when BatchSize is below the floor, which a caller reaches with WithBatchSize(1), and non-positive BatchSize not yielding a zero-capacity slice. - clampCapacity as a direct table, because it is the retention bound itself. - ShutdownIncompleteError.Error(), which lifecycle tools surface even though callers inspect it with errors.As. - Every option individually respecting the freeze, so adding an option without the guard fails a test rather than surfacing later as a data race. Three of these are sabotage-verified: removing the queue capacity bound, the seal release, or Add's blocking path makes the corresponding test fail or hang. The thresholds doc now records that verification as a table, and states the reasoning — a concurrency test that cannot fail manufactures confidence. Coverage 93.0% -> 96.2%, with queue.go push/tryPush/newQueue and clampCapacity now fully covered. --- docs/improvements/thresholds.md | 27 +++ pkg/batcher/options_freeze_test.go | 56 +++++ pkg/batcher/queue_edge_test.go | 210 +++++++++++++++++ pkg/batcher/shutdown_test.go | 18 ++ pkg/batcher/stability_test.go | 346 +++++++++++++++++++++++++++++ 5 files changed, 657 insertions(+) create mode 100644 pkg/batcher/queue_edge_test.go create mode 100644 pkg/batcher/stability_test.go diff --git a/docs/improvements/thresholds.md b/docs/improvements/thresholds.md index 551d207..8e7d41e 100644 --- a/docs/improvements/thresholds.md +++ b/docs/improvements/thresholds.md @@ -151,6 +151,33 @@ bimodal workloads, not just the sparse case the optimisation targets. An EWMA-of-mean estimator was rejected precisely because it regressed alternating traffic to 2.80 MB against 1.64 MB for the current full-capacity strategy. +## Coverage + +`pkg/batcher` statement coverage is enforced at 80% by CI and currently sits at +**96.2%**. The scenario harness under `test/` is excluded, since it is measurement +infrastructure rather than shipped code. + +The number matters less than what is covered. Every lifecycle interleaving that could +hang, panic, or lose accepted work has a dedicated test, and the ones guarding a race +or deadlock are sabotage-verified: the mechanism is removed and the test must fail. +Verified this way so far: + +| Mechanism | Sabotage result | +| --- | --- | +| Coordinator post-seal gate check | `Shutdown` hangs | +| `enter()` rejection routed through `leave()` | quiescence never signalled | +| Per-batch panic recovery | process crashes | +| `BatchHeld` ownership transfer | field never populated | +| Dispatch accounting | flush and terminal counts wrong | +| Option freeze guard | data race under `-race` | +| Queue capacity bound | `tryPush` grows past the bound | +| Queue seal release | blocked publisher never released | +| Blocking `Add` release on shutdown | caller hangs indefinitely | + +A concurrency test that cannot fail is worse than no test, because it manufactures +confidence. If a future change adds one of these guarantees, add the sabotage check +with it. + ## Baselines Stored benchmark baselines live in `docs/improvements/baselines/`. diff --git a/pkg/batcher/options_freeze_test.go b/pkg/batcher/options_freeze_test.go index 6654f30..b8da595 100644 --- a/pkg/batcher/options_freeze_test.go +++ b/pkg/batcher/options_freeze_test.go @@ -88,3 +88,59 @@ func TestOptionsAreFrozenAfterNew(t *testing.T) { require.Zero(t, mutatedCalls.Load(), "a post-start WithProcessor must not replace the original") } + +// TestEveryOptionRespectsTheFreeze covers the frozen-config early return in each +// option individually. +// +// TestOptionsAreFrozenAfterNew proves the guard works under concurrent load, but it +// exercises only the options it applies. This table walks every option, so adding a +// new one without the guard shows up as a coverage and assertion gap rather than as +// a data race discovered later. +func TestEveryOptionRespectsTheFreeze(t *testing.T) { + t.Parallel() + + b := batcher.New( + batcher.WithBatchSize[test.BatchItem](64), + batcher.WithBatchInterval[test.BatchItem](7*time.Millisecond), + batcher.WithMaxQueueSize[test.BatchItem](128), + batcher.WithCloseGrace[test.BatchItem](11*time.Second), + batcher.WithErrorBufferSize[test.BatchItem](32), + batcher.WithProcessor(batcher.NoOpProcessor[test.BatchItem]), + ) + + defer func() { require.NoError(t, b.Close()) }() + + before := b.Config() + + // Every option, applied post-construction, must be inert. + for _, option := range []batcher.Option[test.BatchItem]{ + batcher.WithBatchSize[test.BatchItem](1), + batcher.WithBatchInterval[test.BatchItem](time.Hour), + batcher.WithMaxQueueSize[test.BatchItem](1), + batcher.WithCloseGrace[test.BatchItem](time.Nanosecond), + batcher.WithErrorBufferSize[test.BatchItem](1), + batcher.WithConcurrency[test.BatchItem](16), + batcher.WithoutOrderedProcessing[test.BatchItem](), + batcher.WithSkipAutoStart[test.BatchItem](), + batcher.WithProcessor(batcher.NoOpProcessor[test.BatchItem]), + } { + option(b) + } + + after := b.Config() + + // Config contains a function field, which Go cannot compare for equality even + // when it is the same function. Assert every scalar explicitly and prove the + // processor remains callable below. + require.Equal(t, before.SkipAutoStart, after.SkipAutoStart) + require.Equal(t, before.BatchSize, after.BatchSize) + require.Equal(t, before.BatchInterval, after.BatchInterval) + require.Equal(t, before.Concurrency, after.Concurrency) + require.Equal(t, before.MaxQueueSize, after.MaxQueueSize) + require.Equal(t, before.CloseGrace, after.CloseGrace) + require.Equal(t, before.ErrorBufferSize, after.ErrorBufferSize) + require.Equal(t, before.UnorderedProcessingAcknowledged, after.UnorderedProcessingAcknowledged) + require.NotNil(t, after.ProcessorFunc, + "post-construction WithProcessor must not replace the original processor") + require.NoError(t, after.ProcessorFunc(nil)) +} diff --git a/pkg/batcher/queue_edge_test.go b/pkg/batcher/queue_edge_test.go new file mode 100644 index 0000000..472ed37 --- /dev/null +++ b/pkg/batcher/queue_edge_test.go @@ -0,0 +1,210 @@ +package batcher + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" +) + +// Queue edge cases. +// +// These are package-internal because they exercise the queue directly. Driving them +// through the public API would require winning a race to observe the state, so a +// direct test is both deterministic and honest about what it covers. + +// TestQueueRejectsWhenBoundedAndFull covers tryPush's rejection path, which is how a +// non-blocking Add refuses work in bounded mode. +// +// Reached only when the queue is exactly at capacity, so ordinary tests miss it. +func TestQueueRejectsWhenBoundedAndFull(t *testing.T) { + t.Parallel() + + q := newQueue[int](2) + + require.True(t, q.tryPush(1)) + require.True(t, q.tryPush(2)) + require.False(t, q.tryPush(3), + "tryPush must refuse rather than grow past the configured bound") + require.Equal(t, 2, q.length()) + + // Draining one slot must make room again, so a rejection is transient rather + // than permanent. + item, ok := q.pop() + require.True(t, ok) + require.Equal(t, 1, item) + require.True(t, q.tryPush(3), "space must be reusable after a pop") + require.Equal(t, 2, q.length()) +} + +// TestQueueUnboundedNeverRejects pins that the default mode has no capacity limit. +// +// The library's documented default is unbounded, so a spurious rejection here would +// silently drop accepted work in the configuration most callers use. +func TestQueueUnboundedNeverRejects(t *testing.T) { + t.Parallel() + + q := newQueue[int](0) + + for i := range 10_000 { + require.True(t, q.tryPush(i), "unbounded queue rejected at %d", i) + } + + require.Equal(t, 10_000, q.length()) +} + +// TestNewQueueNormalisesNegativeCapacity pins that a negative bound means unbounded +// rather than a queue that rejects everything. +// +// WithMaxQueueSize clamps at the option layer, but the queue is constructed +// elsewhere too, so the invariant belongs here as well. +func TestNewQueueNormalisesNegativeCapacity(t *testing.T) { + t.Parallel() + + q := newQueue[int](-5) + + require.Equal(t, 0, q.capacity, "a negative bound must normalise to unbounded") + require.True(t, q.tryPush(1)) +} + +// TestQueuePopOnEmptyResetsCursor pins the full-drain reset. +// +// Without it the read cursor would keep advancing past a reused backing array. This +// is distinct from the compaction test: that one covers a partially drained queue, +// this one the fully drained case. +func TestQueuePopOnEmptyResetsCursor(t *testing.T) { + t.Parallel() + + q := newQueue[int](0) + sealCh := make(chan struct{}) + + for i := range 4 { + require.NoError(t, q.push(context.Background(), i, sealCh)) + } + + for range 4 { + _, ok := q.pop() + require.True(t, ok) + } + + // The queue is now empty but head is still advanced; popping resets it. + _, ok := q.pop() + require.False(t, ok) + + q.mu.Lock() + head, length := q.head, len(q.items) + q.mu.Unlock() + + require.Zero(t, head, "a fully drained queue must reset its read cursor") + require.Zero(t, length, "a fully drained queue must reset its length") + + // And it must still be usable afterwards. + require.True(t, q.tryPush(99)) + + item, ok := q.pop() + require.True(t, ok) + require.Equal(t, 99, item) +} + +// TestQueuePushRespectsContextWhenFull pins that a blocking push honours the +// caller's deadline instead of parking indefinitely. +func TestQueuePushRespectsContextWhenFull(t *testing.T) { + t.Parallel() + + q := newQueue[int](1) + sealCh := make(chan struct{}) + + require.NoError(t, q.push(context.Background(), 1, sealCh)) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := q.push(ctx, 2, sealCh) + + require.ErrorIs(t, err, context.Canceled, + "a full queue must release a blocked publisher when its context ends") + require.Equal(t, 1, q.length(), "the refused item must not be queued") +} + +// TestQueuePushReleasedBySeal pins the shutdown release path. +// +// This is what stops a publisher parked on a full queue from deadlocking shutdown: +// it can never leave the admission gate until this returns. +func TestQueuePushReleasedBySeal(t *testing.T) { + t.Parallel() + + q := newQueue[int](1) + sealCh := make(chan struct{}) + + require.NoError(t, q.push(context.Background(), 1, sealCh)) + + result := make(chan error, 1) + + go func() { result <- q.push(context.Background(), 2, sealCh) }() + + close(sealCh) + + require.ErrorIs(t, <-result, ErrClosing, + "sealing must release a publisher parked on a full queue") +} + +// TestCapacityEstimatorClampsBelowFloor pins behaviour for a BatchSize smaller than +// the capacity floor. +// +// A caller can legitimately set WithBatchSize(1), and capacity must never exceed +// BatchSize, so the floor cannot win that comparison. +func TestCapacityEstimatorClampsBelowFloor(t *testing.T) { + t.Parallel() + + for _, batchSize := range []int{1, 2, 8} { + est := newCapacityEstimator(batchSize) + + require.LessOrEqual(t, est.capacity(), batchSize, + "batchSize=%d: capacity must never exceed the configured ceiling", batchSize) + + est.observe(batchSize) + + require.LessOrEqual(t, est.capacity(), batchSize, + "batchSize=%d: capacity must stay within the ceiling after observation", + batchSize) + } +} + +// TestCapacityEstimatorRejectsNonPositiveBatchSize pins that a nonsensical ceiling +// degrades to something usable rather than producing a zero-capacity slice forever. +func TestCapacityEstimatorRejectsNonPositiveBatchSize(t *testing.T) { + t.Parallel() + + for _, batchSize := range []int{0, -1, -100} { + est := newCapacityEstimator(batchSize) + + require.Positive(t, est.capacity(), + "batchSize=%d must not yield a zero-capacity estimator", batchSize) + } +} + +// TestClampCapacityDirectly covers every branch of the helper. The estimator tests +// exercise it indirectly, but an indirect test can miss a branch when the estimator +// changes shape; this helper is the retention bound itself and deserves a direct +// table. +func TestClampCapacityDirectly(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + candidate int + batchSize int + want int + }{ + {"above ceiling", 2_000, 1_000, 1_000}, + {"below floor with small ceiling", 1, 8, 8}, + {"below floor with normal ceiling", 1, 1_000, capacityFloor}, + {"inside bounds", 128, 1_000, 128}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, clampCapacity(tc.candidate, tc.batchSize)) + }) + } +} diff --git a/pkg/batcher/shutdown_test.go b/pkg/batcher/shutdown_test.go index 8b744ef..af3c469 100644 --- a/pkg/batcher/shutdown_test.go +++ b/pkg/batcher/shutdown_test.go @@ -400,3 +400,21 @@ func TestNonReturningProcessorKeepsBatcherDraining(t *testing.T) { require.False(t, b.IsClosed(), "a batcher whose processor never returns is draining, not closed") } + +// TestShutdownIncompleteErrorFormatting pins the public error contract: callers +// inspect it with errors.Is/As, but logs and lifecycle tools also surface Error(). +func TestShutdownIncompleteErrorFormatting(t *testing.T) { + t.Parallel() + + err := &batcher.ShutdownIncompleteError{ + Pending: 7, + PublishersInGate: 2, + Cause: context.DeadlineExceeded, + } + + require.Equal(t, + "batcher shutdown incomplete: 7 pending, 2 publishers still admitting: context deadline exceeded", + err.Error()) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.ErrorIs(t, err, batcher.ErrTimeout) +} diff --git a/pkg/batcher/stability_test.go b/pkg/batcher/stability_test.go new file mode 100644 index 0000000..18f65d2 --- /dev/null +++ b/pkg/batcher/stability_test.go @@ -0,0 +1,346 @@ +package batcher_test + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/NSXBet/batcher/pkg/batcher" + "github.com/stretchr/testify/require" +) + +// Stability tests: lifecycle interleavings that must not hang, panic, or lose work. +// +// Each of these covers behaviour that is currently correct but pinned by nothing, so +// a future refactor could break it silently. They are written as repeated trials +// rather than single runs because every one of them is an ordering race, and a single +// pass proves very little about a race. + +// TestBlockingAddIsReleasedByShutdown covers the one admission path with no error +// return. +// +// Enqueue blocked on a full bounded queue is already tested, but Add is the +// compatibility fast path: it returns nothing, so if shutdown failed to release it +// the caller would hang with no way to observe why. That makes it the worst place to +// leave untested, and it is only reachable in bounded mode. +func TestBlockingAddIsReleasedByShutdown(t *testing.T) { + t.Parallel() + + const trials = 100 + + for trial := range trials { + release := make(chan struct{}) + + var releaseOnce sync.Once + + releaseAll := func() { releaseOnce.Do(func() { close(release) }) } + + b := batcher.New( + batcher.WithBatchSize[int](1), + batcher.WithBatchInterval[int](time.Hour), + batcher.WithMaxQueueSize[int](2), + // Unstarted, so nothing drains and the queue stays full. + batcher.WithSkipAutoStart[int](), + batcher.WithProcessor(func([]int) error { + <-release + + return nil + }), + ) + + func() { + defer releaseAll() + + for i := range 2 { + b.Add(i) + } + + parked := make(chan struct{}) + + go func() { + defer close(parked) + + b.Add(99) // must block: bounded queue full, nothing consuming + }() + + shutdownDone := make(chan error, 1) + + go func() { shutdownDone <- b.Close() }() + + select { + case <-parked: + case <-time.After(15 * time.Second): + t.Fatalf("trial %d: a blocking Add was never released by shutdown; "+ + "Add has no error return, so the caller would hang indefinitely", trial) + } + + releaseAll() + + select { + case <-shutdownDone: + case <-time.After(15 * time.Second): + t.Fatalf("trial %d: shutdown never completed with a parked Add", trial) + } + }() + } +} + +// TestAddRacingCloseKeepsAccountingExact covers the interleaving a caller is most +// likely to produce accidentally: a request handler still enqueuing while the +// process shuts down. +// +// Every accepted item must reach exactly one terminal outcome. A lost item, a +// double-count, or a send on a closed channel all show up here. +func TestAddRacingCloseKeepsAccountingExact(t *testing.T) { + t.Parallel() + + const ( + trials = 200 + perProducer = 50 + ) + + for trial := range trials { + b := batcher.New( + batcher.WithBatchSize[int](8), + batcher.WithBatchInterval[int](time.Millisecond), + batcher.WithProcessor(batcher.NoOpProcessor[int]), + ) + + produced := make(chan struct{}) + + go func() { + defer close(produced) + + for i := range perProducer { + b.Add(i) + } + }() + + require.NoError(t, b.Close(), "trial %d", trial) + <-produced + + stats := b.Stats() + + require.Equal(t, stats.Accepted, stats.Completed+stats.Failed+stats.Panicked, + "trial %d: every accepted item must reach exactly one terminal outcome "+ + "(accepted=%d completed=%d failed=%d panicked=%d)", + trial, stats.Accepted, stats.Completed, stats.Failed, stats.Panicked) + require.Zero(t, stats.Pending, "trial %d", trial) + require.Zero(t, stats.IntakePending, "trial %d", trial) + require.LessOrEqual(t, stats.Accepted, uint64(perProducer), + "trial %d: cannot accept more than was offered", trial) + } +} + +// TestPanicInFinalBatchStillClosesDiagnostics covers the narrowest window in the +// shutdown protocol: a diagnostic published from the last batch, while the +// coordinator is closing the diagnostics channel. +// +// Get the single-closer rule wrong and this panics with "send on closed channel" +// inside the library's own goroutine, where the caller cannot recover it. +func TestPanicInFinalBatchStillClosesDiagnostics(t *testing.T) { + t.Parallel() + + const trials = 150 + + for trial := range trials { + b := batcher.New( + // Large batch, long interval: only shutdown flushes this. + batcher.WithBatchSize[int](100), + batcher.WithBatchInterval[int](time.Hour), + batcher.WithProcessor(func([]int) error { + panic("poison final batch") + }), + ) + + drained := make(chan int, 1) + + go func() { + count := 0 + + for range b.Errors() { + count++ + } + + drained <- count + }() + + b.Add(1) + + require.NoError(t, b.Shutdown(context.Background()), "trial %d", trial) + + select { + case count := <-drained: + require.Positive(t, count, + "trial %d: the recovered panic must be observable", trial) + case <-time.After(15 * time.Second): + t.Fatalf("trial %d: Errors() never closed, so a ranging consumer "+ + "would block forever", trial) + } + + stats := b.Stats() + + require.Equal(t, uint64(1), stats.Panicked, "trial %d", trial) + require.Zero(t, stats.Pending, + "trial %d: a panic must not strand a drain obligation", trial) + } +} + +// TestConcurrentShutdownCallersAllTerminate covers many goroutines calling Shutdown +// at once, which is what happens when several subsystems react to the same signal. +// +// Exactly one drain must run, none may deadlock, and no caller may receive a false +// error for work that completed. +func TestConcurrentShutdownCallersAllTerminate(t *testing.T) { + t.Parallel() + + const ( + trials = 100 + callers = 8 + ) + + for trial := range trials { + var processed atomic.Int64 + + b := batcher.New( + batcher.WithBatchSize[int](4), + batcher.WithBatchInterval[int](time.Millisecond), + batcher.WithProcessor(func(items []int) error { + processed.Add(int64(len(items))) + + return nil + }), + ) + + for i := range 12 { + b.Add(i) + } + + var ( + wg sync.WaitGroup + errs = make([]error, callers) + ) + + for i := range callers { + wg.Add(1) + + go func(idx int) { + defer wg.Done() + + errs[idx] = b.Shutdown(context.Background()) + }(i) + } + + wg.Wait() + + for i, err := range errs { + require.NoError(t, err, + "trial %d caller %d: an unbounded wait must observe the completed drain", + trial, i) + } + + require.True(t, b.IsClosed(), "trial %d", trial) + require.Equal(t, int64(12), processed.Load(), + "trial %d: every item processed exactly once regardless of caller count", trial) + } +} + +// TestErroringProcessorDoesNotStallShutdown covers the failure mode that hangs a +// process rather than reporting: a processor that always errors. +// +// A failed batch must still release its drain obligation. If it did not, Pending +// would never reach zero and shutdown would wait forever on work that already ran. +func TestErroringProcessorDoesNotStallShutdown(t *testing.T) { + t.Parallel() + + failure := errors.New("downstream unavailable") + + const items = 200 + + b := batcher.New( + batcher.WithBatchSize[int](1), + batcher.WithBatchInterval[int](time.Millisecond), + // Deliberately tiny: most diagnostics are dropped, which must not block. + batcher.WithErrorBufferSize[int](2), + batcher.WithProcessor(func([]int) error { + return failure + }), + ) + + for i := range items { + b.Add(i) + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + require.NoError(t, b.Shutdown(ctx), + "a processor that always fails must not prevent the drain from completing") + + stats := b.Stats() + + require.Equal(t, uint64(items), stats.Failed) + require.Zero(t, stats.Completed) + require.Zero(t, stats.Pending) + require.Positive(t, stats.DroppedErrors, + "with a 2-entry buffer and %d failures, diagnostics must be dropped and counted", + items) +} + +// TestRepeatedStartAndShutdownIsStable covers Start and Shutdown racing from several +// goroutines, which is reachable through Fx lifecycle hooks and supervisors. +// +// Exactly one processing lifecycle must exist, and the batcher must always reach a +// terminal state. +func TestRepeatedStartAndShutdownIsStable(t *testing.T) { + t.Parallel() + + const trials = 150 + + for trial := range trials { + var processed atomic.Int64 + + b := batcher.New( + batcher.WithBatchSize[int](1), + batcher.WithBatchInterval[int](time.Millisecond), + batcher.WithSkipAutoStart[int](), + batcher.WithProcessor(func(items []int) error { + processed.Add(int64(len(items))) + + return nil + }), + ) + + b.Add(1) + + var wg sync.WaitGroup + + for range 4 { + wg.Add(1) + + go func() { + defer wg.Done() + + b.Start() + }() + } + + wg.Add(1) + + go func() { + defer wg.Done() + + _ = b.Shutdown(context.Background()) + }() + + wg.Wait() + + require.NoError(t, b.Shutdown(context.Background()), "trial %d", trial) + require.True(t, b.IsClosed(), "trial %d", trial) + require.Equal(t, int64(1), processed.Load(), + "trial %d: the item must be processed exactly once by one lifecycle", trial) + } +} From c0a2e6276feb085592ac575b20622146a26bba3a Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Fri, 7 Aug 2026 10:41:35 -0300 Subject: [PATCH 5/8] fix: make Start inert until construction completes Five findings from the Phase 5 adversarial review. The substantive one is a live data race the review reproduced and I confirmed. Option[T] is func(*Batcher[T]) -- an arbitrary closure, not a declarative value -- so a caller can pass an option that calls Start during New's option loop. The aggregator then launched while New was still assigning fields: WARNING: DATA RACE Read at ... batcher.(*Batcher[int]).run() batcher.go:445 Previous write at ... batcher.New[int]() batcher.go:108 That is a race inside the library, triggered by caller code. Fixed structurally rather than by narrowing the doc claim: Start is now inert until New sets constructed, which happens after every field the pipeline touches is assigned. It degrades to a no-op rather than panicking, because New starts the batcher itself unless WithSkipAutoStart was given, and an explicit Start afterwards still works. Two regression tests cover the hostile option and the skip-auto-start path. The fx.go comment claiming auto-start "cannot be re-enabled by a caller-supplied option" was true only for declarative options. It now says why the flag ordering works and points at Start for the closure case, instead of asserting something the type system does not guarantee. Three documentation corrections: - default-window.md's 5ms row said p99.9 measured "worse" than 10ms while its own numbers show 13.1ms versus 21.7ms -- better, not worse. The row now states the comparison correctly and makes the real argument the coalescing one: 6 items per batch versus 11, for a latency gain the measurement cannot distinguish. - constants.go cited ~22ms as p99 at 1k/s. That is the p99.9 figure; p99 is ~12ms, which is what the README, PR body and decision record all say. - fx.go said repeated stop observes "the same terminal result", implying a sticky error. shutdown.go and compatibility.md already document the opposite, and TestShutdownDeadlineDoesNotPoisonLaterCallers proves it: later callers wait on the same drain and return their own result. --- docs/improvements/default-window.md | 2 +- docs/improvements/thresholds.md | 2 + pkg/batcher/batcher.go | 22 ++++++++++ pkg/batcher/constants.go | 2 +- pkg/batcher/fx.go | 16 ++++++-- pkg/batcher/options_freeze_test.go | 64 +++++++++++++++++++++++++++++ 6 files changed, 102 insertions(+), 6 deletions(-) diff --git a/docs/improvements/default-window.md b/docs/improvements/default-window.md index cd7d5be..7d1db0e 100644 --- a/docs/improvements/default-window.md +++ b/docs/improvements/default-window.md @@ -134,7 +134,7 @@ timer-only model would predict, because batches still fill during the burst. | --- | --- | | **1ms** | Increases downstream calls under saturation (495/s versus 199/s) and coalesces only 2 items at 1k/s. It optimises latency by nearly removing batching. | | **2ms** | At 1k/s, 3 items per batch and 373 calls/s. Too little coalescing for a default. | -| **5ms** | Defensible, and better latency. But at 1k/s it yields 6 items per batch versus 11 at 10ms, and p99.9 measured *worse* than 10ms at that rate (13.1ms versus 21.7ms is within noise at 1k/s, but the coalescing gap is not). 10ms is the more conservative default. | +| **5ms** | Defensible, and genuinely lower latency: at 1k/s its p99.9 measured *better* than 10ms (13.1ms versus 21.7ms), though both are within noise at that rate. The reason to prefer 10ms is coalescing, not latency — 6 items per batch versus 11 — which halves the batching benefit for a latency gain the measurement cannot even distinguish. | | **20-100ms** | Better coalescing at low rates, but reintroduces the per-hop latency this work exists to remove, and provides no additional protection above 10ms under saturation. | 10ms is the point where latency is bounded at roughly the window while coalescing diff --git a/docs/improvements/thresholds.md b/docs/improvements/thresholds.md index 8e7d41e..476681a 100644 --- a/docs/improvements/thresholds.md +++ b/docs/improvements/thresholds.md @@ -170,6 +170,8 @@ Verified this way so far: | `BatchHeld` ownership transfer | field never populated | | Dispatch accounting | flush and terminal counts wrong | | Option freeze guard | data race under `-race` | +| Runtime config snapshot | data race under `-race` | +| `Start` construction guard | data race inside `New` under `-race` | | Queue capacity bound | `tryPush` grows past the bound | | Queue seal release | blocked publisher never released | | Blocking `Add` release on shutdown | caller hangs indefinitely | diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go index b32e11e..70ecd37 100644 --- a/pkg/batcher/batcher.go +++ b/pkg/batcher/batcher.go @@ -53,6 +53,14 @@ type Batcher[T any] struct { // struct a caller may still hold a reference to. runtime runtimeConfig[T] + // constructed becomes true at the end of New. Start is inert before that. + // + // Option is an arbitrary func(*Batcher[T]), so a caller can pass one that calls + // Start during New's option loop. Without this guard the aggregator would launch + // while New was still assigning fields such as input, which is a data race + // inside the library triggered by caller code. + constructed atomic.Bool + gate *admissionGate counters counters @@ -145,6 +153,11 @@ func New[T any](options ...Option[T]) *Batcher[T] { b.input = newQueue[T](b.config.MaxQueueSize) b.errorsChan = make(chan error, b.config.ErrorBufferSize) + // Everything the pipeline touches is now assigned, so Start may launch it. A + // Start attempted earlier -- from a hostile or buggy Option -- was a no-op, and + // this is where that stops being true. + b.constructed.Store(true) + if !b.config.SkipAutoStart { b.Start() } @@ -180,6 +193,15 @@ func (b *Batcher[T]) validateConfig() { // the drain waiting on a consumer that never exists, which deadlocks shutdown when // Start and Shutdown race. func (b *Batcher[T]) Start() { + // Inert until New has finished wiring the batcher. Option is an arbitrary + // closure, so a caller can invoke Start from inside the option loop; launching + // the aggregator then would read fields New has not assigned yet. Ignoring the + // call is safe: New starts the batcher itself unless WithSkipAutoStart was + // given, and a caller that meant to start it can call Start after New returns. + if !b.constructed.Load() { + return + } + b.startOnce.Do(func() { b.gate.state.CompareAndSwap(stateNew, stateRunning) diff --git a/pkg/batcher/constants.go b/pkg/batcher/constants.go index 4281dcc..fc56243 100644 --- a/pkg/batcher/constants.go +++ b/pkg/batcher/constants.go @@ -10,7 +10,7 @@ const ( // // 10ms is the evidence-backed latency/coalescing default. The decision matrix // measured a 1s default adding up to ~1s p99 latency at 1k items/s, while 10ms - // reduced it to ~22ms and still coalesced ~11 items per batch. At 10k items/s it + // reduced it to ~12ms and still coalesced ~11 items per batch. At 10k items/s it // coalesced ~101 items per batch while reducing p99 ~100ms to ~10ms. See // docs/improvements/default-window.md for the complete matrix and caveats. // diff --git a/pkg/batcher/fx.go b/pkg/batcher/fx.go index 89c31a9..9d56715 100644 --- a/pkg/batcher/fx.go +++ b/pkg/batcher/fx.go @@ -76,8 +76,10 @@ func ProvideBatcherInFXWithOptions[T any]( // background. Accepted work is never discarded to make shutdown look clean. // - A batcher that never started but holds queued work still drains, because // Shutdown starts the consumer when one is needed. -// - Repeated application stop is idempotent: later Shutdown calls observe the -// same terminal result. +// - Repeated application stop is idempotent: later Shutdown calls wait on the +// same drain and return their own wait result. A caller that waits long enough +// sees nil even if an earlier caller timed out; an expired deadline is never +// stored and replayed to later callers. func provideBatcherModule[T any]( processorFactory any, buildOptions func(Processor[T]) []Option[T], @@ -92,8 +94,14 @@ func provideBatcherModule[T any]( func(processorFunc Processor[T]) *Batcher[T] { options := buildOptions(processorFunc) - // Fx owns the lifecycle, so auto-start is forced off last and cannot - // be re-enabled by a caller-supplied option. + // Fx owns the lifecycle, so auto-start is forced off last: a + // caller-supplied WithSkipAutoStart cannot re-enable it, because the + // flag is only read after every option has run. + // + // Option is an arbitrary closure rather than a declarative value, so + // a caller could also try to start the batcher directly from an + // option. That is handled in Start, which is inert until New has + // finished wiring the batcher, not here. options = append(options, WithSkipAutoStart[T]()) return New(options...) diff --git a/pkg/batcher/options_freeze_test.go b/pkg/batcher/options_freeze_test.go index b8da595..93d9cec 100644 --- a/pkg/batcher/options_freeze_test.go +++ b/pkg/batcher/options_freeze_test.go @@ -144,3 +144,67 @@ func TestEveryOptionRespectsTheFreeze(t *testing.T) { "post-construction WithProcessor must not replace the original processor") require.NoError(t, after.ProcessorFunc(nil)) } + +// TestHostileOptionCannotStartDuringConstruction pins that Start is inert until New +// has finished wiring the batcher. +// +// Option is an arbitrary func(*Batcher[T]), not a declarative value, so a caller can +// pass one that calls Start inside New's option loop. Before the guard that launched +// the aggregator while New was still assigning fields, producing a data race inside +// the library rather than in the caller's code: +// +// WARNING: DATA RACE +// Read at ... by goroutine 9: batcher.(*Batcher[int]).run() +// Previous write at ... by goroutine 8: batcher.New[int]() +// +// Run under -race, this fails if the guard is removed. +func TestHostileOptionCannotStartDuringConstruction(t *testing.T) { + t.Parallel() + + var processed atomic.Int64 + + b := batcher.New[int]( + // Hostile: tries to start the pipeline mid-construction. + batcher.Option[int](func(inner *batcher.Batcher[int]) { inner.Start() }), + batcher.WithBatchSize[int](1), + batcher.WithBatchInterval[int](time.Millisecond), + batcher.WithProcessor(func(items []int) error { + processed.Add(int64(len(items))) + + return nil + }), + ) + + // The batcher must still be fully functional: New starts it after wiring. + b.Add(1) + + require.NoError(t, b.Join(10*time.Second)) + require.Equal(t, int64(1), processed.Load(), + "the batcher must process normally despite the hostile option") + require.NoError(t, b.Close()) +} + +// TestStartBeforeConstructionIsIgnoredNotFatal pins that the guard degrades to a +// no-op rather than panicking, so a buggy option cannot take the process down. +func TestStartBeforeConstructionIsIgnoredNotFatal(t *testing.T) { + t.Parallel() + + require.NotPanics(t, func() { + b := batcher.New[int]( + batcher.Option[int](func(inner *batcher.Batcher[int]) { + inner.Start() + inner.Start() + }), + batcher.WithSkipAutoStart[int](), + batcher.WithProcessor(batcher.NoOpProcessor[int]), + ) + + // SkipAutoStart plus a suppressed in-option Start means nothing is running; + // an explicit Start afterwards must still work. + b.Start() + b.Add(1) + + require.NoError(t, b.Join(10*time.Second)) + require.NoError(t, b.Close()) + }) +} From a50fad9d3e3354bdcd3ace4d927a915174ee786a Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Fri, 7 Aug 2026 10:50:35 -0300 Subject: [PATCH 6/8] fix: keep the default decision test aligned with scenario metrics Phase 1 renamed PendingPeak to PendingWorkPeak because Len() is sampled in-system work, not queue depth. The Phase 5 decision test was added after that layer and still referenced the old field, making the fully-rebased stack fail vet and build. Use the renamed field and label its diagnostic column work_peak so the test does not revive the discarded queue-depth claim. --- test/scenario/decision_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/scenario/decision_test.go b/test/scenario/decision_test.go index 3653c09..075dfdb 100644 --- a/test/scenario/decision_test.go +++ b/test/scenario/decision_test.go @@ -115,7 +115,7 @@ func TestDefaultWindowUnderSustainedOverload(t *testing.T) { t.Logf("offered=%d/s processor=%s (deliberately saturated)", rate, serviceTime) t.Logf("") t.Logf("%-8s %-12s %-12s %-12s %-11s %-10s %s", - "window", "accepted/s", "completed/s", "pending_peak", "heap_peak", "calls/s", "batch") + "window", "accepted/s", "completed/s", "work_peak", "heap_peak", "calls/s", "batch") for _, window := range []time.Duration{ time.Millisecond, @@ -138,7 +138,7 @@ func TestDefaultWindowUnderSustainedOverload(t *testing.T) { window, result.AcceptedRate, result.CompletedRate, - result.PendingPeak, + result.PendingWorkPeak, formatMB(result.HeapHighWater), result.DownstreamPerSec, result.MeanBatchSize) From c4f5d387f4b62534f141a486d74dfb7972b8ee1e Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Fri, 7 Aug 2026 11:29:13 -0300 Subject: [PATCH 7/8] fix: keep the 1s default and panic on Start before construction Two review decisions from PR #33. Keep DefaultBatchInterval at 1s. The latency evidence for 10ms stands and is unchanged, but it was only half the trade: at 1,000 items/s the measured downstream call rate moves from ~1/s to ~94/s, and that cost falls on every caller who never set an interval and never asked for lower latency. Latency is visible to whoever measures it and is fixed with one option; a silent ~90x increase in downstream calls is not visible until something else saturates. So 10ms becomes a published recommendation and the default stays where existing deployments already are. The decision record now records that outcome rather than a default change, with an "Adopting 10ms" section carrying the call-rate cost per arrival rate so the trade is visible at the point of choosing. compatibility.md drops the breaking-change entry -- there is nothing to migrate -- and plan-perf.md records why its own gate ("change the default only if latency, batching efficiency, allocation and overload behaviour all support it") was not met: downstream call rate is the axis that fails. TestDefaultBatchIntervalStaysOneSecond pins the literal value through both the constant and Config. Nothing asserted the concrete default before, only that Config agreed with whatever the constant said, so the constant could drift back with a green suite. Sabotage-verified. Start now panics when called before construction finishes, instead of returning silently. Reachable only from inside an Option, since Option is an arbitrary func(*Batcher[T]). The previous no-op avoided the data race but left a caller who wrote Start believing a batcher was running when it was not; a panic names the cause, which the stack alone does not since it points into New. Also corrected the slow-processor examples flagged in review. Both documents cited a 10ms window against 100ms while TestReproducesInlineSlowProcessorInversion measures 5ms against 100ms, and quoted 120ms/100ms as if pinned when the test asserts only the ordering. Re-measured on this host (5ms p50 120ms, 100ms p50 100ms), corrected the window, labelled the environment, and stated what the test actually guarantees. The unpinned "4ms with 8 workers" claim is replaced by a reference to TestConcurrencyRemovesSlowProcessorCoupling, which pins the improvement without quoting a number no test defends. --- README.md | 57 ++++++++++++------ docs/improvements/compatibility.md | 43 ++++++-------- docs/improvements/default-window.md | 90 ++++++++++++++++++++--------- docs/improvements/plan-perf.md | 12 +++- pkg/batcher/batcher.go | 19 ++++-- pkg/batcher/compatibility_test.go | 26 +++++++++ pkg/batcher/constants.go | 22 ++++--- pkg/batcher/options_freeze_test.go | 75 ++++++++++++------------ 8 files changed, 220 insertions(+), 124 deletions(-) diff --git a/README.md b/README.md index a8b95d5..c233238 100644 --- a/README.md +++ b/README.md @@ -348,7 +348,7 @@ fmt.Printf("batcher has %d items\n", batcher.Len()) | --- | --- | --- | | `WithProcessor(fn)` | no-op | The function called with each batch. | | `WithBatchSize[T](n)` | 1000 | Flush as soon as a batch holds `n` items. | -| `WithBatchInterval[T](d)` | **10ms** | Maximum age of a partial batch. See [Choosing a batch interval](#choosing-a-batch-interval). | +| `WithBatchInterval[T](d)` | 1s | Maximum age of a partial batch. 1s is conservative; see [Choosing a batch interval](#choosing-a-batch-interval). | | `WithMaxQueueSize[T](n)` | unbounded | Bound queued items to get back-pressure instead of unbounded growth. | | `WithConcurrency[T](n)` | 1 | Process `n` batches at once. Requires `WithoutOrderedProcessing`. | | `WithoutOrderedProcessing[T]()` | off | Acknowledges that `n > 1` gives up cross-batch ordering and processor mutual exclusion. | @@ -374,15 +374,31 @@ reaches 1,000 items after about 20ms, so configuring a 100ms interval changes no Use the formula to pick an interval from the coalescing you actually need: -| Arrival rate | 10ms (default) | 100ms | -| --- | --- | --- | -| 1,000/s | ~11 items/batch, ~94 calls/s, p99 ~12ms | ~100 items/batch, ~10 calls/s, p99 ~100ms | -| 10,000/s | ~101 items/batch, ~99 calls/s, p99 ~10ms | ~1000 items/batch, ~10 calls/s, p99 ~99ms | -| 50,000/s | ~500 items/batch, ~100 calls/s, p99 ~10ms | ~1000 items/batch (capped by `BatchSize`), ~50 calls/s, p99 ~20ms | +| Arrival rate | 10ms | 100ms | 1s (default) | +| --- | --- | --- | --- | +| 1,000/s | ~11 items/batch, ~94 calls/s, p99 ~12ms | ~100 items/batch, ~10 calls/s, p99 ~100ms | ~1000 items/batch, ~1 call/s, p99 ~981ms | +| 10,000/s | ~101 items/batch, ~99 calls/s, p99 ~10ms | ~1000 items/batch, ~10 calls/s, p99 ~99ms | ~1000 items/batch, ~10 calls/s, p99 ~99ms | +| 50,000/s | ~500 items/batch, ~100 calls/s, p99 ~10ms | ~1000 items/batch (capped by `BatchSize`), ~50 calls/s, p99 ~20ms | ~1000 items/batch, ~50 calls/s, p99 ~20ms | + +**The default is 1s, and it is deliberately conservative.** It favours coalescing and +a low downstream call rate over latency, which is the safe direction for a shared +library: a caller who needs lower latency can measure it and set one option, whereas a +default that quietly multiplied everyone's downstream call rate would not show up +until something else saturated. + +**Set 10ms when latency matters.** At sparse rates it is the difference between p99 +~981ms and ~12ms while still coalescing ~11 items per batch. Note the cost in the +table: at 1,000/s it moves ~1 call/s to ~94 calls/s. Above ~10k/s, `BatchSize` closes +batches before either timer fires and the choice stops mattering. -Raise the interval when your downstream is expensive enough that the call rate at -10ms is unacceptable — most relevant below a few thousand items/s. The full matrix -and the reasoning behind the default are in +```go +b := batcher.New( + batcher.WithProcessor(processor), + batcher.WithBatchInterval[Item](10*time.Millisecond), +) +``` + +The full matrix, and why 10ms is recommended rather than made the default, are in [`docs/improvements/default-window.md`](./docs/improvements/default-window.md). **Shrinking the interval is not overload protection.** Measured with a saturated @@ -392,17 +408,24 @@ completing the same amount, with the queue absorbing the difference. Use **A slow processor bounds the effective interval.** At the default `Concurrency` of 1, one batch must finish before the next starts, so the effective interval is -`max(BatchInterval, processor duration)`. Measured with a 50ms processor at 10k -items/s: +`max(BatchInterval, processor duration)`. A window below the processor duration buys +nothing and can measure worse, because queueing dominates once the processor is the +bottleneck. + +Measured with a 50ms processor at 10k items/s on darwin/arm64 (Apple M4 Pro, Go +1.26.5, `GOMAXPROCS=12`): | Configured interval | Effective behaviour | p50 latency | | --- | --- | --- | -| 5ms | bounded by the 50ms processor | **120ms** | -| 100ms | bounded by the interval | **100ms** | - -The smaller interval is *worse*, because queueing dominates. If your processor is -slow, raise `WithConcurrency` (with `WithoutOrderedProcessing`) rather than lowering -the interval. With 8 workers, the same scenario measured a p50 latency of 4ms. +| 5ms | bounded by the 50ms processor | ~120ms | +| 100ms | bounded by the interval | ~100ms | + +`TestReproducesInlineSlowProcessorInversion` asserts the ordering — the smaller +window is slower — and not these millisecond values, which are host-specific. If your +processor is slow, raise `WithConcurrency` (with `WithoutOrderedProcessing`) rather +than lowering the interval: `TestConcurrencyRemovesSlowProcessorCoupling` pins that +8 workers reduce p50 against 1 worker in the same scenario, because workers add the +capacity a shorter window cannot. ## FX Integration diff --git a/docs/improvements/compatibility.md b/docs/improvements/compatibility.md index d9d5482..dd7981c 100644 --- a/docs/improvements/compatibility.md +++ b/docs/improvements/compatibility.md @@ -11,7 +11,8 @@ listed below with what it was, what it is now, and what a caller has to do. The module is `v0.x`, so a minor bump is the correct signal for breaking changes under Go module semantics: no import path change is required, and callers opt in by updating. A `v1.0.0` would be the wrong claim to make while the default batch -interval is still under review (see the [decision record](./default-window.md)). +interval is still under review (see the [decision record](./default-window.md), which +keeps 1s for now and recommends 10ms as an explicit setting). ## Requires Go 1.25.0 @@ -63,46 +64,38 @@ b = batcher.New(batcher.WithBatchSize[Item](500), batcher.WithProcessor(fn)) Code that only *reads* `Config()` compiles unchanged. Code that assigned through it will now fail to compile, which is the intended outcome: it was racing. -### `DefaultBatchInterval` changes 1s → 10ms +### `DefaultBatchInterval` is unchanged at 1s | | | | --- | --- | | **Before** | `DefaultBatchInterval = 1 * time.Second` | -| **After** | `DefaultBatchInterval = 10 * time.Millisecond` | +| **After** | `DefaultBatchInterval = 1 * time.Second` | -This affects every caller that did not set `WithBatchInterval`, and it is a -behaviour change rather than a bug fix: batches close sooner, so latency drops and -the downstream call rate rises whenever `BatchSize` is not reached first. +An earlier draft of this release changed the default to 10ms on latency evidence. It +was not shipped. The measurement holds — at 1,000 items/s the 1s default measures p99 +981ms against 12ms at 10ms — but the same tables show the downstream call rate moving +from ~1 call/s to ~94 calls/s at that rate. That is a CPU and downstream-load +increase for every caller who never configured an interval, so the low-latency value +is published as a recommendation instead of imposed as a default. -The decision and its full measurement matrix are in -[`default-window.md`](./default-window.md). In short: at 1,000 items/s the old -default measured p99 981ms; 10ms measured p99 12ms while still coalescing ~11 items -per batch. +**Nothing to migrate.** Callers keep the behaviour they have today. -**Who is affected:** - -| Situation | Effect | Action | -| --- | --- | --- | -| Sets a **positive** `WithBatchInterval` | None | None | -| Passes `WithBatchInterval(0)` or a negative value | Falls back to the default, so it moves to 10ms too | Pass an explicit positive interval to keep 1s | -| Relies on the default, traffic ≥ 10k/s | None measurable — `BatchSize` was already binding | None | -| Relies on the default, sparse traffic | Latency drops sharply; more downstream calls | Confirm the new call rate is acceptable | -| Depends on ~1000-item batches at low traffic | Batches become much smaller | Set `WithBatchInterval` explicitly, or raise `BatchSize` | - -**Migration:** +To adopt the recommendation, set it explicitly: ```go -// Keep the previous behaviour explicitly. b := batcher.New( batcher.WithProcessor(fn), - batcher.WithBatchInterval[Item](time.Second), + batcher.WithBatchInterval[Item](10*time.Millisecond), ) ``` +The decision, its full matrix, and the call-rate cost per arrival rate are in +[`default-window.md`](./default-window.md). + Note that shrinking the interval is **not** overload protection, and neither is keeping it large: under a saturated downstream every interval from 10ms to 1s -accepted the same load. If the concern is memory rather than latency, bound the -queue instead: +accepted the same load. If the concern is memory rather than latency, bound the queue +instead: ```go b := batcher.New( diff --git a/docs/improvements/default-window.md b/docs/improvements/default-window.md index 7d1db0e..a3c2982 100644 --- a/docs/improvements/default-window.md +++ b/docs/improvements/default-window.md @@ -1,13 +1,24 @@ # Decision record: default batch interval -**Decision:** `DefaultBatchInterval` changes from **1s to 10ms** in v0.3.0. - -**Status:** accepted, with the caveat in [Where 10ms is the wrong -choice](#where-10ms-is-the-wrong-choice). - -This record exists because the plan required the default to follow measurement -rather than intuition. If the evidence had been ambiguous, the correct outcome was to -keep 1s and publish a configuration recommendation instead. It was not ambiguous. +**Decision:** `DefaultBatchInterval` **stays at 1s**. 10ms is published as the +recommended value for latency-sensitive services, and callers opt in with +`WithBatchInterval`. + +**Status:** accepted. Supersedes an earlier draft of this record that changed the +default to 10ms. + +The plan required the default to follow measurement rather than intuition, and the +measurement below is not in dispute: 10ms is a large latency win at sparse rates and +still coalesces meaningfully. What the latency tables alone do not settle is who pays +for it. At 1k items/s the downstream call rate moves from ~10 calls/s to ~94 calls/s — +roughly 10x more downstream requests and the CPU that goes with them — for every +existing caller who never set an interval and never asked for lower latency. + +A default is the setting imposed on people who have not thought about the trade-off. +Latency is visible to whoever is measuring it and can be fixed with one option; a +silent 10x increase in downstream load is not visible to that same caller until +something else saturates. So the evidence is published as tuning guidance, and the +default stays where existing deployments already are. ## The problem being solved @@ -18,7 +29,9 @@ worst for sparse traffic, which is exactly the traffic that gains least from batching. With a 1s default, a five-hop sparse path could accumulate several seconds before any -downstream work happened. +downstream work happened. That is the cost this record quantifies, and the reason +10ms is *recommended*; it is not, on its own, a reason to move the default for +callers whose downstream cost this record cannot see. ## Environment @@ -109,8 +122,8 @@ are the same. This is the check that prevented the decision resting on latency a — a smaller default would have been wrong if it traded latency for stability. Only 1ms differs, and it is *worse* on downstream load (495 calls/s versus 199) -because it flushes before batches fill. That is one reason the default is 10ms rather -than the lowest measured value. +because it flushes before batches fill. That is one reason the recommendation is 10ms +rather than the lowest measured value. ## Evidence 4: bursts @@ -128,7 +141,7 @@ worst on paper, since each burst coalesces from a standing start: 10ms halves p99 against 100ms while only doubling downstream calls — not the 10x a timer-only model would predict, because batches still fill during the burst. -## Why 10ms and not 5ms or 1ms +## Why the recommendation is 10ms, and not 5ms or 1ms | Candidate | Rejected because | | --- | --- | @@ -139,8 +152,10 @@ timer-only model would predict, because batches still fill during the burst. 10ms is the point where latency is bounded at roughly the window while coalescing remains meaningful at every rate tested (11x at 1k/s, 101x at 10k/s, 500x at 50k/s). +That makes it the right *recommendation*; the call-rate cost above is why it is not +the default. -## Where 10ms is the wrong choice +## Where 10ms is the wrong choice (and the default is right) **A service at ~1k items/s with an expensive downstream.** 10ms gives ~11x coalescing and ~94 downstream calls/s. If a downstream call is costly enough that 94 @@ -162,25 +177,43 @@ with the queue absorbing the difference. Protection requires `WithMaxQueueSize` `Enqueue`, or upstream flow control. **A service with a slow processor and `Concurrency=1`.** The effective interval is -`max(BatchInterval, processor duration)`. With a 50ms processor, a 10ms window -behaves like a 50ms one — and measured p50 *worse* than a 100ms window (120ms versus -100ms), because queueing dominates. Raise `WithConcurrency` (with -`WithoutOrderedProcessing`) before lowering the interval. +`max(BatchInterval, processor duration)`, so a window below the processor duration +buys nothing and can measure *worse*. With a 50ms processor at 10k items/s, a **5ms** +window measured p50 120ms against 100ms for a 100ms window, because queueing +dominates once the processor is the bottleneck. + +`TestReproducesInlineSlowProcessorInversion` pins the ordering (5ms slower than +100ms), not those millisecond values; the figures are from the environment recorded +above (darwin/arm64, Apple M4 Pro, Go 1.26.5, `GOMAXPROCS=12`) and will differ +elsewhere. Raise `WithConcurrency` (with `WithoutOrderedProcessing`) before lowering +the interval. + +## Adopting 10ms + +There is no migration: the default is unchanged, so every existing caller keeps the +behaviour it has today. Adopting the recommendation is opt-in. -## Migration +```go +b := batcher.New( + batcher.WithProcessor(processor), + batcher.WithBatchInterval[Item](10*time.Millisecond), +) +``` -This is a **behaviour change for every caller who did not set `WithBatchInterval`**. +Before adopting it, check the effect on your own traffic. The interval only binds +when it closes a batch before `BatchSize` does — expected batch size is +approximately `min(arrival rate × interval, BatchSize)` — so the change is largest at +sparse rates and vanishes once `BatchSize` is reached first. -| Situation | Effect | Action | -| --- | --- | --- | -| Sets `WithBatchInterval` to a **positive** value | None | None | -| Passes `WithBatchInterval(0)` or a negative duration | Falls back to `DefaultBatchInterval`, so it moves 1s → 10ms with everyone else | Pass an explicit positive interval if you relied on the old 1s fallback | -| Relies on the 1s default, traffic ≥ 10k/s | None measurable — `BatchSize` was already the binding constraint | None | -| Relies on the 1s default, sparse traffic | Latency drops sharply; downstream call rate rises | Verify the new call rate is acceptable; set a larger interval if not | -| Depends on ~1000-item batches at low traffic | Batches become much smaller | Set `WithBatchInterval` explicitly, or raise `BatchSize` | +| Your arrival rate | 1s default | 10ms | What to check | +| --- | --- | --- | --- | +| ~1k/s | ~1000 items/batch, ~1 call/s | ~11 items/batch, ~94 calls/s | Downstream cost per call. This is the ~90x call-rate increase; it is the whole reason 10ms is not the default | +| ~10k/s | ~1000 items/batch, ~10 calls/s | ~101 items/batch, ~99 calls/s | Still ~10x more calls. `BatchSize` binds at 1s, the timer binds at 10ms | +| ~50k/s | ~1000 items/batch, ~50 calls/s | ~500 items/batch, ~100 calls/s | ~2x more calls. `BatchSize` caps both, so the gap narrows | +| Saturated | `BatchSize` binds | `BatchSize` binds | Nothing. Identical under saturation | `Stats().BatchesFlushed` with the terminal counters gives mean batch size after a -drain, which is the fastest way to confirm what the change did to your own +drain, which is the fastest way to confirm what an interval change did to your own coalescing: ```go @@ -201,3 +234,6 @@ if s.BatchesFlushed > 0 { darwin/arm64 only. - A reported workload where 10ms coalescing is insufficient *and* the sizing rule fails to predict it, which would mean the guidance is wrong rather than the default. +- Evidence that the downstream call-rate increase is acceptable for the general case — + for example a major version where callers must revisit configuration anyway. That is + what would move 10ms from recommendation to default. diff --git a/docs/improvements/plan-perf.md b/docs/improvements/plan-perf.md index 24d808e..89f2017 100644 --- a/docs/improvements/plan-perf.md +++ b/docs/improvements/plan-perf.md @@ -1149,7 +1149,10 @@ rather than intuition. - Run the full matrix against the completed implementation. - Publish the decision and its evidence. - Change `DefaultBatchInterval` only if p99 latency, batching efficiency, - allocation, and overload behaviour all support it. + allocation, and overload behaviour all support it. **Outcome: not changed.** Latency + supports 10ms strongly, but the downstream call-rate cost at sparse traffic falls on + callers who never configured an interval, so the value is recommended explicitly + instead. **Acceptance criteria** @@ -1207,8 +1210,11 @@ valid stopping point: - Stopping after **3.x**: small windows are honoured under slow processors. - Stopping after **4.x**: operators can see overload developing. - Stopping after **5.x**: the compatibility story is published, the tuning guidance - is measured rather than asserted, and `DefaultBatchInterval` is 10ms on the - evidence recorded in `default-window.md`. + is measured rather than asserted, and `DefaultBatchInterval` stays 1s while 10ms is + published as the recommended explicit setting. The gate above required latency, + batching efficiency, allocation *and* overload behaviour to support a change; the + downstream call rate at sparse traffic (~1/s to ~94/s at 1k items/s) does not, so + the evidence in `default-window.md` ships as guidance rather than as a new default. ## Conclusion diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go index 70ecd37..2792cce 100644 --- a/pkg/batcher/batcher.go +++ b/pkg/batcher/batcher.go @@ -192,14 +192,21 @@ func (b *Batcher[T]) validateConfig() { // the drain needs a consumer to make progress. Declining to start here would leave // the drain waiting on a consumer that never exists, which deadlocks shutdown when // Start and Shutdown race. +// +// Start panics if called before New has finished constructing the batcher, which is +// reachable only from inside an Option, since Option is an arbitrary +// func(*Batcher[T]). Starting there would launch the aggregator against fields New +// has not assigned yet — a data race inside the library. It panics rather than +// ignoring the call because a caller who wrote Start meant it, and silently doing +// nothing would leave them believing a batcher is running when it is not. func (b *Batcher[T]) Start() { - // Inert until New has finished wiring the batcher. Option is an arbitrary - // closure, so a caller can invoke Start from inside the option loop; launching - // the aggregator then would read fields New has not assigned yet. Ignoring the - // call is safe: New starts the batcher itself unless WithSkipAutoStart was - // given, and a caller that meant to start it can call Start after New returns. if !b.constructed.Load() { - return + panic("batcher: Start called before construction finished. " + + "Option is an arbitrary func(*Batcher[T]), so an option that calls " + + "Start runs while New is still assigning fields, and starting the " + + "aggregator there is a data race on the queue. Do not call Start from " + + "an Option: New starts the batcher itself unless WithSkipAutoStart is " + + "given, in which case call Start after New returns.") } b.startOnce.Do(func() { diff --git a/pkg/batcher/compatibility_test.go b/pkg/batcher/compatibility_test.go index 314af88..2c24e3b 100644 --- a/pkg/batcher/compatibility_test.go +++ b/pkg/batcher/compatibility_test.go @@ -165,3 +165,29 @@ func TestLegacyProvideBatcherInFXSignatureStillCompiles(t *testing.T) { require.Equal(t, 2, p.seen) } + +// TestDefaultBatchIntervalStaysOneSecond pins the default's literal value, not just +// that Config reports whatever the constant happens to say. +// +// The 10ms decision record makes a strong latency case, and it is easy to read as a +// mandate to change this constant. It was deliberately not changed: at 1,000 items/s +// the measured downstream call rate moves from ~1/s to ~94/s, and that cost lands on +// callers who never configured an interval. Lowering the default is a behaviour change +// for every such caller, so it needs its own decision rather than arriving as a tidy-up. +// +// Asserting through Config as well as the constant covers both halves: the advertised +// value and the value New actually applies. +func TestDefaultBatchIntervalStaysOneSecond(t *testing.T) { + t.Parallel() + + require.Equal(t, time.Second, batcher.DefaultBatchInterval, + "DefaultBatchInterval must stay 1s; see docs/improvements/default-window.md, "+ + "which recommends 10ms as an explicit setting rather than as the default") + + b := batcher.New(batcher.WithProcessor(batcher.NoOpProcessor[int])) + + t.Cleanup(func() { _ = b.Close() }) + + require.Equal(t, time.Second, b.Config().BatchInterval, + "a batcher constructed without WithBatchInterval must use the 1s default") +} diff --git a/pkg/batcher/constants.go b/pkg/batcher/constants.go index fc56243..fe48ed9 100644 --- a/pkg/batcher/constants.go +++ b/pkg/batcher/constants.go @@ -8,16 +8,22 @@ const ( // DefaultBatchInterval is the maximum age of a partial batch. // - // 10ms is the evidence-backed latency/coalescing default. The decision matrix - // measured a 1s default adding up to ~1s p99 latency at 1k items/s, while 10ms - // reduced it to ~12ms and still coalesced ~11 items per batch. At 10k items/s it - // coalesced ~101 items per batch while reducing p99 ~100ms to ~10ms. See - // docs/improvements/default-window.md for the complete matrix and caveats. + // It stays at 1s. A smaller default would lower latency for callers who never + // set an interval, but it raises the downstream call rate for all of them: the + // measured matrix shows ~10 calls/s at 1s versus ~94-100 calls/s at 10ms once + // the timer rather than BatchSize closes each batch. That is a CPU and + // downstream-load change imposed on existing users who did not ask for it, so + // the low-latency setting is opt-in rather than the default. + // + // Callers who want lower latency should set WithBatchInterval explicitly; 10ms + // measured p99 ~12ms at 1k items/s while still coalescing ~11 items per batch. + // docs/improvements/default-window.md holds the full matrix and its caveats, and + // the README's "Choosing a batch interval" section carries the call-rate columns + // for picking a value. // // This is a maximum partial-batch age, not overload protection. Services that - // need larger batches at low traffic must set WithBatchInterval deliberately; - // services that need process protection must use WithMaxQueueSize and Enqueue. - DefaultBatchInterval = 10 * time.Millisecond + // need process protection must use WithMaxQueueSize and Enqueue. + DefaultBatchInterval = 1 * time.Second // DefaultConcurrency is how many batches are processed at once by default. // diff --git a/pkg/batcher/options_freeze_test.go b/pkg/batcher/options_freeze_test.go index 93d9cec..b5ea656 100644 --- a/pkg/batcher/options_freeze_test.go +++ b/pkg/batcher/options_freeze_test.go @@ -145,27 +145,48 @@ func TestEveryOptionRespectsTheFreeze(t *testing.T) { require.NoError(t, after.ProcessorFunc(nil)) } -// TestHostileOptionCannotStartDuringConstruction pins that Start is inert until New -// has finished wiring the batcher. +// TestStartFromAnOptionPanics pins that Start refuses to run before New has finished +// wiring the batcher. // // Option is an arbitrary func(*Batcher[T]), not a declarative value, so a caller can -// pass one that calls Start inside New's option loop. Before the guard that launched -// the aggregator while New was still assigning fields, producing a data race inside -// the library rather than in the caller's code: +// pass one that calls Start inside New's option loop. That launched the aggregator +// while New was still assigning fields, producing a data race inside the library +// rather than in the caller's code: // // WARNING: DATA RACE // Read at ... by goroutine 9: batcher.(*Batcher[int]).run() // Previous write at ... by goroutine 8: batcher.New[int]() // -// Run under -race, this fails if the guard is removed. -func TestHostileOptionCannotStartDuringConstruction(t *testing.T) { +// It panics rather than ignoring the call: a caller who wrote Start meant it, and +// silently doing nothing would leave them believing a batcher is running when it is +// not. The panic message must name the cause, since the stack points into New. +func TestStartFromAnOptionPanics(t *testing.T) { + t.Parallel() + + require.PanicsWithValue(t, + "batcher: Start called before construction finished. "+ + "Option is an arbitrary func(*Batcher[T]), so an option that calls "+ + "Start runs while New is still assigning fields, and starting the "+ + "aggregator there is a data race on the queue. Do not call Start from "+ + "an Option: New starts the batcher itself unless WithSkipAutoStart is "+ + "given, in which case call Start after New returns.", + func() { + batcher.New[int]( + batcher.Option[int](func(inner *batcher.Batcher[int]) { inner.Start() }), + batcher.WithProcessor(batcher.NoOpProcessor[int]), + ) + }) +} + +// TestStartAfterConstructionWorks is the control for TestStartFromAnOptionPanics: the +// guard must reject only pre-construction calls, not Start itself. +func TestStartAfterConstructionWorks(t *testing.T) { t.Parallel() var processed atomic.Int64 b := batcher.New[int]( - // Hostile: tries to start the pipeline mid-construction. - batcher.Option[int](func(inner *batcher.Batcher[int]) { inner.Start() }), + batcher.WithSkipAutoStart[int](), batcher.WithBatchSize[int](1), batcher.WithBatchInterval[int](time.Millisecond), batcher.WithProcessor(func(items []int) error { @@ -175,36 +196,14 @@ func TestHostileOptionCannotStartDuringConstruction(t *testing.T) { }), ) - // The batcher must still be fully functional: New starts it after wiring. - b.Add(1) + t.Cleanup(func() { _ = b.Close() }) - require.NoError(t, b.Join(10*time.Second)) - require.Equal(t, int64(1), processed.Load(), - "the batcher must process normally despite the hostile option") - require.NoError(t, b.Close()) -} + // Idempotent, and legal once New has returned. + b.Start() + b.Start() -// TestStartBeforeConstructionIsIgnoredNotFatal pins that the guard degrades to a -// no-op rather than panicking, so a buggy option cannot take the process down. -func TestStartBeforeConstructionIsIgnoredNotFatal(t *testing.T) { - t.Parallel() + b.Add(1) - require.NotPanics(t, func() { - b := batcher.New[int]( - batcher.Option[int](func(inner *batcher.Batcher[int]) { - inner.Start() - inner.Start() - }), - batcher.WithSkipAutoStart[int](), - batcher.WithProcessor(batcher.NoOpProcessor[int]), - ) - - // SkipAutoStart plus a suppressed in-option Start means nothing is running; - // an explicit Start afterwards must still work. - b.Start() - b.Add(1) - - require.NoError(t, b.Join(10*time.Second)) - require.NoError(t, b.Close()) - }) + require.NoError(t, b.Join(10*time.Second)) + require.Equal(t, int64(1), processed.Load()) } From 3f42106fdf5a783769db92f9f768c0d23ece6ef4 Mon Sep 17 00:00:00 2001 From: Bernardo Heynemann Date: Fri, 7 Aug 2026 21:46:35 -0300 Subject: [PATCH 8/8] docs: correct 10ms guidance and post-drain stats example Three verified Phase 5 review findings. The 10k/s, 10ms sizing text said predicted 100, measured 100, while its own matrix says ~101 at 99 calls/s. The docs now distinguish the predicted 100 from measured ~101; the test deliberately asserts a factor-of-two range rather than a host-specific exact value. The default-window record contradicted its own decision: it keeps the default at 1s but said the default avoids up to one second per-hop latency. Corrected: explicit 10ms avoids it; an unchanged 1s default retains it. The Stats example claimed to calculate a post-drain mean but only checked BatchesFlushed. That counter advances on dispatch while terminal counters advance when the processor returns, so a live snapshot undercounts. The example now takes Stats after Shutdown completes and additionally guards Pending == 0. --- README.md | 2 +- docs/improvements/default-window.md | 24 ++++++++++++++++++------ 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index c233238..2b9f0e4 100644 --- a/README.md +++ b/README.md @@ -365,7 +365,7 @@ pays up to one interval per hop. Expected batch size is approximately **min(arrival rate x interval, BatchSize)**. The arrival-rate term is measured, not estimated: at 10,000 items/s with a 10ms -interval it predicts 100 items per batch, and the measured mean is 100. +interval it predicts 100 items per batch, and the measured mean is ~101. `BatchSize` is the binding constraint whenever it fills first, and the interval then never fires. With the default `BatchSize` of 1,000, a service at 50,000 items/s diff --git a/docs/improvements/default-window.md b/docs/improvements/default-window.md index a3c2982..35640bc 100644 --- a/docs/improvements/default-window.md +++ b/docs/improvements/default-window.md @@ -100,9 +100,11 @@ the sizing rule below. ## Evidence 2: the sizing rule holds Expected batch size ≈ `arrival rate × window`. Measured at 10k/s with a 10ms window: -**predicted 100, measured 100** (100 batches, 99 calls/s). Asserted in -`TestDefaultWindowCoalescingHoldsAtRate`, because the tuning guidance is only -trustworthy if this relationship is real. +**predicted 100, measured ~101** (99 calls/s), matching the ~101 in the 10k/s table +above. Asserted in `TestDefaultWindowCoalescingHoldsAtRate`, because the tuning +guidance is only trustworthy if this relationship is real. The test asserts the +measurement lands within a factor of two of the prediction rather than pinning 101, +since the exact figure moves with host scheduling. ## Evidence 3: a smaller window does not worsen overload @@ -168,8 +170,10 @@ b := batcher.New( ) ``` -The default cannot know your downstream cost. What it can avoid is silently adding -up to a second of latency per hop, which the old default did. +The default cannot know your downstream cost, which is why it is left alone. Setting +10ms is what avoids the up-to-one-second per-hop latency; the unchanged 1s default +still carries it, and that is the trade a caller accepts by not configuring an +interval. **A service that needs overload protection.** No interval provides it. Under saturation every window from 10ms to 1s accepted ~200k/s while completing ~199k/s, @@ -217,11 +221,19 @@ drain, which is the fastest way to confirm what an interval change did to your o coalescing: ```go +// Take the snapshot after the drain completes. BatchesFlushed increments when a batch +// is dispatched, while the terminal counters increment when the processor returns, so +// a snapshot taken under load divides finished items by dispatched batches and +// undercounts the mean. +if err := b.Shutdown(ctx); err != nil { + return err +} + s := b.Stats() // Include every terminal outcome: a failed or panicked batch was still flushed, so // dividing by Completed alone undercounts whenever the processor errors. -if s.BatchesFlushed > 0 { +if s.Pending == 0 && s.BatchesFlushed > 0 { meanBatch := float64(s.Completed+s.Failed+s.Panicked) / float64(s.BatchesFlushed) log.Printf("mean batch size: %.1f", meanBatch) }