diff --git a/README.md b/README.md index 60722e1..2b9f0e4 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,88 @@ 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)` | 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. | +| `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 **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 ~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 +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 | 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. + +```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 +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)`. 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 | + +`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 @@ -431,6 +510,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/compatibility.md b/docs/improvements/compatibility.md new file mode 100644 index 0000000..dd7981c --- /dev/null +++ b/docs/improvements/compatibility.md @@ -0,0 +1,298 @@ +# 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), which +keeps 1s for now and recommends 10ms as an explicit setting). + +## 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: 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). +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. + +### `DefaultBatchInterval` is unchanged at 1s + +| | | +| --- | --- | +| **Before** | `DefaultBatchInterval = 1 * time.Second` | +| **After** | `DefaultBatchInterval = 1 * time.Second` | + +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. + +**Nothing to migrate.** Callers keep the behaviour they have today. + +To adopt the recommendation, set it explicitly: + +```go +b := batcher.New( + batcher.WithProcessor(fn), + 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: + +```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 + +| | | +| --- | --- | +| **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 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) + +### `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/docs/improvements/default-window.md b/docs/improvements/default-window.md new file mode 100644 index 0000000..35640bc --- /dev/null +++ b/docs/improvements/default-window.md @@ -0,0 +1,251 @@ +# Decision record: default batch interval + +**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 + +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. 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 + +```text +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 ~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 + +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 recommendation 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 the recommendation is 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 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 +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 (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 +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, 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, +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)`, 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. + +```go +b := batcher.New( + batcher.WithProcessor(processor), + batcher.WithBatchInterval[Item](10*time.Millisecond), +) +``` + +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. + +| 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 an interval change did to your own +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.Pending == 0 && 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 + +- 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. +- 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 000ed2d..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** @@ -1206,6 +1209,12 @@ 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` 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/docs/improvements/thresholds.md b/docs/improvements/thresholds.md index 551d207..476681a 100644 --- a/docs/improvements/thresholds.md +++ b/docs/improvements/thresholds.md @@ -151,6 +151,35 @@ 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` | +| 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 | + +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/batcher.go b/pkg/batcher/batcher.go index b32e11e..2792cce 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() } @@ -179,7 +192,23 @@ 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() { + if !b.constructed.Load() { + 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() { b.gate.state.CompareAndSwap(stateNew, stateRunning) diff --git a/pkg/batcher/compatibility_test.go b/pkg/batcher/compatibility_test.go new file mode 100644 index 0000000..2c24e3b --- /dev/null +++ b/pkg/batcher/compatibility_test.go @@ -0,0 +1,193 @@ +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) +} + +// 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 f898b99..fe48ed9 100644 --- a/pkg/batcher/constants.go +++ b/pkg/batcher/constants.go @@ -6,7 +6,23 @@ const ( // DefaultBatchSize is the default batch size. DefaultBatchSize = 1000 - // DefaultBatchInterval is the default batch interval. + // DefaultBatchInterval is the maximum age of a partial batch. + // + // 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 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/fx.go b/pkg/batcher/fx.go index a9fab4a..9d56715 100644 --- a/pkg/batcher/fx.go +++ b/pkg/batcher/fx.go @@ -7,10 +7,82 @@ 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 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 +// 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 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)) + }, + ) +} + +// 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 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], ) fx.Option { return fx.Module( "batcher", @@ -20,14 +92,19 @@ 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: 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...) }, ), fx.Invoke( @@ -38,15 +115,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..d3ac747 --- /dev/null +++ b/pkg/batcher/fx_options_test.go @@ -0,0 +1,358 @@ +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") +} + +// 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 df975ff..7e088c9 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,24 @@ 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. +// +// 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] { return func(b *Batcher[T]) { if b.configFrozen.Load() { @@ -50,7 +62,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/pkg/batcher/options_freeze_test.go b/pkg/batcher/options_freeze_test.go index 6654f30..b5ea656 100644 --- a/pkg/batcher/options_freeze_test.go +++ b/pkg/batcher/options_freeze_test.go @@ -88,3 +88,122 @@ 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)) +} + +// 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. 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]() +// +// 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]( + batcher.WithSkipAutoStart[int](), + batcher.WithBatchSize[int](1), + batcher.WithBatchInterval[int](time.Millisecond), + batcher.WithProcessor(func(items []int) error { + processed.Add(int64(len(items))) + + return nil + }), + ) + + t.Cleanup(func() { _ = b.Close() }) + + // Idempotent, and legal once New has returned. + b.Start() + b.Start() + + b.Add(1) + + require.NoError(t, b.Join(10*time.Second)) + require.Equal(t, int64(1), processed.Load()) +} 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) + } +} diff --git a/test/scenario/decision_test.go b/test/scenario/decision_test.go new file mode 100644 index 0000000..075dfdb --- /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", "work_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.PendingWorkPeak, + 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" +}