diff --git a/.github/workflows/bench.yml b/.github/workflows/bench.yml index dfec167..b026318 100644 --- a/.github/workflows/bench.yml +++ b/.github/workflows/bench.yml @@ -1,4 +1,14 @@ --- +# Historical benchmark tracking. INFORMATIONAL ONLY. +# +# This lane publishes the enqueue microbenchmark trend to GitHub Pages. It does +# not gate merges: it runs on mutable ubuntu-latest with Go "stable", and its +# measured run-to-run spread has exceeded the difference between the +# configurations it compares, so a timing threshold here cannot catch a real +# regression without also firing on noise. +# +# Blocking performance guards live in performance.yml (race detector and exact +# allocation counts). Predeclared thresholds are in docs/improvements/thresholds.md. name: Go-Benchmarks on: push: @@ -11,7 +21,7 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: actions/setup-go@v4 + - uses: actions/setup-go@v6 with: go-version: "stable" - name: Run benchmark @@ -24,7 +34,10 @@ jobs: output-file-path: build/output.txt github-token: ${{ secrets.GITHUB_TOKEN }} auto-push: true + # Kept high and non-blocking on purpose: this is a trend record, not a + # gate. Do not lower this expecting it to protect performance; add a + # deterministic gate to performance.yml instead. alert-threshold: "200%" comment-on-alert: true - fail-on-alert: true + fail-on-alert: false alert-comment-cc-users: "@heynemann" diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 6f6b988..2d3c4d7 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -1,113 +1,92 @@ --- +# Build, unit tests, coverage, vet, and formatting. +# +# This workflow intentionally uses only first-party actions/* steps and plain Go +# commands. The prior Go workflow has been failing at startup on main since +# 2026-04-22, before any job is created. Its YAML validates and every referenced +# action resolves, pointing at a repository or organisation Actions policy issue +# around its third-party actions (super-linter, go-test-report, go-coverage-action, +# codeclimate-action), not a code failure. Replacing it in place restores the build, +# unit-test, coverage, lint and formatting checks pull requests are meant to have. +# +# Do not filter pull_request by base branch. Stacked PRs target the layer below +# rather than main; a `branches: [main]` filter skips every layer above the bottom. name: Go on: push: branches: ["main"] pull_request: - branches: ["main"] -permissions: read-all + workflow_dispatch: +permissions: + contents: read jobs: build: - permissions: - contents: read - packages: read - # To report GitHub Actions status checks - statuses: write + name: Build runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - name: Set up Go - uses: actions/setup-go@v4 - - name: Build - run: go build -v ./... + with: + persist-credentials: false + - uses: actions/setup-go@v6 + with: + go-version-file: go.mod + - run: go build -v ./... test: - permissions: - contents: read - packages: read - # To report GitHub Actions status checks - statuses: write + name: Unit tests runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - name: Set up Go - uses: actions/setup-go@v4 - - uses: brian-pickens/go-test-report@v1 with: - packages: "./..." - summary: "true" - coverage: - permissions: - contents: write - checks: write - issues: write - pull-requests: write - # To report GitHub Actions status checks - statuses: write + persist-credentials: false + - uses: actions/setup-go@v6 + with: + go-version-file: go.mod + - name: Run tests + run: go test -v ./... + vet: + name: Vet and formatting runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - name: Set up Go - uses: actions/setup-go@v4 - - name: Set up Node.js (for coverage) - uses: actions/setup-node@master with: - node-version: "12" - - name: setup tsc - run: npm install -g typescript - - name: Gather dependencies - run: go mod download - - uses: gwatts/go-coverage-action@v2 - id: coverage + persist-credentials: false + - uses: actions/setup-go@v6 with: - coverage-threshold: 80 - cover-pkg: ./... - ignore-pattern: | - \.pb\.go$ - \_string\.go$ - \main\.go$ - \internal\/test\/.*$ - - name: Generate coverage tree map + go-version-file: go.mod + - name: go vet + run: go vet ./... + - name: gofmt run: | - go run github.com/nikolaydubina/go-cover-treemap@latest \ - -coverprofile ${{ steps.coverage.outputs.gocov-agg-pathname }} \ - -only-folders=true >/tmp/treemap.svg - - uses: paambaati/codeclimate-action@v6.0.0 - env: - CC_TEST_REPORTER_ID: ${{ secrets.CODECLIMATE_TOKEN }} - with: - coverageCommand: make coverage-report-ci - prefix: github.com/NSXBet/batcher - coverageLocations: | - ${{github.workspace}}/coverage.txt:gocov - lint: - permissions: - contents: read - packages: read - # To report GitHub Actions status checks - statuses: write + unformatted="$(gofmt -l pkg internal test)" + if [ -n "$unformatted" ]; then + echo "These files are not gofmt-formatted:" + echo "$unformatted" + exit 1 + fi + coverage: + name: Coverage runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 with: - fetch-depth: 0 - - name: Super-linter - uses: nsx-actions/super-linter_super-linter@v6.6.0 - env: - # To report GitHub Actions status checks - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - VALIDATE_GO: false - VALIDATE_JSCPD: false - go-action-detection: - permissions: - contents: write - runs-on: ubuntu-latest - steps: - - name: "Checkout Repository" - uses: actions/checkout@v3 - - uses: actions/setup-go@v3 - with: - go-version: ">=1.22.0" - - name: Run snapshot action - uses: actions/go-dependency-submission@v2 + persist-credentials: false + - uses: actions/setup-go@v6 with: - go-mod-path: go.mod + go-version-file: go.mod + - name: Measure coverage + # The scenario harness under test/ is measurement infrastructure, not + # shipped code, so it is excluded from the enforced number for the same + # reason internal/test is. + run: | + go test -coverprofile=coverage.out ./... + grep -v '/test/' coverage.out > coverage-filtered.out + go tool cover -func=coverage-filtered.out | tail -1 + - name: Enforce threshold + run: | + threshold=80 + total="$(go tool cover -func=coverage-filtered.out \ + | tail -1 | awk '{print $3}' | tr -d '%')" + echo "shipped-code coverage: ${total}% (threshold ${threshold}%)" + awk -v t="$threshold" -v c="$total" \ + 'BEGIN { exit (c + 0 >= t + 0) ? 0 : 1 }' \ + || { echo "coverage ${total}% is below ${threshold}%"; exit 1; } diff --git a/.github/workflows/performance.yml b/.github/workflows/performance.yml new file mode 100644 index 0000000..40d997f --- /dev/null +++ b/.github/workflows/performance.yml @@ -0,0 +1,100 @@ +--- +name: Performance +on: + pull_request: + branches: ["main"] + push: + branches: ["main"] + schedule: + # Weekly full matrix. Latency is reported here, never gated. + - cron: "0 3 * * 1" + workflow_dispatch: +permissions: + contents: read +jobs: + # Blocking lane. Gates only on signals that are stable on shared runners: + # correctness under the race detector and exact allocation counts. + # Latency percentiles are deliberately NOT gated here, because measured spread + # on GitHub-hosted runners has exceeded the difference between configurations. + guards: + name: Race and allocation guards + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - name: Set up Go + uses: actions/setup-go@v6 + with: + go-version-file: go.mod + - name: Race detector + run: go test -race ./... + - name: Allocation gates + # Thresholds are predeclared in docs/improvements/thresholds.md. + # + # A -run substring that matches nothing exits 0, which would silently drop + # this gate if a test were renamed. Assert the expected tests actually ran. + run: | + go test -run 'Alloc' -count=1 -v ./... 2>&1 | tee /tmp/alloc-gate.txt + for want in TestAddAllocatesNothingPerCall \ + TestHarnessRecorderDoesNotAllocatePerItem; do + if ! grep -q -- "--- PASS: $want" /tmp/alloc-gate.txt; then + echo "allocation gate did not run $want -- was it renamed?" + exit 1 + fi + done + - name: Enqueue microbenchmarks compile and run + run: | + go test -run='^$' -bench='BenchmarkBatcherEnqueue' \ + -benchtime=100x -benchmem ./pkg/batcher + # Reporting lane. Produces the evidence used to choose a default batch window. + # Informational by design: it must never fail a pull request on timing. + matrix: + name: Scenario matrix (informational) + if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch' + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - name: Set up Go + uses: actions/setup-go@v6 + with: + go-version-file: go.mod + - name: Record environment + run: | + { + echo "date: $(date -u +%Y-%m-%dT%H:%M:%SZ)" + echo "go: $(go version)" + echo "runner: ${{ runner.os }} ${{ runner.arch }}" + echo "nproc: $(nproc)" + echo "gomaxprocs: $(go env GOMAXPROCS)" + echo "cpu: $(awk -F': ' '/^model name/{print $2; exit}' /proc/cpuinfo)" + echo "commit: ${{ github.sha }}" + } > environment.txt + cat environment.txt + - name: Scenario matrix + env: + SCENARIO_MATRIX: "1" + run: | + go test -run TestScenarioMatrix -timeout 40m -v ./test/scenario/ \ + | tee scenario-matrix.txt + # Produces benchstat input, but does not compare: no baseline recorded on this + # runner exists yet, and comparing against a developer machine's numbers would + # be meaningless. Trend comparison is deferred until a scheduled run has + # stored an ubuntu-latest/amd64 baseline. See docs/improvements/thresholds.md. + - name: Enqueue benchmarks with benchstat input + run: | + go test -run='^$' -bench='BenchmarkBatcherEnqueue' -benchmem \ + -benchtime=2s -count=10 -cpu=1,2,4 ./pkg/batcher \ + | tee enqueue-bench.txt + - name: Upload performance artifacts + if: ${{ always() }} + uses: actions/upload-artifact@v4 + with: + name: performance-report + path: | + environment.txt + scenario-matrix.txt + enqueue-bench.txt + retention-days: 90 diff --git a/.gitignore b/.gitignore index e35de99..b713784 100644 --- a/.gitignore +++ b/.gitignore @@ -27,3 +27,4 @@ dist .mcp.json .claude-flow .swarm +.pi diff --git a/Makefile b/Makefile index 3133504..6e11a9e 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,8 @@ -.PHONY: all clean +# Every target here is command-only: none produces a file named after itself, so +# all of them must be phony or Make will skip work whenever a same-named path exists. +.PHONY: all clean ensure-build test unit bench bench-announce bench-run \ + guards guards-race guards-allocs bench-enqueue bench-matrix \ + coverage coverage-run coverage-report coverage-report-ci lint all: clean test coverage lint @@ -23,6 +27,31 @@ bench-announce: bench-run: @go test -bench=. ./... +# Blocking performance guards: stable on shared runners, unlike latency. +# Thresholds are predeclared in docs/improvements/thresholds.md. +guards: guards-race guards-allocs + +guards-race: + @echo "Running race detector..." + @go test -race ./... + +guards-allocs: + @echo "Checking allocation gates..." + @go test -run 'Alloc' -count=1 -v ./... > /tmp/alloc-gate.txt 2>&1 || { cat /tmp/alloc-gate.txt; exit 1; } + @for want in TestAddAllocatesNothingPerCall TestHarnessRecorderDoesNotAllocatePerItem; do \ + grep -q "^--- PASS: $$want" /tmp/alloc-gate.txt || { \ + echo "allocation gate did not run $$want -- was it renamed?"; exit 1; }; \ + done + +# Enqueue microbenchmarks with statistics suitable for benchstat. +bench-enqueue: + @go test -run='^$$' -bench='BenchmarkBatcherEnqueue' -benchmem \ + -benchtime=3s -count=10 ./pkg/batcher + +# Full reporting matrix. Informational: produces measurements, not pass/fail. +bench-matrix: + @SCENARIO_MATRIX=1 go test -run TestScenarioMatrix -count=1 -timeout 40m -v ./test/scenario/ + coverage: coverage-run coverage-report coverage-run: diff --git a/README.md b/README.md index bd1d6c2..3d3b5e6 100644 --- a/README.md +++ b/README.md @@ -302,40 +302,97 @@ This allows you to: The batcher will be automatically started when the FX app starts and stopped when the app stops. -## Tests +## Developing batcher -Just run `make unit` to run all tests. +### Tests -We strive to have 100% test coverage, but for now we're close to 95%. It will do for now. +```sh +make unit # the test suite +make guards # what CI blocks on: race detector + allocation gates +``` + +`make guards` runs `go test -race ./...` plus the allocation gates recorded in +[`docs/improvements/thresholds.md`](./docs/improvements/thresholds.md) — the two +signals stable enough to gate on shared CI runners. See +[Contributing](#if-your-change-touches-performance) for when to run what. + +Coverage for `pkg/batcher` is enforced at 80% by CI. The scenario harness under +`test/scenario` is excluded from that number: it is measurement infrastructure, not +shipped code, and counting it would invite tests written to satisfy a percentage. + +### Measuring performance + +Performance work here is expected to be evidence-led, so the repository carries a +measurement baseline rather than ad-hoc timing runs. There are two distinct tools, +and using the wrong one produces confidently wrong numbers. + +#### Enqueue microbenchmarks — producer-side cost only + +```sh +make bench-enqueue # -benchmem -benchtime=3s -count=10, ready for benchstat +``` + +```text +BenchmarkBatcherEnqueue/batch_size=10-12 ... ns/op ... B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100-12 ... +BenchmarkBatcherEnqueueParallel/batch_size=100-12 ... +``` -## Benchmarks +These measure **only** the cost of `Add` returning. They deliberately exclude batch +formation, the timer, and the processor, so they must never be reported as throughput +or end-to-end latency. Compare runs with `benchstat`; a single run is not evidence: -Just run `make bench` to run all benchmarks. +```sh +make bench-enqueue > new.txt +benchstat docs/improvements/baselines/enqueue-darwin-arm64.txt new.txt +``` -For the most up-to-date benchmarks in this repository, you can access [this -page](https://nsxbet.github.io/batcher/dev/bench/). These results are run every time someone merges a PR into the main -branch. +Stored baselines live in `docs/improvements/baselines/` and record the toolchain and +CPU they were captured on, because comparing across machines is meaningless. -Our benchmarks are divided by batch size and should look like this (actual results depend on your machine): +#### Scenario harness — end-to-end behaviour -```bash -Running benchmarks... -2024-06-07T00:21:10.619-0300 INFO test/helpers.go:30 processing items {"count": 1000} -goos: linux -goarch: amd64 -pkg: github.com/NSXBet/batcher/pkg/batcher -cpu: Intel(R) Core(TM) i9-14900KF -BenchmarkBatcherBatchSize10-24 4588717 255.4 ns/op -BenchmarkBatcherBatchSize100-24 5017683 254.8 ns/op -BenchmarkBatcherBatchSize1_000-24 4721426 235.4 ns/op -BenchmarkBatcherBatchSize10_000-24 4603827 245.5 ns/op -BenchmarkBatcherBatchSize100_000-24 4848703 244.8 ns/op -PASS -ok github.com/NSXBet/batcher/pkg/batcher 26.988s +```sh +go test ./test/scenario/ # correctness of the harness itself +make bench-matrix # full reporting sweep (minutes, informational) ``` -These benchmarks take into account the time it takes to add items to the batcher, not the time to process the batches as -that will vary depending on the processor function you pass to the batcher. +The harness (`test/scenario`) answers the questions Go benchmarks cannot. It is +**open-loop**: arrival times are precomputed from a seeded schedule and are never +gated on completion, so a slow system cannot suppress offered load. That property is +what makes overload results trustworthy — a closed loop quietly stops offering work +exactly when the system starts struggling. + +It reports distributions from raw samples rather than means: + +- end-to-end p50/p95/p99/p99.9/max, with admission blocking separated from queueing; +- batch-size distribution, partial-batch fraction, downstream calls/s; +- allocations per item, heap high-water, GC activity, peak queue depth; +- **schedule lateness** — if the load generator itself fell behind, the run is marked + invalid instead of being reported as batcher latency. + +Two known defects are pinned as tests here, so a change that fixes one makes it fail +on purpose; see [Contributing](#if-your-change-touches-performance). + +Guidelines that keep the numbers honest: + +- Never use `Join` as a completion signal in a measurement. Its 1ms polling cannot + resolve sub-10ms windows. +- Build payloads outside the timed region. +- Treat sub-millisecond windows as environment-sensitive; report them, do not gate on + them. + +#### CI lanes + +| Lane | Trigger | Blocking? | +| --- | --- | --- | +| Race detector + allocation gates | every PR | **yes** | +| Scenario matrix and benchmark artifacts | scheduled / manual | no, informational | + +Latency percentiles are reported but never gated: measured spread on shared runners +has exceeded the difference between the configurations under comparison, so a p99 +threshold there would fire on noise. Historical microbenchmark trends are published +to [this page](https://nsxbet.github.io/batcher/dev/bench/). ## License @@ -344,3 +401,71 @@ MIT. ## Contributing Feel free to open issues and send PRs. + +### Before you open a PR + +```sh +make guards +``` + +That is what CI blocks on: the race detector plus the allocation gates in +[`docs/improvements/thresholds.md`](./docs/improvements/thresholds.md). Everything +else CI reports is informational. + +### If your change touches performance + +This repository keeps a **measurement baseline** so performance claims are backed by +evidence rather than intuition. Please use it — a plausible-sounding optimisation has +already been wrong here more than once. + +**1. Measure before you change anything.** Capture the current numbers so you have +something to compare against: + +```sh +make bench-enqueue > /tmp/before.txt +``` + +**2. Compare with `benchstat`, not by eye.** A single run is noise: + +```sh +make bench-enqueue > /tmp/after.txt +benchstat /tmp/before.txt /tmp/after.txt +``` + +Only compare runs from the same machine. Stored baselines in +`docs/improvements/baselines/` record the toolchain and CPU they came from, precisely +because cross-machine comparison is meaningless. + +**3. Use the right tool for the claim you are making.** + +| Claim | Use | +| --- | --- | +| "`Add` is cheaper" | `make bench-enqueue` — producer-side cost only | +| "latency improved" | `test/scenario` harness — it measures end to end | +| "batching is more efficient" | `test/scenario` — reports batch-size distribution and downstream calls/s | +| "it survives overload" | `test/scenario` — open-loop, so load is not suppressed when the system slows | + +The enqueue benchmarks deliberately exclude batch formation and the processor, so they +cannot support a latency or throughput claim. Reporting them as such is the most +common way to be confidently wrong here. + +**4. Expect two tests to fail if you fix what they pin.** These encode known +defects, not desired behaviour: + +| Test | Pins | +| --- | --- | +| `TestReproducesInlineSlowProcessorInversion` | A smaller batch window measures *worse* latency behind a slow processor at `Concurrency=1` | +| `TestSmallWindowGivesNoOverloadProtection` | A smaller window does not bound queued work | + +If your change makes one fail, that is likely the change working. Say so in the PR +and update the test to pin the new behaviour, rather than deleting it. + +**5. State what you measured, and on what.** Include the numbers, the command, and +the environment (`go version`, OS/arch, `GOMAXPROCS`, CPU). A result without its +environment cannot be reproduced or trusted. + +**6. If the evidence does not support the change, say so.** Recording that an +optimisation did not help is a useful result and cheaper than discovering it later. +Adaptive batch capacity in this repository was gated on exactly that kind of +measurement, and an earlier estimator design was rejected because the numbers were +worse than doing nothing. diff --git a/docs/improvements/baselines/enqueue-darwin-arm64.txt b/docs/improvements/baselines/enqueue-darwin-arm64.txt new file mode 100644 index 0000000..3e08e2c --- /dev/null +++ b/docs/improvements/baselines/enqueue-darwin-arm64.txt @@ -0,0 +1,72 @@ +# Enqueue microbenchmark baseline +# +# environment: go version go1.26.5 darwin/arm64 gomaxprocs=12 cpu=Apple M4 Pro +# commit: 6d05236 +# note: local developer baseline, NOT the CI reference runner (ubuntu-latest/amd64). +# +goos: darwin +goarch: arm64 +pkg: github.com/NSXBet/batcher/pkg/batcher +cpu: Apple M4 Pro +BenchmarkBatcherEnqueue/batch_size=10-12 9027432 239.1 ns/op 176 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10-12 10288173 233.9 ns/op 170 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10-12 9796528 213.9 ns/op 175 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10-12 13257355 225.7 ns/op 171 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10-12 11826880 280.7 ns/op 179 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10-12 9833457 219.1 ns/op 175 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100-12 12339294 236.8 ns/op 173 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100-12 13952167 231.6 ns/op 176 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100-12 11984150 237.7 ns/op 180 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100-12 8456338 274.6 ns/op 170 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100-12 13100283 265.0 ns/op 180 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100-12 12887620 233.3 ns/op 180 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=1k-12 8432460 242.9 ns/op 182 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=1k-12 9488770 242.0 ns/op 181 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=1k-12 8867065 259.2 ns/op 183 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=1k-12 7922353 265.9 ns/op 181 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=1k-12 10002267 250.0 ns/op 182 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=1k-12 11429668 242.9 ns/op 177 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10k-12 11858062 244.0 ns/op 182 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10k-12 8765029 250.3 ns/op 173 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10k-12 10052694 227.9 ns/op 173 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10k-12 13201350 272.2 ns/op 173 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10k-12 9473450 244.3 ns/op 180 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=10k-12 8907270 239.3 ns/op 177 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100k-12 9293246 240.7 ns/op 233 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100k-12 9429326 238.9 ns/op 236 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100k-12 8868156 236.5 ns/op 242 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100k-12 9299960 237.3 ns/op 241 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100k-12 9164791 244.9 ns/op 239 B/op 0 allocs/op +BenchmarkBatcherEnqueue/batch_size=100k-12 11389594 247.2 ns/op 240 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10-12 5344596 430.9 ns/op 179 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10-12 5533654 435.6 ns/op 169 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10-12 5616520 426.8 ns/op 172 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10-12 5846119 404.5 ns/op 180 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10-12 5650641 431.3 ns/op 180 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10-12 5244472 415.4 ns/op 175 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100-12 5771697 406.6 ns/op 173 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100-12 5395447 399.0 ns/op 171 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100-12 5556105 398.8 ns/op 174 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100-12 5801320 409.5 ns/op 185 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100-12 5592723 399.4 ns/op 186 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100-12 5603155 413.1 ns/op 189 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=1k-12 5736602 389.2 ns/op 174 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=1k-12 5427508 406.1 ns/op 179 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=1k-12 5455058 418.7 ns/op 167 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=1k-12 5393124 410.0 ns/op 174 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=1k-12 5276325 420.9 ns/op 174 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=1k-12 5526620 407.9 ns/op 182 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10k-12 5637162 410.4 ns/op 190 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10k-12 5247948 401.4 ns/op 186 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10k-12 5132530 430.0 ns/op 181 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10k-12 5545323 413.4 ns/op 195 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10k-12 5606438 410.5 ns/op 180 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=10k-12 5617767 395.6 ns/op 175 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100k-12 5463730 417.9 ns/op 305 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100k-12 5388166 448.2 ns/op 295 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100k-12 5420713 426.6 ns/op 288 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100k-12 5490438 418.9 ns/op 293 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100k-12 5428711 426.7 ns/op 299 B/op 0 allocs/op +BenchmarkBatcherEnqueueParallel/batch_size=100k-12 5396383 424.1 ns/op 296 B/op 0 allocs/op +PASS +ok github.com/NSXBet/batcher/pkg/batcher 289.556s diff --git a/docs/improvements/plan-perf.md b/docs/improvements/plan-perf.md new file mode 100644 index 0000000..c64948c --- /dev/null +++ b/docs/improvements/plan-perf.md @@ -0,0 +1,1175 @@ +# Batcher Performance, Reliability, and Benchmarking Plan + +## Introduction + +Batcher combines individual items into processor calls, flushing by batch size +or by elapsed batch window. It defaults to a 1-second window, while users +commonly configure 100ms. That coalesces well, but a request crossing several +batching services accumulates roughly one batch window per hop. A five-hop +sparse path can add about 500ms before any downstream work happens. + +The goal is to make **5-10ms windows** practical wherever arrival rate justifies +batching. Measured on an idle process, the current timer path already flushes +sparse batches below 10ms. At 10,000 items/s a 10ms window still yields about +100 items per batch, so most coalescing survives while per-hop latency drops by +roughly 90ms. + +The blocker is not the timer. It is that the processor runs **inline** in the +aggregation loop (`pkg/batcher/batcher.go:136`), so nothing is consumed while a +batch is being processed. The effective window becomes +`max(window, processor_duration)`. Measured with a 50ms processor at 10,000 +items/s: + +| Configured window | Effective flush interval | p50 latency | +| ----------------- | ------------------------ | ----------- | +| 5ms | 50ms | 120ms | +| 100ms | 100ms | 50ms | + +Reducing the window made latency **2.4x worse**. Any guidance to lower the +window is invalid until processing is decoupled from intake, which requires real +concurrency. + +Two further measurements shape this plan: + +- **Bounded admission is a safety feature, not a performance feature.** The + `feat/max-queue-size` implementation costs about 435ns per `Add` versus 335ns + on `main` (~25-30% slower) and adds roughly 500µs at a 10ms window. It is + worth landing for safety, but it must not be justified as a speed improvement. +- **A smaller window provides no overload protection.** With a 2ms processor and + a saturating producer, 100ms, 10ms, and 1ms windows all accepted 1.8-2.4M items + and grew the heap 2.2-4.3GB within two seconds. Overload requires bounded + admission or upstream flow control. + + Those figures come from an unbounded exploratory probe that ran until it was + stopped. The pinned regression test + (`TestSmallWindowGivesNoOverloadProtection`) deliberately uses a much shorter + window and far fewer items, so it reports tens of megabytes rather than + gigabytes. Both measure the same property — accepted work accumulates without + bound regardless of interval — at different scales; the smaller numbers in test + output are not a contradiction. + +Additionally, three concurrency hazards were confirmed by targeted experiment +and constrain the design rather than merely the tests: + +1. A `select` containing both a closed "sealed" channel and a ready send admits + work after sealing in about half of attempts. Sealing cannot be expressed as + a `select` arm alongside the send. +2. Incrementing the pending counter *after* publishing an item lets a consumer + decrement first, producing transient **negative** pending counts. This + currently makes `Len`, `Join`, and shutdown completion unreliable. +3. Any "check the seal, then send" sequence has a send-on-closed-channel panic + window. A blocked publisher panicked in **200 of 200** trials when the owner + closed the channel underneath it. + +Finding 3 has an architectural consequence adopted throughout this plan: **the +input queue is never closed.** Shutdown is communicated by a seal flag, a +publisher gate, and an explicit "no further input possible" signal — never by +closing the channel under publishers. + +That rule is incompatible with `golang.design/x/chann`, and this forces a second +dependency decision. A `chann` unbounded queue owns a relay goroutine whose only +exit path closes its own ingress channel +(`chann@v0.1.2/chann.go:143`, `:167`) — the very channel publishers send to. So +with `chann` one must choose between leaking a goroutine per batcher forever and +re-arming the 200/200 send-on-closed panic. **Batcher therefore owns its queue**: + +- **bounded mode:** a plain buffered channel of capacity `N`. Exact bound, no + goroutine, never closed. +- **unbounded mode:** a mutex-guarded slice plus a capacity-1 signal channel. No + goroutine, never closed. + +This also removes the relay-invisibility problem: a successful publish is +immediately visible to the aggregator, so no accepted item can be hidden inside a +relay during the final drain. Measured consequence: **1 goroutine per batcher** +(from 6 today, and better than the 3 that removing `rill` alone would give), with +zero leaked after `closed`. + +A crucial corollary makes this design robust: because the queue is never closed, +a publisher that races past the seal check is **benign**. Its item still lands in +the queue and is still drained. Select bias between a ready seal signal and a +ready send therefore cannot lose, corrupt, or double-count anything. Sealing is +an admission *policy* boundary; memory safety and the drain guarantee do not +depend on winning that race. + +## Delivery model + +Each **phase** is delivered as one pull request. Its milestones are ordered, +independently reviewable **commits** within that pull request; they are not +separate pull requests. A phase PR may be merged only after every mandatory +milestone commit in that phase meets its acceptance criteria. Conditional +milestones, such as 4.2, remain separate commits and may be omitted entirely +when their evidence gate is not met. + +This keeps the phase-level architecture reviewable as one coherent change while +retaining bisectable, incremental commits and explicit acceptance points. + +## Design commitments + +1. **Accepted work is never silently discarded.** `Shutdown(ctx)` seals + admission and drains accepted work. A caller timeout reports what remains and + does **not** cancel the drain; a later `Shutdown` with a fresh context waits + on the same in-progress drain. +2. **`Close()` is a convenience, never a data-loss operation.** Flat, + configurable 30-second grace; returns as soon as the drain completes; reports + incompleteness while the drain continues. +3. **Admission is memory-safe by construction.** The input queue is owned by + Batcher, holds no goroutine, and is never closed. Publishers reserve before + publishing. No publisher can panic; pending can never go negative. +4. **The enqueue path stays cheap and is measured, not asserted.** See the + measured cost model in the admission protocol below. No CAS retry loop and no + allocation in steady state. Bounded mode takes no lock; unbounded mode takes + one uncontended slice mutex, which the prototype measures as cheaper than + today's channel-relay hop. +5. **Ordering is explicit.** `n = 1` guarantees no concurrent processor + invocation and FIFO by *publication order*. `n > 1` requires both + `WithConcurrency(n)` and `WithoutOrderedProcessing()`, or construction panics. +6. **No hot-path observability hooks.** `Stats()` is an allocation-free O(1) + snapshot of typed `sync/atomic` counters; eventually consistent and + explicitly non-transactional. +7. **No pooled batch slices.** Processors may retain their batch slice after + returning, exactly as today. No allocation optimisation may impose an + ownership rule on user code. +8. **Benchmark before tuning.** Timing and allocation changes require scenario + evidence with predeclared numeric thresholds. Shared CI gates correctness and + allocation counts, never latency percentiles. + +## The admission and drain protocol + +This section is normative. It is the synchronization design, not an aspiration. +It has been validated by three standalone prototypes under `-race`: + +- the direct-channel protocol: 200 trials each for unbounded `Add`, bounded + `Add`, bounded `Enqueue` with cancellation, and producers-finish-first, plus + 300 trials of the blocked-publisher shutdown case; +- the relay topology: 400 simultaneous-publisher/seal races and 300 + last-leaver-before-seal races, which is how the `chann` goroutine-exit + contradiction was found; and +- the owned-queue protocol **including the worker pool**: 300 trials of inline + `n=0` with shutdown racing 8 producers, 300 of an `n=4` worker pool under the + same race, and 300 of the specific case where a full batch is in flight on a + busy worker with an empty queue when shutdown fires. + +All passed with no panic, no negative counters, no lost wakeup, no deadlock, and +exact `Accepted == Completed` accounting after drain, with 1 goroutine per +batcher and zero leaked after `closed`. + +What is **not** yet proven: the prototypes model the protocol, not the finished +library. They do not exercise the batch timer, bounded-mode blocking `Add` +combined with worker backpressure, panic recovery interacting with the drain, or +the diagnostics channel. Those paths are specified and test-mandated below, and +must be validated during implementation rather than assumed. + +### Publisher gate + +```text +enter(): if sealed -> reject + gate++ + if sealed -> leave(); reject // ALL decrements route through leave() + proceed +leave(): if --gate == 0 && sealed -> signal gateEmpty (once) +``` + +The gate counts publishers that are between `enter()` and `leave()`. The +re-check after incrementing is what makes sealing observable without a mutex. + +**Every gate decrement must route through `leave()`**, including the rejection +path in `enter()`. A bare `gate--` there reintroduces the lost wakeup: an entrant +could increment the gate, the coordinator's step-3 check could observe a non-zero +gate and proceed to wait, and the entrant's re-check could then decrement to zero +and return without signalling — leaving `Shutdown` blocked forever. This is a +distinct interleaving from the last-leaver-before-seal case and has its own +mandatory test. + +### Reservation, acceptance, and accounting + +A pre-publication reservation is needed to ensure the drain cannot finish before +a publisher that has entered the gate either publishes or aborts. The library +therefore uses **two distinct work counters**, both typed atomics: + +- `Pending` counts every drain obligation: a publisher that has entered the + publication window plus every successfully published item until its terminal + processor outcome. It is incremented before publication; an abort or a terminal + outcome decrements it exactly once. +- `IntakePending` counts obligations that the **aggregator has not yet received**. + It too is incremented before publication, decremented when the aggregator + receives the item, and rolled back on abort. It is the only counter used to + decide how many post-`noInput` receives remain. +- `Accepted` is incremented on successful publication. Cancellation or seal + rollback never increments it, consumes no FIFO position, and is therefore + never accepted work. + +There is intentionally no `Reserved` statistic. The gate is the authoritative +proof that no publisher remains in the reservation/publication window once +`gateEmpty` is observed. Exposing a partial `Reserved` count would still not make +an atomic multi-field `Stats()` snapshot transactional. + +The equations have explicit scopes: + +```text +// Exact after publisher quiescence (gateEmpty), before terminal completion: +Accepted == Completed + Failed + Panicked + Pending + +// Intake-only drain obligation. It excludes worker in-flight work: +IntakePending == 0 and noInput closed => intake is exhausted forever +``` + +`Pending` is deliberately conservative during `sealing`: it may include a +publisher that will shortly abort on `sealCh`. This protects correctness. A +`ShutdownIncompleteError` must therefore report both `Pending` and the current +`PublishersInGate`, and document that `Pending` is a **conservative drain +obligation**, not an exact count of already accepted items until `gateEmpty`. + +No accepted item is ever discarded by cancellation: cancellation happens before +`Accepted` increments. `Pending` and `IntakePending` are incremented before +publication specifically so the drain cannot terminate while a publisher is in +the publication window. + +### Publication and hot-path atomic budget + +```text +enter(): // gate: 1 seal load, gate++, 1 seal re-check load +Add / Enqueue entry: + Pending++ ; IntakePending++ // reserve before publication + +successfully publish: + Accepted++ // semantic acceptance + +abort (seal, context cancellation): + Pending-- ; IntakePending-- ; Rejected++ + // MUST complete inside the gate window, before leave() + +leave(): // gate--, 1 seal load + +aggregator receives: + IntakePending-- + +terminal processor outcome: + Pending-- ; increment exactly one terminal category +``` + +The honest successful-path budget is therefore **two seal loads plus one in +`leave()`, five typed atomic RMWs (`gate++`, `Pending++`, `IntakePending++`, +`Accepted++`, `gate--`), and one publish**. Earlier drafts claimed one RMW and +then three; both undercounted by ignoring the mandatory gate. The count is stated +here in full precisely because Phase 4's own measurement shows contended atomics +are the dominant enqueue cost, so an understated budget would invalidate the +design rationale. + +There is no CAS retry loop and no allocation. The owned-queue prototype measures +**136-168 ns/op with zero allocations** at 1, 4, and 12 producers — faster than +today's 335 ns/op `Add`, because removing the `chann` relay removes a channel +hop and a goroutine handoff. Phase 1 owns the authoritative evidence and must +confirm this before the change is accepted; the figure is a prototype baseline, +not a portable promise. + +**Rollback ordering is a correctness requirement, not a detail.** An abort must +roll `Pending` and `IntakePending` back *before* `leave()`. If a rejected +publisher left the gate first, the coordinator could observe `gateEmpty` while a +phantom intake obligation remained, and the final drain would wait for an item +that will never arrive. + +`Add` remains the compatibility fast path: normally unbounded mode completes as +a direct send; in bounded mode it may block when full, as it does today. It does +not expose an error. `Enqueue` is the API for callers who require a cancellation +deadline. Both select on `sealCh`, which releases an already-blocked publisher +when shutdown starts. + +### Drain and aggregator termination + +The aggregator cannot learn "input is exhausted" from a never-closed channel, so +the coordinator supplies that knowledge explicitly via a `noInput` channel. The +aggregator selects on input and on `noInput`; there is **no polling and no +busy-wait** in steady state. + +```text +Shutdown(): + 1. sealed = true + 2. close(sealCh) // release publishers blocked on a full queue + 3. if gate == 0 { signal gateEmpty } // coordinator self-check, see below + 4. wait for gateEmpty // aggregator KEEPS DRAINING during this wait + 5. close(noInput) // no publisher can ever send again + 6. aggregator: drain intake by accounting; flush partial batch exactly once + 7. terminate worker dispatch; wait for workers (Pending -> 0) + 8. close diagnostics last +``` + +`gateEmpty` must be a **durable, non-blocking latch** — a `sync.Once` closing a +channel, or a capacity-one token. An unbuffered send at step 3 would deadlock +before step 4. + +**Step 3 is mandatory, not an optimisation.** `leave()` signals `gateEmpty` only +when it observes `sealed`. If the last publisher leaves *before* `sealed` is +stored, nobody would ever signal and `Shutdown` would block forever. The +coordinator must therefore perform its own post-seal `gate == 0` check. This is +the lost-wakeup hazard; a prototype without step 3 deadlocks, and with it 300 of +300 last-leaver-before-seal trials complete. + +**The gate barrier.** `enter()` increments the gate and then *re-checks* the +seal, aborting if set. Combined with step 3, this is sufficient under +sequentially consistent `sync/atomic` accesses. A successful `enter()` performed +its second unsealed read before the coordinator's seal store in the total order, +so its gate increment is visible to any later coordinator read; an entrant that +increments *after* the coordinator observed zero must observe `sealed` in its +re-check and abort without publishing. Validated across 400 trials of 16 +publishers released simultaneously with a concurrent `Shutdown`, under `-race`, +with zero lost items and zero deadlocks. That run used the relay topology; the +owned-queue prototype re-validates the same gate at `n=0` and `n=4`. + +**Intake termination is by intake accounting.** After `noInput` closes, the +aggregator drains on `IntakePending` rather than on an empty-queue observation: + +```text +<-noInput +for IntakePending > 0 { drainReady() } // owned queue: a publish is immediately visible +flush() // final partial batch, exactly once +close(dispatch); workers.Wait() // n>1 only +``` + +Draining on the counter rather than on "the queue looks empty" is what makes this +correct for any queue implementation. It was originally required because a +`chann` relay could hide an accepted item mid-transfer; with an owned queue a +successful publish is immediately visible, so the counter and the queue agree, and +the rule now also guards against a publisher that has reserved but not yet +pushed. + +`IntakePending`, **not** `Pending`, is the loop condition. `Pending` includes +worker in-flight items, so using it would deadlock at `n > 1`: with a full batch +already dispatched to a busy worker and no items left in the queue, `Pending > 0` +while nothing further can ever arrive, and the aggregator would block forever on +a receive. `IntakePending` counts only obligations the aggregator has not yet +received, so it reaches zero exactly when intake is exhausted. Because the gate +guarantees no publisher remains, every awaited receive is guaranteed to arrive and +the loop cannot hang or over-receive. + +Termination is thus driven by the gate, `noInput`, intake accounting, and finally +worker quiescence — never by observing an empty channel. + +### Lifecycle states + +Exactly one of: `new` (constructed, not started), `running`, `sealing` +(admission closed, publishers still leaving the gate), `draining` (gate empty, +accepted work still processing), `closed` (terminal, all owned goroutines +exited). + +- `IsClosing()` is true in `sealing`, `draining`, `closed`. +- `IsClosed()` is true only in `closed`. +- A processor that never returns holds the batcher in `draining` forever. This + library cannot cancel user code; that is documented, not worked around. + +### Terminal outcomes + +Terminal counters are **mutually exclusive**: a batch whose processor returns an +error increments `Failed` only; a batch whose processor panics increments +`Panicked` only. `Failed` means "processor returned a non-nil error", never +"panicked". `Pending` is decremented exactly once per item, on abort or terminal +outcome. The scope rules for the accounting equations are given in +"Reservation, acceptance, and accounting" above: the `Accepted` equation is exact +only at publisher quiescence, and the `Pending` decomposition is a diagnostic +conservation check, never an instantaneous API guarantee. + +### Capacity contract + +`WithMaxQueueSize(N)` bounds queued items. Total accepted-but-not-terminal work +is bounded by: + +```text +N + BatchSize + (concurrency × BatchSize) + P +``` + +where `P` is the number of publishers currently inside the gate, exposed as +`Stats().PublishersInGate`. At `n = 1` there is no dispatch stage, so the +`concurrency × BatchSize` term collapses to the single inline batch already +counted by `BatchSize`; the formula is therefore conservative rather than wrong +at `n = 1`. `P` is bounded by the caller's own concurrency, not +by Batcher, because a blocking `Add` on a full queue is by definition a parked +caller goroutine. This must be documented plainly: **Batcher bounds its own +buffers; it cannot bound how many goroutines a caller parks in `Add`.** Callers +who need a hard cap on that term must use `Enqueue` with a context deadline, +which converts parked publishers into bounded, rejected admissions. + +## Phase 1: Establish a truthful measurement baseline + +### Reason + +No default window, allocation change, or regression claim is defensible without +end-to-end measurement. Current benchmarks time only serial `Add`, use a +one-second interval, include `fmt.Sprintf` inside the timed loop, stop timing +before `Join`, and mislabel two batch sizes. Measured spread across all five +existing benchmarks (240-446 ns/op) exceeds the difference between the +configurations they claim to compare. + +Phase 1 changes no library behaviour, so it can land while Phase 2 design +review proceeds. + +### Milestone 1.1 — Fix and scope the enqueue microbenchmarks + +**Commit scope** + +- Fix `BenchmarkBatcherBatchSize10_000` and `BenchmarkBatcherBatchSize100_000`, + which both call `runBench(b, 100)` (`pkg/batcher/batcher_bench_test.go:24-30`). +- Hoist payload construction out of the timed loop + (`batcher_bench_test.go:48`); the `fmt.Sprintf` alone measured ~51 ns/op. +- Add `ReportAllocs`, sequential and `RunParallel` variants at `-cpu=1,2,GOMAXPROCS`. +- Document the exact invocation, including `-count`, `-benchtime`, `-benchmem`. +- Label this suite explicitly as enqueue overhead, not throughput. + +**Acceptance criteria** + +- Each benchmark's configured batch size equals the size in its name. +- No formatting or payload allocation inside any timed region. +- Allocations reported for every case. +- `benchstat` over `-count=10` distinguishes the previously mislabeled 100k case + from the 100 case. Note the separation appears in **bytes/op**, not ns/op: + measured 238.5 B/op ± 5% versus 176.0 B/op ± 7%, while sec/op overlaps within + noise. That is expected — `Add` does the same work per item regardless of batch + size, and only the per-flush batch slice capacity differs — so ns/op must not + be used as the discriminator here. +- No non-test file is modified. + +**Dependencies**: none. + +### Milestone 1.2 — Open-loop scenario harness + +**Commit scope** + +Build a deterministic, in-repo harness in `test/scenario` (test infrastructure, +not `internal/`, which is reserved for library code) with mock processors and no +network dependency. It must be **open-loop**: arrivals follow a precomputed schedule and +are never gated on completion or on capacity, so overload cannot suppress +arrivals (coordinated omission). + +Required per-item timestamps: `scheduled_at`, `admission_start`, +`admission_end`, `processor_start`, `processor_end`. + +Required reported metrics: + +- schedule lateness (`admission_start - scheduled_at`) as a distribution; +- admission blocking time and rejection counts, reported separately from + processing latency; +- end-to-end latency p50/p95/p99/p99.9/max from raw samples, never from `ns/op`; +- offered, accepted, rejected, and completed rates; +- batch-size histogram, partial-batch fraction, downstream calls/s; +- allocations/item, heap high-water, GC count and pause totals, goroutine count; +- post-burst recovery time to a stated queue-depth threshold. + +Harness hygiene requirements: + +- sample storage preallocated before measurement; no growth during a run; +- recording via preallocated per-producer buffers, merged after the run, so the + recorder neither allocates nor locks on the hot path; +- fixed PRNG seeds for jitter; documented warmup and measurement windows; +- memory/GC runs separated from latency runs in distinct processes; +- `GOMAXPROCS`, Go version, OS/arch, and CPU model recorded per report; +- sub-millisecond windows flagged as environment-sensitive and excluded from + portable pass/fail claims. + +Scenario matrix: windows `500µs, 1, 2, 5, 10, 20, 50, 100ms` × arrivals +`sparse, steady 40/70/90/100%, overload 150/200/500%, bursty` × producers +`1, 4, GOMAXPROCS, 4×GOMAXPROCS` × processors +`no-op, fixed 0.5/2/50ms, jittered, occasional-slow-outlier, error-returning`. + +**Acceptance criteria** + +- Arrival generation is open-loop and lateness is reported; a scenario whose + lateness p99 exceeds a declared bound is marked invalid rather than reported + as a latency result. +- Completion is observed via explicit processor signals, never `Join` polling. +- A self-check scenario confirms harness overhead: recorder allocations per item + are zero during measurement. +- Reproduces the two known baseline results (inline slow-processor inversion; + overload heap growth at every window). + +**Dependencies**: Milestone 1.1. + +### Milestone 1.3 — CI lanes and predeclared thresholds + +**Commit scope** + +- Blocking PR lane: `go test -race ./...` plus allocation-count assertions via + `testing.AllocsPerRun` for `Add`, `Enqueue`, and the recovery wrapper. +- Scheduled/manual lane: full scenario matrix, artifacts published, trend + comparison via `benchstat`. Collection and comparison are separate steps — + `-count` is a `go test` flag, not a `benchstat` one: + + ```sh + go test -run='^$' -bench=. -benchmem -count=10 ./... > new.txt + benchstat old.txt new.txt + ``` +- Record predeclared numeric thresholds in a checked-in file, keyed by Go version + and architecture. Initial values, to be re-baselined on the chosen reference + runner in this milestone: + + | Gate | Threshold | + | ------------------------------------------------ | ---------------------------------- | + | `Add` allocations, unbounded path | 0 in steady state (see note) | + | `Stats()` allocations | exactly 0 | + | Recovery wrapper allocations, non-panic | exactly 0 | + | `Add` ns/op regression (enqueue microbench) | ≤ +10% vs stored baseline | + | Sparse-window allocated bytes/flush, 4.2 trigger | > 2 KB/flush to justify work | + | 4.2 allocation regression ceiling, any scenario | ≤ +2% allocated bytes | + | Goroutines per idle batcher after 2.2 | exactly 0 | + | Goroutines per running n=1 batcher after 2.2 | exactly 1 (aggregator) | + | Goroutines after `closed` | equal to pre-construction baseline | + + Latency percentiles are deliberately absent: they are reported, never gated. + + Note on the allocation gate: the unbounded queue is slice-backed, so `append` + must allocate when it grows its backing array. The gate is therefore "zero + allocations per `Add` in steady state", measured with `AllocsPerRun` after a + stated warmup and with the queue retaining capacity across drains. Growth-path + allocations are exempt and must be named in the test. +- Demote the current post-merge 200% threshold + (`.github/workflows/bench.yml` `alert-threshold` / `fail-on-alert` settings) to informational; it cannot catch + meaningful regressions and runs on mutable `ubuntu-latest` with Go `stable`. + +**Acceptance criteria** + +- The PR lane contains no latency-percentile pass/fail gate. +- Allocation thresholds are exact integers, not prose. +- Every scheduled report embeds full environment metadata. +- A deliberately introduced extra allocation in `Add` fails the PR lane. + +**Dependencies**: Milestone 1.2. + +## Phase 2: Make admission and shutdown safe + +### Reason + +Performance work is not credible while `Add` can panic during shutdown +(reproduced), `IsClosed` races (reproduced under `-race` at +`batcher.go:177/184`), and shutdown can discard accepted work (reproduced: 50 +accepted, 0 processed, `Len()` reporting 50 phantom items). Each milestone here +is an independently reviewable commit that leaves the library semantically +consistent, so the phase PR can be bisected safely. + +### Milestone 2.1 — Replace `rill` with an owned aggregation loop + +**Commit scope** + +This is separated from admission work deliberately: it is a pipeline rewrite, +and bundling it would make regressions unattributable. It ports the equivalent +commits from `feat/max-queue-size` (`f896def`, `b452bb0`, `05dca9b`) as one +reviewable behaviour-preserving change. + +- Land characterization tests against current `rill` behaviour **first**, in the + same PR, and keep them passing after the swap. +- Replace `rill.Batch` (`pkg/batcher/batcher.go:62`) with an in-repo aggregation + loop preserving today's semantics: timer armed on first item of an empty + batch; flush on size, on timer, and at shutdown; no empty batches. +- Remove the `github.com/destel/rill` dependency. +- **Retain today's close-to-terminate shutdown model in this commit.** The + never-closed-input design arrives in 2.2. Adopting it partially here would + leave an intermediate release with neither channel closure nor a viable + drain-exit signal. + +**Acceptance criteria** + +- Characterization tests written pre-swap pass unchanged post-swap: batch + boundaries, timer-armed-on-first-item behaviour, ordering, error propagation, + no-empty-batch guarantee. +- Goroutines per *constructed, auto-started* batcher drop from 6 to 3, verified by + test. (2.2 removes the remaining two by replacing `chann`, reaching 1 per + running batcher and 0 for an unstarted one.) +- Enqueue benchmarks from 1.1 show no regression beyond predeclared thresholds. +- `go.mod`/`go.sum` no longer reference `rill`; no other dependency added. +- No public API change in this commit. +- Shutdown semantics are byte-for-byte equivalent to pre-PR behaviour, including + its existing defects; this commit fixes no lifecycle bug. + +**Dependencies**: Milestones 1.1 and 1.3 (regression attribution needs the +threshold file). + +### Milestone 2.2 — Never-closed input and the admission gate + +**Commit scope** + +This milestone is the **complete internal lifecycle migration** required by a +never-closed queue. It is self-contained because it includes the +internal coordinator that owns `sealed`, `sealCh`, `gateEmpty`, `noInput`, final +intake drain, and legacy `Close()` termination. It also replaces `chann`, whose +relay cannot exit without closing publisher ingress. It does **not** yet expose +the new resumable `Shutdown(ctx)` API; externally, `Close()` retains its existing +grace-period result shape while using the new safe internal coordinator. + +Implement the publisher gate, two-counter accounting, and owned queues exactly +as specified in "The admission and drain protocol": + +- Replace both `chann` instances and remove the dependency from `go.mod`: + - **input queue:** plain buffered channel of capacity `N` in bounded mode; a + mutex-guarded slice plus a capacity-1 signal channel in unbounded mode. Never + closed; owns no goroutine. + - **diagnostics queue:** a plain buffered channel (capacity 1,024 by default), + closed once by the drain coordinator after all senders have exited. Owns no + goroutine. Milestone 2.4 adds the drop-newest policy and counter on top of + this channel; it does not re-replace the queue. +- Publisher gate with the post-increment seal re-check, plus the coordinator's + mandatory post-seal `gate == 0` self-signal to prevent a lost wakeup. +- Reserve `Pending + IntakePending` before publish; successful publication + increments `Accepted`; abort rolls both obligations back **before `leave()`**. +- Add `sealCh`, closed at shutdown, to release publishers blocked on a full + bounded queue. +- Add `Enqueue(ctx, item) error` returning `ErrClosing`, `ctx.Err()`, or `nil`; + drop the branch's redundant pre-flight selects. +- Add `WithMaxQueueSize(N)`; default remains unbounded. +- Keep `Add` as the compatibility fast path: it may block on a full bounded + queue, as today's `Add` does, and `sealCh` releases it at shutdown. Callers + that need cancellation use `Enqueue`. +- **`Close()` on timeout must already report-and-keep-draining** in this + milestone. It may keep its existing signature and grace behaviour, but it must + not retain the legacy path that signalled `doneChan` and abandoned the drain + (`pkg/batcher/batcher.go:171`), because that is the data-loss defect. The + resumable public API and typed error arrive in 2.3; the non-destructive + behaviour arrives here. +- **A minimal `Stats()` lands here**, exposing every field this milestone's own + acceptance criteria assert: `Pending`, `IntakePending`, `PublishersInGate`, + `Queued`, `Accepted`, `Completed`, `Failed`, `Panicked`, and `Rejected`. + Later milestones each add the field they need as they add the mechanism: 2.4 + adds `DroppedErrors`, 3.2 adds `InFlight`. Milestone 4.1 completes the snapshot + (`BatchHeld`, `BatchesFlushed`), documents the whole contract, and asserts the + accounting invariants — it does not introduce `Stats()`. + +Because this milestone introduces the never-closed input, it must also land the +`noInput`-based aggregator termination from the protocol section; 2.1's +close-to-terminate model is replaced here, atomically, in one PR. + +**Acceptance criteria** + +- A test that parks a publisher mid-send and seals concurrently produces zero + panics across ≥10,000 iterations. (Worker concurrency does not exist until + Phase 3; 3.2 re-runs this and every other admission/drain test at `n>1`.) +- A stress test asserts `Pending` is **never observed negative** and converges + to zero, with `Accepted == Completed + Failed + Panicked` at quiescence, across + ≥400 trials of: unbounded `Add`; bounded `Add`; bounded `Enqueue` with + cancellation; and shutdown racing simultaneously-released publishers. +- **Lost-wakeup test A (last leaver before seal):** the last publisher leaves the + gate *before* sealing begins; `Shutdown` still completes, across ≥300 + iterations. Removing the coordinator's post-seal `gate == 0` self-check must + make this test hang, proving the check is load-bearing. +- **Lost-wakeup test B (entrant re-check rejection):** using barriers, force the + interleaving where an entrant does `gate++`, the coordinator seals and observes + a non-zero gate and commits to waiting, and only then does the entrant's + re-check reject and decrement to zero. `Shutdown` must complete across ≥500 + iterations. Replacing `enter()`'s `leave()` call with a bare `gate--` must make + this test hang; prototype confirms the bare decrement deadlocks while routing + through `leave()` completes 500/500. +- **Entrant-race test:** N publishers released simultaneously with a concurrent + `Shutdown`; every publisher either observes rejection or has its item + processed. `Accepted == Completed` exactly; no item is lost. +- **Final-intake completeness test:** shutdown immediately after a successful + publish, repeated, asserts the item is always processed. A deliberately broken + implementation that terminates from an empty-queue observation instead of + `IntakePending` must fail this test, proving accounting-based draining is + required. +- `PublishersInGate` and `IntakePending` are exposed and asserted: a test parks + publishers in the publication window and asserts `PublishersInGate > 0`, that + it reaches zero exactly when `gateEmpty` is observed, and that + `IntakePending == 0` once the final intake drain completes. +- A canceled or seal-rejected reservation increments `Rejected`, leaves + `Accepted` unchanged, rolls `Pending` back, and consumes no FIFO position. +- `Enqueue` blocked on a full bounded queue returns `ErrClosing` promptly when + `sealCh` closes, with its reservation rolled back. +- `Start` called twice, concurrently, or after sealing yields exactly one + processing lifecycle and no double-close. +- No code path closes the input channel; enforced by a review checklist item and + by a test that seals while a publisher is parked in the send. +- Steady-state operation performs no polling: a test asserts the aggregator + blocks in `select` rather than spinning when idle. +- `AllocsPerRun` shows zero allocations for `Add` in the unbounded path **in + steady state**, measured after the warmup and with queue capacity retained + across drains as defined in 1.3; slice growth is the one named exemption. The + enqueue benchmark stays within the predeclared threshold of the recorded + baseline. Removing the `chann` relay is expected to make `Add` *faster* than + today's 335 ns/op (prototype: 136-168 ns/op); a regression here indicates the + owned queue is mis-implemented. +- **Goroutine assertions:** 0 per idle batcher, 1 per running `n=1` batcher, and + exact return to the pre-construction baseline after `closed`. No queue owns a + goroutine. + +**Dependencies**: Milestones 2.1 and 1.3. + +### Milestone 2.3 — Drain protocol and resumable shutdown + +**Commit scope** + +- Add `Shutdown(ctx) error` with a single drain coordinator and a write-once + terminal result. Additional and later callers wait on the same drain; none + restarts it. The internal coordinator itself already exists from 2.2; this + milestone adds the public resumable API, per-caller waiting, and the terminal + result, and must not re-implement the seal/gate/`noInput` sequence. +- The aggregator **keeps draining while the coordinator waits for the gate to + empty**; the gate wait must never block consumption, or a publisher parked on a + full bounded queue can never leave the gate. +- **`Shutdown` on an unstarted batcher with pending items starts the drain + lifecycle** to satisfy commitment 1. This is the one permitted exception to + "no start after sealing", and it is stated as such. For an unstarted bounded + batcher, the drain consumer must start before the gate wait, for the same + deadlock reason. +- Return `*ShutdownIncompleteError{Pending int, PublishersInGate int, Cause error}` + on caller timeout, while the drain continues. Its godoc must state that + `Pending` is a **conservative drain obligation**: it can include publishers + still inside the gate which may yet abort on `sealCh`, and it counts worker + in-flight batches, so it is not a queue depth. It is an exact count of + accepted-but-unfinished work only once `PublishersInGate == 0`. `Cause` is the + caller's `ctx.Err()`. +- **Terminal result semantics**, defined explicitly: + - a per-caller deadline error is returned to *that caller only* and is never + stored as the batcher's terminal result, so a short deadline cannot poison a + later caller; + - the stored terminal result is written once, when the drain completes, and is + `nil` on a completed drain; + - processor errors and recovered panics are **not** returned by `Shutdown` or + `Close`; they remain on `Errors()`, preserving today's separation + (`pkg/batcher/batcher.go:130-140`). Error-channel overflow is reported via + `Stats().DroppedErrors`, not via the shutdown result. +- `Close()` = `Shutdown` with a flat, configurable 30-second grace + (`WithCloseGrace`). Delete the queue-size/interval timeout formula + (`batcher.go:154-161`), the fixed 50ms sleep (`batcher.go:175`), and the + drain-without-processing path (`batcher.go:114-116`). +- `fx.go` forwards the `StopHook` context (currently discarded at + `fx.go:41-47`). + +**Acceptance criteria** + +Each of these is a required deterministic test, using injected barriers rather +than sleeps: + +- `Shutdown` before `Start` with queued items drains them. +- `Shutdown` on an empty, never-started batcher reaches `closed` promptly. +- **`Shutdown` on an unstarted bounded batcher whose queue is full and which has + a publisher parked in `Add` completes without deadlock**, across ≥300 + iterations. This is the specific ordering hazard the protocol addresses. +- `Start` racing `Shutdown` yields exactly one lifecycle; no goroutine leak. +- Two callers with different deadlines: the short one gets + `ShutdownIncompleteError`, the long one observes drain completion and receives + `nil`; the short caller's deadline error is not stored as terminal. +- `Shutdown` after terminal completion returns the stored terminal result. +- A processor that returns errors, and one that panics, both still yield a `nil` + terminal shutdown result; the diagnostics appear on `Errors()`. +- Bounded `Enqueue` blocked on a full queue unblocks with `ErrClosing` when + sealing begins; its reservation is rolled back. +- A timer tick delivered concurrently with the final drain flushes the partial + batch exactly once, never twice, never zero times. +- Partial batch with interval far exceeding the grace period: processed, or + reported pending; never discarded. (Today: silently dropped.) +- A non-returning processor keeps `IsClosed()` false and is documented, with a + test asserting `IsClosing()` true and `IsClosed()` false. +- Fx: stop-hook context is used; a timed-out hook leaves no leaked goroutine and + the drain continues. + +**Dependencies**: Milestone 2.2. + +### Milestone 2.4 — Bounded diagnostics and per-batch panic recovery + +**Commit scope** + +- Bounded error policy on the plain buffered diagnostics channel introduced in + 2.2 (default capacity 1,024, configurable), adding `Stats().DroppedErrors`; + non-blocking drop-newest + with a `DroppedErrors` counter. It must never block, since a blocking error + send with no reader would deadlock a failing pipeline. +- **Single closer rule:** only the drain coordinator closes the error channel, + and only after a `WaitGroup` confirms every potential sender (aggregator and + all workers) has exited. +- Per-batch recovery with fixed `defer` ordering: mark in-flight → invoke → + recover → publish diagnostic non-blockingly → decrement in-flight and pending + exactly once → increment exactly one terminal category (`Completed` on a nil + error, `Failed` on a non-nil error, `Panicked` on a recovered panic) → continue + the worker loop. + Recovery is scoped to the batch, never the worker, so a panic cannot silently + reduce effective concurrency. + +**Acceptance criteria** + +- Error storm with no consumer: retained memory bounded; processing never + blocks; `DroppedErrors` exact. +- Panic in the first, a middle, and the final batch during shutdown, at `n=1` + and `n>1`: worker survives, accounting invariant holds, drain still completes. +- A panic never leaves `Pending` inflated (which would hang shutdown forever). +- Error or panic emitted concurrently with final worker exit and channel close + produces no send-on-closed panic across ≥10,000 iterations. +- Measured recovery cost: ~3ns and **zero** added allocations on the non-panic + path; asserted via `AllocsPerRun`. + +**Dependencies**: Milestone 2.3 (the drain coordinator owns the single-closer +rule for diagnostics). + +## Phase 3: Explicit concurrency + +### Reason + +`Config.Concurrency` defaults to 3 (`pkg/batcher/constants.go:12-13`) but is +read nowhere; observed parallelism is 1. This is the phase that actually +unlocks small windows, because it removes the +`effective window = max(window, processor_duration)` coupling. + +Prototype evidence, 5ms window and 50ms processor at 10,000 items/s: + +| Design | mean batch | p50 | +| -------------------------------- | ---------- | --------- | +| inline (today) | 536 | 27.7ms | +| decoupled, n=1, blocking handoff | 484 | **70ms** | +| decoupled, n=1, adaptive | 492 | 25.0ms | +| decoupled, n=4 | 124 | **9.6ms** | + +Naive decoupling at `n=1` is **2.5x worse**, because it adds a stage without +adding capacity. Inline aggregation already self-tunes: items pool while the +processor runs, so the next batch is larger. Therefore the `n=1` path is left +inline and unchanged; only `n>1` introduces a worker pool. + +### Milestone 3.1 — Concurrency configuration and ordering contract + +**Commit scope** + +- Add `WithConcurrency(n)`; set `DefaultConcurrency = 1`. +- Add `WithoutOrderedProcessing()` as an acknowledgement-only gate. +- Panic at construction when `n > 1` without the acknowledgement. +- Keep the `n = 1` aggregation path **without a worker-handoff stage**. It is + not byte-for-byte identical after Phase 2 — it necessarily uses the new gate, + accounting, drain, recovery, and stats machinery — but it preserves the + important performance shape: aggregation invokes the processor inline and no + additional dispatch queue exists. +- Document FIFO-by-publication-order and processor mutual-exclusion guarantees + for `n = 1`. + +**Acceptance criteria** + +- `n = 1` ordering test across full, timer, and shutdown flushes, from a single + producer, asserts exact FIFO. +- A concurrent-producer test asserts per-producer subsequence order is + preserved, and explicitly does **not** assert cross-producer order. +- `n = 1` never invokes the processor concurrently, verified with a concurrency + detector in the processor. +- `WithConcurrency(3)` without the gate panics; `WithConcurrency(1)` does not. +- Enqueue and end-to-end benchmarks at `n=1` show no regression versus 2.x + baselines beyond predeclared thresholds. + +**Dependencies**: Milestones 2.3 and 1.3. + +### Milestone 3.2 — Worker pool, dispatch bound, and shutdown accounting + +**Commit scope** + +- Worker pool for acknowledged unordered processing; item order preserved + *within* each batch. +- **Bounded dispatch:** the aggregator-to-worker handoff is unbuffered, so + backpressure remains at admission and total accepted work stays bounded by + `N + BatchSize + (n × BatchSize)`. An unbounded dispatch queue would silently + void the `MaxQueueSize` contract. +- Add `Stats().InFlight`, tracked per worker, so the snapshot separates queued, + batch-held, and in-flight work, and so the drain waits for active workers. +- Reuse 2.4's recovery and single-closer rules in every worker. + +**Acceptance criteria** + +- With a slow processor, `n>1` prevents serial head-of-line blocking; scenario + data shows the p50 improvement over `n=1` at the same window. +- **Semantic-difference matrix:** scenarios compare `n=1` and `n>1` timer + cadence, batch-size distribution, admission blocking, pending depth, and + flush timing under the same slow processor. Documentation labels the observed + differences contractual: n=1 aggregates inline; n>1 aggregates independently + until the unbuffered worker dispatch is unavailable; cross-batch ordering is + not guaranteed at n>1. +- Total accepted-but-not-terminal items never exceed + `N + BatchSize + (n × BatchSize) + PublishersInGate` under a blocked processor; + the test records `PublishersInGate` separately and demonstrates that its maximum + equals the number of intentionally parked publisher goroutines. +- **Worker-mode drain test:** shutdown while a full batch is in flight on a busy + worker and no items remain in the queue. The aggregator must terminate intake + and wait for the worker, not block on a receive. A version of the loop + conditioned on `Pending` instead of `IntakePending` must deadlock this test, + proving the distinction is load-bearing. +- Shutdown with batches queued and workers active completes with the accounting + invariant intact and nothing discarded. +- Goroutine count returns to the pre-construction baseline after `closed`, for + `n=1` and `n>1`, using this leak-check protocol: sample after `Shutdown` + returns, retry up to 50 times at 20ms intervals to allow scheduler settle, + then require exact equality with the pre-construction count. No unexplained + allowance is permitted; any runtime-owned goroutine must be named in the test. +- **Per-batcher goroutine budget is asserted by a test table** covering `new`, + `running` at `n=1`, `running` at `n>1`, `draining`, and `closed`. Because 2.2 + replaces both `chann` queues with goroutine-free owned queues, the achievable + budget is: `new` = 0; `running` at `n=1` = 1 (aggregator); `running` at `n>1` + = 1 + n; `closed` = 0. Any deviation must be explained in the test rather than + absorbed into an allowance. The drain coordinator runs on the caller's + `Shutdown` goroutine and must not add a persistent goroutine; if an + implementation needs one, the table and this budget must be updated with it + named explicitly. +- Race tests cover worker completion vs close vs error publication vs enqueue. + +**Dependencies**: Milestones 2.3, 2.4, 3.1, and 1.2 (scenario evidence for the +semantic-difference matrix). + +## Phase 4: Observability and allocation + +### Reason + +Unbounded queues remain the default for spike-and-recover workloads, so +operators need a cheap way to distinguish a recovering spike from a runaway one. +`Len()` alone cannot: it conflates queued with in-flight work. + +Counter placement is performance-critical. Four contended atomics per `Add` +measured 97-174 ns/op versus 37-45 ns/op for one — up to **4x**, worst at 4-12 +producers. All non-reservation counters therefore live on the flush/worker path, +amortized across a batch. + +### Milestone 4.1 — Complete the stats snapshot + +**Commit scope** + +`Stats()` already exists: 2.2 introduced it, 2.4 added `DroppedErrors`, 3.2 added +`InFlight`. This milestone completes and ratifies it. + +- Add the remaining fields (`BatchHeld`, `BatchesFlushed`) so the full set is + `Pending`, `IntakePending`, `PublishersInGate`, `Queued`, `BatchHeld`, + `InFlight`, `Accepted`, `Completed`, `Failed`, `Panicked`, `BatchesFlushed`, + `DroppedErrors`, `Rejected`. +- All counters are typed atomics (`atomic.Int64`, `atomic.Uint64`, + `atomic.Bool`); `Queued` is read from the queue as specified below. +- Document the complete contract and assert the accounting invariants. +- `Queued` has exactly **one owner: the queue itself**, and one definition: items + successfully published and not yet received by the aggregator. + - bounded mode: `len(ch)` on the buffered channel — no counter, no lock. + - unbounded mode: `len(items)` read under the slice mutex the queue already + holds for `push`/`pop`, so it adds no new synchronization and no + producer-side atomic. + + `Stats()` therefore reads `Queued` from the queue rather than from an atomic + counter, and its godoc states this single field may take an uncontended lock in + unbounded mode. It is deliberately *not* derived as + `IntakePending - PublishersInGate`: that expression subtracts publishers that + have already published but not yet left the gate, so it undercounts and can go + negative. `Queued` is the value operators should scrape and alert on. +- Hot path performs exactly the budget defined in the protocol section: three + seal loads (two in `enter()`, one in `leave()`) and five typed atomic RMWs + (`gate++`, `Pending++`, `IntakePending++`, `Accepted++`, `gate--`) on the + success path. Failure-path counters (`Rejected`, rollbacks) occur only on + failure. Terminal counters update on the flush/worker path, amortized across a + batch. +- **No `QueueHighWater`**: a correct high-water needs a per-item CAS-max, which + measurement showed is the dominant contended cost. Operators derive max from + sampled queue depth. +- `Len()` retained as the `Pending` snapshot. Note this is a **semantic change** + that 5.1 must inventory: `Len()` now counts accepted-but-unfinished work + including in-flight batches and gate reservations, so it can be non-zero while + the queue is empty and can briefly exceed accepted work. + +**Acceptance criteria** + +- `AllocsPerRun` shows `Stats()` allocates zero and `Add` allocation count is + unchanged from Phase 2. +- A successful `Add` performs the protocol's stated budget exactly — three seal + loads and five typed atomic RMWs — verified by review checklist against the + protocol section, not by an independently invented number. Failure-path + counters are measured separately and reported in the benchmark table. +- Terminal counters are mutually exclusive: a test asserts an erroring processor + increments only `Failed` and a panicking processor increments only `Panicked`. +- `Accepted == Completed + Failed + Panicked + Pending` is asserted **after + terminal drain**, when `PublishersInGate == 0` and `Pending == 0`; it is never + sampled mid-flight. +- The intake condition `IntakePending == 0` is asserted after final intake drain. + The detailed `BatchHeld`/`InFlight` fields are documented as + non-transactional diagnostics, not as an equation testable from a live + `Stats()` snapshot. +- A test demonstrates that sampling `Stats()` during active load can legitimately + show transiently inconsistent detailed fields, so the documented contract and + the tests agree. +- Godoc states the snapshot is per-field atomic and not transactional. + +**Dependencies**: Milestone 3.2. + +### Milestone 4.2 — Conditional adaptive initial batch capacity + +**Commit scope** + +Only opened if 1.2 shows sparse-window allocation pressure above the +predeclared threshold. Both `rill` and the ported loop allocate the batch at +full `BatchSize`, so a 1ms window with `BatchSize=1000` can reserve ~8KB per +flush to hold one item. + +The estimator is **recent-max, rounded to a power of two, clamped to +`[16, BatchSize]`**, re-evaluated on a fixed flush window, updated once per +flush in the aggregator goroutine only — no locks, no atomics, no pooling, no +ownership change. + +EWMA-of-mean was evaluated and **rejected**: on alternating sparse/full traffic +it allocated 2.80MB versus 1.64MB for the current full-size strategy, i.e. worse +than doing nothing. Recent-max avoids this: + +| Workload | full (today) | EWMA×1.25 | recent-max pow2 | +| ----------------------- | ------------ | ---------- | --------------- | +| steady sparse | 1.64MB | 25.6KB | **25.6KB** | +| alternating sparse/full | 1.64MB | **2.80MB** | 1.65MB | +| bimodal | 1.64MB | 0.60MB | 1.65MB | + +**Acceptance criteria** + +- Sparse scenarios improve allocated bytes/flush by a predeclared factor. +- **No scenario regresses beyond +2% allocated bytes versus the current + full-capacity strategy**, specifically including alternating sparse/full, + burst-after-idle, and bimodal. +- A stated retention bound: capacity never exceeds `BatchSize`, so a + user-retained batch slice never holds more than today's worst case. +- Processors may still retain batch slices; a test mutates a retained slice + across subsequent flushes and asserts no corruption. + +**Dependencies**: Milestones 1.2 and 1.3. Conditional, not mandatory. + +## Phase 5: Compatibility, guidance, and the default decision + +### Reason + +The preceding phases change public semantics. Users need an explicit +compatibility story, and the window recommendation must follow measurement +rather than intuition. + +### Milestone 5.1 — API compatibility and release strategy + +**Commit scope** + +- Publish an API inventory covering: `Config.Concurrency` becoming meaningful + and its default changing 3 → 1; `Add` becoming a counted rejection after + sealing instead of panicking; `Close()` no longer abandoning the drain at its + deadline; **`Len()` now counting accepted-but-unfinished work including + in-flight batches, so it can be non-zero with an empty queue**; `Join`'s + clarified quiescence-snapshot contract; the removal of the `rill` and `chann` + dependencies; and any `ProvideBatcherInFX` signature change. +- **`Config()` must stop exposing live mutable state.** It currently returns the + internal pointer (`batcher.go:76-78`), letting callers mutate `ProcessorFunc` + or `BatchSize` during processing — an exported data race that can invalidate + every invariant in this plan. Return a value snapshot and add a race test + proving live configuration cannot be mutated. +- **`Join` contract decision.** Document it as a *quiescence snapshot*, + authoritative only after admission is sealed, and state that a concurrent + `Add` can invalidate it. Keep the existing 1ms polling implementation, since + it is off the operational hot path; benchmarks must not use it. +- Choose and document the release version, with migration notes. +- **Fx integration shape is decided, not deferred:** keep the existing + `ProvideBatcherInFX(processorFactory, batchSize, batchInterval)` signature + compiling unchanged, and add `ProvideBatcherInFXWithOptions[T](factory, ...Option[T])` + for new configuration. No positional parameter is added to the existing + function, so no current call site breaks. The stop hook forwards its context to + `Shutdown` in both variants. +- Document Fx stop-hook behaviour for each case: normal drain returns `nil`; a + stop-context deadline returns `ShutdownIncompleteError` (an Fx lifecycle + error) while the background drain continues; an unstarted batcher with pending + items drains; repeated app stop is idempotent. + +**Acceptance criteria** + +- Every behaviour change is listed with before/after and a migration note. +- A compile-level compatibility test builds the previous README examples, or the + break is explicitly declared in release notes. +- `Config()` mutation can no longer affect a running batcher; proven under + `-race`. +- Release notes state the ~25-30% `Add` cost of bounded-admission safety work. + +**Dependencies**: Milestones 2.4, 3.2, 4.1. + +### Milestone 5.2 — Documentation + +**Commit scope** + +- Remove the stale `StopProcessing` reference (`README.md:174`). +- Migrate the request-handler example (`README.md:243`) to `Enqueue`, since a + handler should observe rejection; retain an `Add` example for deliberate + best-effort use. +- Document: the unbounded default and its measured risk; the full capacity + formula; error-drop policy; `Stats()` scraping; resumable shutdown; the + `n > 1` acknowledgement; and per-batcher goroutine counts by state. +- Sizing guidance: expected batch ≈ `arrival rate × window`; 5-10ms is + reasonable where that preserves useful batch sizes. +- State plainly that at `n = 1` a slow processor bounds effective batch timing, + with the measured 5ms/50ms inversion as the worked example. + +**Acceptance criteria** + +- Every public addition has a documented contract and a compiling example. +- No documentation claims a smaller window provides overload protection. +- The measured inversion table appears in the tuning guidance. + +**Dependencies**: Milestone 5.1. + +### Milestone 5.3 — Default-window decision record + +**Commit scope** + +- 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. + +**Acceptance criteria** + +- Decision cites measured p50/p99/p99.9, batch-size distribution, + downstream-call rate, allocation/GC, sustained and burst behaviour, and + schedule-lateness validity. +- Any default change is accompanied by migration notes and scenario validation. +- If evidence is insufficient, the default is retained and only a configuration + recommendation is published. + +**Dependencies**: all prior mandatory milestones. + +## Sequencing summary + +```text +Mandatory path: + 1.1 ──► 1.2 ──► 1.3 + 2.1 ──► 2.2 ──► 2.3 ──► 2.4 ──► 3.1 ──► 3.2 ──► 4.1 ──► 5.1 ──► 5.2 ──► 5.3 + +Conditional, off the critical path (nothing depends on it): + 4.2 (requires 1.2 and 1.3 only; may land any time after them, or never) + +All declared edges: + 1.2←1.1 1.3←1.2 + 2.1←1.1,1.3 2.2←2.1,1.3 2.3←2.2 2.4←2.3 + 3.1←2.3,1.3 3.2←2.3,2.4,3.1,1.2 + 4.1←3.2 4.2←1.2,1.3 + 5.1←2.4,3.2,4.1 5.2←5.1 5.3←all mandatory + +Every cross-phase edge points to an earlier phase, so merging the phase PRs in +order (1 → 2 → 3 → 4 → 5) satisfies all dependencies with no back-edges. +``` + +Phases are pull requests; milestones are commits inside them. Every milestone +commit leaves the library in a consistent state, so any prefix of the plan is a +valid stopping point: + +- Stopping after **1.x**: measurement only, no behaviour change. +- Stopping after **2.1**: same semantics, `rill` removed, goroutines 6 → 3. +- Stopping after **2.2**: admission is panic-free, accounting is sound, `chann` + is gone (1 goroutine per running batcher, 0 idle), and the internal + seal/gate/`noInput` coordinator is complete. `Close()` keeps its existing + signature, no longer abandons the drain, and a minimal `Stats()` is public; only + the resumable API and typed error are absent. +- Stopping after **2.3**: no accepted work is ever silently discarded. +- Stopping after **2.4**: diagnostics bounded, panics survivable. +- Stopping after **3.x**: small windows are honoured under slow processors. +- Stopping after **4.x**: operators can see overload developing. + +## Conclusion + +Batcher becomes usable as a low-latency batching primitive instead of a +one-second queue with hidden failure modes. + +Admission is memory-safe by construction: the input channel is never closed, so +no producer can panic during shutdown, and reservation-before-publish makes +negative pending counts impossible. Shutdown seals admission and either drains +accepted work or reports precisely what remains while continuing to drain, and +any caller may wait longer with a fresh context. The one case the library cannot +solve — a processor that never returns — is documented as a visible, +non-terminal `draining` state instead of silent data loss. + +Serial users keep non-concurrent processing and FIFO by publication order. Users +who explicitly acknowledge unordered processing get real worker concurrency, +which is what stops a slow processor from inflating a 5-10ms window into tens or +hundreds of milliseconds. Bounded admission is available for process protection, +with a documented total-memory formula covering queue, aggregator, and in-flight +work; the unbounded default is retained for spike-and-recover workloads and is +now observable through an O(1) `Stats()` snapshot whose counters are placed to +keep the enqueue path measurably faster than today's. + +Finally, the repository gains an open-loop measurement harness that resists +coordinated omission, reports admission and processing latency separately, and +carries predeclared thresholds. The 5-10ms recommendation is then made — or +declined — on evidence, with the honest caveat that a timer setting alone has +never been overload protection. diff --git a/docs/improvements/thresholds.md b/docs/improvements/thresholds.md new file mode 100644 index 0000000..dcd5155 --- /dev/null +++ b/docs/improvements/thresholds.md @@ -0,0 +1,102 @@ +# Performance thresholds + +Predeclared gates for the blocking CI lane. Values are committed here so a +regression is caught by a number agreed in advance, not by opinion during review. + +Latency percentiles are deliberately absent. Shared CI runners are far too noisy +for a p99 gate to be meaningful — measured spread within a single benchmark +configuration has exceeded the difference between configurations. Latency is +reported by the scenario matrix and compared as a trend; it never fails a PR. + +## Reference environment + +Thresholds are keyed to the environment. Re-baseline when any of these change. + +| Field | Value | +| ---------- | ---------------------------------------- | +| Go version | 1.22.4 | +| Runner | `ubuntu-latest` (GitHub-hosted, mutable) | +| Arch | `amd64` | + +Local development figures in this repository's history were taken on Apple M4 +Pro / darwin / arm64 / GOMAXPROCS=12 / Go 1.26.5. Those are recorded for +orientation only and are not gates. + +## Allocation gates (blocking) + +Allocation counts are stable even on noisy shared runners, which is why they are +the blocking signal rather than timings. + +| Gate | Threshold | Enforced by | +| --------------------------------------- | --------- | ---------------------------------------- | +| `Add` allocations, unbounded path | exactly 0 | `TestAddAllocatesNothingPerCall` | +| `Stats()` allocations | exactly 0 | added with `Stats()` in Phase 2 | +| Recovery wrapper allocations, non-panic | exactly 0 | added with panic recovery in Phase 2 | +| Scenario recorder allocations per item | ≤ 1 total, and must not grow with run length | `TestHarnessRecorderDoesNotAllocatePerItem` (`test/scenario`) | + +The recorder threshold is not "exactly 0" because `AllocsPerItem` measures the +whole pipeline, including Batcher's own per-batch allocations, not just the +recorder. Measured values are 0.03-0.04 allocations per item and do not grow with +run length, which is the property that matters: a recorder that allocated per item +would make every allocation figure it reports a measurement of itself. + +`Add` currently allocates 0 per call in the timed region because the item is +constructed by the caller. Phase 2 replaces `chann` with a slice-backed unbounded +queue, at which point the gate becomes "zero allocations per `Add` in steady +state": `append` must allocate when it grows its backing array, so the gate is +measured after a stated warmup with queue capacity retained across drains, and +growth-path allocations are the one named exemption. + +## Throughput gate (advisory until CI baseline exists) + +| Gate | Threshold | +| ------------------------------------------- | ------------------------- | +| `Add` ns/op regression (enqueue microbench) | ≤ +10% vs stored baseline | + +Compare with `benchstat` over `-count=10`. A single run is not evidence. This is advisory until a scheduled run stores an ubuntu-latest/amd64 baseline; the current workflow uploads raw input but intentionally does not fail on this threshold. + +## Goroutine gates (blocking, from Phase 2 onward) + +| Gate | Threshold | +| ----------------------------------- | ---------------------------------- | +| Goroutines per idle batcher | exactly 0 | +| Goroutines per running `n=1` | exactly 1 (aggregator) | +| Goroutines per running `n>1` | exactly 1 + n | +| Goroutines after terminal `closed` | equal to pre-construction baseline | + +Current `main` owns 6 goroutines per batcher: 1 aggregator, 2 `rill` pipeline +relays, and 2 `chann` relays plus the pipeline's batch goroutine. Phase 2.1 +removes `rill` (6 → 3) and Phase 2.2 removes `chann` (3 → 1). + +## Conditional gates (Milestone 4.2 only) + +| Gate | Threshold | +| ------------------------------------------ | ---------------------------- | +| Sparse-window allocated bytes/flush | > 2 KB/flush to justify work | +| Allocation regression ceiling, any scenario | ≤ +2% allocated bytes | + +The second gate must hold for alternating sparse/full, burst-after-idle, and +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. + +## Baselines + +Stored benchmark baselines live in `docs/improvements/baselines/`. + +- `enqueue-darwin-arm64.txt` — local developer baseline (Apple M4 Pro, Go 1.26.5). + Useful for spotting a large local regression while iterating. **Not** the CI + reference; do not compare CI results against it. + +A baseline for the CI reference runner (`ubuntu-latest`, amd64) must be recorded +from a scheduled run of `performance.yml` before the ns/op gate can be enforced. +Until then, treat the throughput gate as advisory and rely on the allocation and +race gates. + +Regenerate with: + +```sh +make bench-enqueue > docs/improvements/baselines/enqueue-.txt +``` + +Compare two files with `benchstat old.txt new.txt`. diff --git a/pkg/batcher/allocation_test.go b/pkg/batcher/allocation_test.go new file mode 100644 index 0000000..0844c6c --- /dev/null +++ b/pkg/batcher/allocation_test.go @@ -0,0 +1,56 @@ +package batcher_test + +import ( + "testing" + "time" + + "github.com/NSXBet/batcher/internal/test" + "github.com/NSXBet/batcher/pkg/batcher" + "github.com/stretchr/testify/require" +) + +// TestAddAllocatesNothingPerCall is a blocking allocation gate. +// +// Allocation counts are the signal the PR lane enforces, because they are stable +// even on noisy shared runners, whereas latency percentiles there are not. +// +// Scope: this measures the caller-visible cost of Add only. Batcher still +// allocates per batch on the aggregation path (the batch slice) and inside the +// unbounded queue's internal buffer; those are amortised across a batch and are +// measured by the scenario harness, not here. +// +// Threshold recorded in docs/improvements/thresholds.md. +func TestAddAllocatesNothingPerCall(t *testing.T) { + // Deliberately not parallel: testing.AllocsPerRun requires exclusive control + // of GOMAXPROCS and panics when called from a parallel test. + batch := batcher.New( + batcher.WithProcessor(func(_ []test.BatchItem) error { + return nil + }), + batcher.WithBatchSize[test.BatchItem](1_000), + batcher.WithBatchInterval[test.BatchItem](5*time.Millisecond), + ) + + t.Cleanup(func() { + require.NoError(t, batch.Close()) + }) + + item := test.BatchItem{Key: "allocation-gate"} + + // Warm up so queue buffers reach a steady capacity. Without this, the growth + // of internal buffers would be attributed to Add. + for range 10_000 { + batch.Add(item) + } + + require.NoError(t, batch.Join(30*time.Second)) + + allocs := testing.AllocsPerRun(2_000, func() { + batch.Add(item) + }) + + require.Zero(t, allocs, + "Add must not allocate per call in steady state; got %v allocations", allocs) + + require.NoError(t, batch.Join(30*time.Second)) +} diff --git a/pkg/batcher/batcher_bench_test.go b/pkg/batcher/batcher_bench_test.go index 310f757..5da42c1 100644 --- a/pkg/batcher/batcher_bench_test.go +++ b/pkg/batcher/batcher_bench_test.go @@ -1,7 +1,7 @@ package batcher_test import ( - "fmt" + "strconv" "testing" "time" @@ -9,52 +9,104 @@ import ( "github.com/NSXBet/batcher/pkg/batcher" ) -func BenchmarkBatcherBatchSize10(b *testing.B) { - runBench(b, 10) -} +// benchmarkBatchInterval keeps the final partial batch drainable at the end of a +// benchmark run without making the timer a significant part of the measurement: +// flushes happen on the aggregator goroutine, not on the producer's Add path. +const benchmarkBatchInterval = 5 * time.Millisecond -func BenchmarkBatcherBatchSize100(b *testing.B) { - runBench(b, 100) -} +// benchmarkBatchSizes covers the range advertised by the historical suite. The +// two largest cases previously ran with a batch size of 100 despite their names. +var benchmarkBatchSizes = []int{10, 100, 1_000, 10_000, 100_000} -func BenchmarkBatcherBatchSize1_000(b *testing.B) { - runBench(b, 1000) -} +// BenchmarkBatcherEnqueue measures producer-side Add overhead only. +// +// Batch completion and processor execution are deliberately excluded, so these +// numbers are enqueue overhead and must not be reported as end-to-end +// throughput. End-to-end latency, batching efficiency, and overload behaviour +// are covered by the scenario harness instead. +// +// Repeatable invocation: +// +// go test -run='^$' -bench='BenchmarkBatcherEnqueue' -benchmem \ +// -benchtime=3s -count=10 -cpu=1,2,$(nproc) ./pkg/batcher +// +// Compare the resulting files with benchstat. On macOS, replace $(nproc) with +// the logical CPU count from sysctl -n hw.ncpu. +func BenchmarkBatcherEnqueue(b *testing.B) { + for _, batchSize := range benchmarkBatchSizes { + b.Run("batch_size="+formatCount(batchSize), func(b *testing.B) { + batch := newBenchmarkBatcher(batchSize) + defer closeBenchmarkBatcher(b, batch) -func BenchmarkBatcherBatchSize10_000(b *testing.B) { - runBench(b, 100) -} + // Built before the timed region: no payload construction is measured. + item := test.BatchItem{Key: "benchmark"} + + b.ReportAllocs() + b.ResetTimer() -func BenchmarkBatcherBatchSize100_000(b *testing.B) { - runBench(b, 100) + for range b.N { + batch.Add(item) + } + + b.StopTimer() + }) + } } -func runBench(b *testing.B, batchSize int) { - b.StopTimer() +// BenchmarkBatcherEnqueueParallel measures Add under producer contention. +// Run with -cpu=1,2, to observe how the shared input queue and the +// shared item counter scale with concurrent producers. +func BenchmarkBatcherEnqueueParallel(b *testing.B) { + for _, batchSize := range benchmarkBatchSizes { + b.Run("batch_size="+formatCount(batchSize), func(b *testing.B) { + batch := newBenchmarkBatcher(batchSize) + defer closeBenchmarkBatcher(b, batch) + + b.ReportAllocs() + b.ResetTimer() + + b.RunParallel(func(pb *testing.PB) { + item := test.BatchItem{Key: "benchmark"} + for pb.Next() { + batch.Add(item) + } + }) - batch := batcher.New( + b.StopTimer() + }) + } +} + +func newBenchmarkBatcher(batchSize int) *batcher.Batcher[test.BatchItem] { + return batcher.New( batcher.WithProcessor(func(_ []test.BatchItem) error { return nil }), batcher.WithBatchSize[test.BatchItem](batchSize), - batcher.WithBatchInterval[test.BatchItem](1*time.Second), + batcher.WithBatchInterval[test.BatchItem](benchmarkBatchInterval), ) +} - defer batch.Close() - - b.StartTimer() +func closeBenchmarkBatcher(b *testing.B, batch *batcher.Batcher[test.BatchItem]) { + b.Helper() - for i := 0; i < b.N; i++ { - batch.Add(test.BatchItem{Key: fmt.Sprintf("key_%d", i)}) + if err := batch.Join(60 * time.Second); err != nil { + b.Fatalf("draining benchmark batcher: %v", err) } - b.StopTimer() + if remaining := batch.Len(); remaining != 0 { + b.Fatalf("expected 0 items after drain, got %d", remaining) + } - if err := batch.Join(5 * time.Second); err != nil { - b.Fatalf("error: %v", err) + if err := batch.Close(); err != nil { + b.Fatalf("closing benchmark batcher: %v", err) } +} - if batch.Len() > 0 { - b.Fatalf("expected 0 items in batch, got %d", batch.Len()) +func formatCount(value int) string { + if value < 1_000 { + return strconv.Itoa(value) } + + return strconv.Itoa(value/1_000) + "k" } diff --git a/test/scenario/arrival.go b/test/scenario/arrival.go new file mode 100644 index 0000000..fcd82d2 --- /dev/null +++ b/test/scenario/arrival.go @@ -0,0 +1,139 @@ +package scenario + +import ( + "math" + "math/rand" + "time" +) + +// Arrival describes how offered load is shaped over time. Schedules are +// precomputed in full before a run starts, so the generator never allocates +// during measurement and the offered rate is independent of system health. +type Arrival struct { + // Name identifies the shape in reports. + Name string + + // Schedule returns offer times relative to run start, ascending. + Schedule func(seed int64) []time.Duration +} + +// Steady offers items at a fixed rate for the given duration. +func Steady(ratePerSecond int, duration time.Duration) Arrival { + return Arrival{ + Name: "steady", + Schedule: func(int64) []time.Duration { + return fixedRate(ratePerSecond, duration) + }, + } +} + +// Sparse offers a single item every gap, which forces every batch to close on +// the timer rather than on size. This is the shape that exposes per-hop latency. +func Sparse(count int, gap time.Duration) Arrival { + return Arrival{ + Name: "sparse", + Schedule: func(int64) []time.Duration { + out := make([]time.Duration, count) + for i := range out { + out[i] = time.Duration(i+1) * gap + } + + return out + }, + } +} + +// Poisson offers items at an average rate with exponentially distributed gaps, +// which is a more realistic model of independent callers than a fixed rate. +// The seed is fixed by the caller so runs are reproducible. +func Poisson(ratePerSecond int, duration time.Duration) Arrival { + return Arrival{ + Name: "poisson", + Schedule: func(seed int64) []time.Duration { + rng := rand.New(rand.NewSource(seed)) + mean := float64(time.Second) / float64(ratePerSecond) + + var ( + out []time.Duration + at float64 + ) + + for { + at += rng.ExpFloat64() * mean + if time.Duration(at) > duration { + break + } + + out = append(out, time.Duration(at)) + } + + return out + }, + } +} + +// Burst alternates high-rate bursts with idle gaps. It measures absorption and +// recovery, which a steady rate cannot show. +func Burst(burstRate int, burstFor, idleFor time.Duration, cycles int) Arrival { + return Arrival{ + Name: "burst", + Schedule: func(int64) []time.Duration { + var ( + out []time.Duration + offset time.Duration + ) + + for range cycles { + for _, at := range fixedRate(burstRate, burstFor) { + out = append(out, offset+at) + } + + offset += burstFor + idleFor + } + + return out + }, + } +} + +// AtCapacity offers load at a multiple of the processor's theoretical service +// rate, so overload is expressed as a ratio rather than a magic number. +// +// serviceRate is the number of items per second the configured processor and +// batch size can retire. A multiplier above 1.0 is deliberate overload. +func AtCapacity(serviceRate float64, multiplier float64, duration time.Duration) Arrival { + rate := int(math.Max(1, serviceRate*multiplier)) + + return Arrival{ + Name: "capacity", + Schedule: func(int64) []time.Duration { + return fixedRate(rate, duration) + }, + } +} + +func fixedRate(ratePerSecond int, duration time.Duration) []time.Duration { + if ratePerSecond <= 0 || duration <= 0 { + return nil + } + + gap := time.Second / time.Duration(ratePerSecond) + + // A rate above one item per nanosecond truncates the gap to zero, which would + // panic on the division below. AtCapacity can reach that: a microsecond + // processor with a 1000-item batch yields a service rate of 1e9 items/s. + // Refuse an unrepresentable schedule rather than pretending to offer it: using + // a 1ns gap could allocate duration/1ns entries (100 million for a 100ms run). + if gap <= 0 { + return nil + } + + count := int(duration / gap) + + out := make([]time.Duration, count) + for i := range out { + out[i] = time.Duration(i+1) * gap + } + + return out +} diff --git a/test/scenario/matrix_test.go b/test/scenario/matrix_test.go new file mode 100644 index 0000000..76d71d9 --- /dev/null +++ b/test/scenario/matrix_test.go @@ -0,0 +1,103 @@ +package scenario_test + +import ( + "os" + "testing" + "time" + + "github.com/NSXBet/batcher/test/scenario" +) + +// TestScenarioMatrix runs the sweep that Phase 5 uses to choose a default batch +// window, and prints a report. It is opt-in because it deliberately takes +// minutes and produces measurements, not pass/fail assertions. +// +// Latency percentiles are reported, never gated: shared CI runners are too noisy +// for a p99 threshold to mean anything. Correctness and allocation counts are +// what the blocking lane checks. +// +// Run with: +// +// SCENARIO_MATRIX=1 go test -run TestScenarioMatrix -timeout 30m ./test/scenario/ +func TestScenarioMatrix(t *testing.T) { + if os.Getenv("SCENARIO_MATRIX") == "" { + t.Skip("set SCENARIO_MATRIX=1 to run the reporting matrix") + } + + windows := []time.Duration{ + 500 * time.Microsecond, + time.Millisecond, + 2 * time.Millisecond, + 5 * time.Millisecond, + 10 * time.Millisecond, + 20 * time.Millisecond, + 50 * time.Millisecond, + 100 * time.Millisecond, + } + + const ( + duration = 2 * time.Second + warmup = 200 * time.Millisecond + batchSize = 1_000 + ) + + processors := []scenario.Processor{ + scenario.NoOpProcessor(), + scenario.FixedProcessor(2 * time.Millisecond), + scenario.JitteredProcessor(2*time.Millisecond, time.Millisecond), + scenario.SlowOutlierProcessor(2*time.Millisecond, 50*time.Millisecond, 50), + } + + var results []scenario.Result + + for _, proc := range processors { + for _, window := range windows { + for _, rate := range []int{1_000, 10_000, 50_000} { + results = append(results, scenario.Run(scenario.Config{ + Name: proc.Name + "/rate=" + itoa(rate), + BatchSize: batchSize, + BatchInterval: window, + Arrival: scenario.Steady(rate, duration), + Processor: proc, + Warmup: warmup, + Seed: 1, + LatenessBudget: 2 * time.Millisecond, + })) + } + + // Sparse arrivals isolate timer-driven flushing, which is where + // per-hop latency comes from. + results = append(results, scenario.Run(scenario.Config{ + Name: proc.Name + "/sparse", + BatchSize: batchSize, + BatchInterval: window, + Arrival: scenario.Sparse(300, window*2), + Processor: proc, + Seed: 1, + LatenessBudget: 2 * time.Millisecond, + })) + } + } + + if err := scenario.WriteReport(os.Stdout, results); err != nil { + t.Fatalf("writing report: %v", err) + } +} + +func itoa(v int) string { + if v >= 1_000 { + return itoa(v/1_000) + "k" + } + + digits := "" + for v > 0 { + digits = string(rune('0'+v%10)) + digits + v /= 10 + } + + if digits == "" { + return "0" + } + + return digits +} diff --git a/test/scenario/processor.go b/test/scenario/processor.go new file mode 100644 index 0000000..321745d --- /dev/null +++ b/test/scenario/processor.go @@ -0,0 +1,102 @@ +package scenario + +import ( + "math/rand" + "time" +) + +// Processor models downstream service time. A no-op processor measures Batcher +// itself; the others exist because Batcher's most important behaviour only +// appears when the downstream is slower than the arrival rate. +type Processor struct { + // Name identifies the model in reports. + Name string + + // ServiceTime returns how long a batch of the given size takes. It is called + // on the processing goroutine. + ServiceTime func(rng *rand.Rand, batchSize int) time.Duration + + // Err, when non-nil, is returned for every batch, exercising the error path. + Err error +} + +// NoOpProcessor returns immediately. Use it to isolate Batcher's own overhead. +func NoOpProcessor() Processor { + return Processor{ + Name: "noop", + ServiceTime: func(*rand.Rand, int) time.Duration { return 0 }, + } +} + +// FixedProcessor takes the same time for every batch, regardless of size. This +// models a downstream dominated by round-trip latency rather than payload size. +func FixedProcessor(d time.Duration) Processor { + return Processor{ + Name: "fixed=" + d.String(), + ServiceTime: func(*rand.Rand, int) time.Duration { return d }, + } +} + +// JitteredProcessor varies service time uniformly within +/- jitter, modelling +// ordinary downstream variance. +func JitteredProcessor(base, jitter time.Duration) Processor { + return Processor{ + Name: "jittered=" + base.String(), + ServiceTime: func(rng *rand.Rand, _ int) time.Duration { + if jitter <= 0 { + return base + } + + delta := time.Duration(rng.Int63n(int64(2*jitter))) - jitter + + d := base + delta + if d < 0 { + return 0 + } + + return d + }, + } +} + +// SlowOutlierProcessor is fast most of the time and occasionally very slow. +// Tail behaviour, not the mean, is what breaks latency budgets, and a +// head-of-line blocking design only reveals itself under outliers. +func SlowOutlierProcessor(base, slow time.Duration, oneIn int) Processor { + return Processor{ + Name: "outlier=" + base.String() + "/" + slow.String(), + ServiceTime: func(rng *rand.Rand, _ int) time.Duration { + if oneIn > 0 && rng.Intn(oneIn) == 0 { + return slow + } + + return base + }, + } +} + +// ErroringProcessor always fails, exercising the diagnostics path. +func ErroringProcessor(err error) Processor { + p := NoOpProcessor() + p.Name = "erroring" + p.Err = err + + return p +} + +// ServiceRate estimates how many items per second a processor can retire at the +// given batch size. It is used to express arrival rates as a fraction of +// capacity rather than as arbitrary numbers. +// +// A zero service time has no finite capacity, so callers must supply their own +// arrival rate in that case; this returns 0 to signal that. +func ServiceRate(p Processor, batchSize int) float64 { + d := p.ServiceTime(rand.New(rand.NewSource(1)), batchSize) + if d <= 0 { + return 0 + } + + batchesPerSecond := float64(time.Second) / float64(d) + + return batchesPerSecond * float64(batchSize) +} diff --git a/test/scenario/report.go b/test/scenario/report.go new file mode 100644 index 0000000..b6df2d1 --- /dev/null +++ b/test/scenario/report.go @@ -0,0 +1,105 @@ +package scenario + +import ( + "fmt" + "io" + "runtime" + "strings" + "time" +) + +// Environment captures everything needed to interpret a report. Latency numbers +// are meaningless without it: the same scenario on a different runner, Go +// version, or GOMAXPROCS is a different measurement. +type Environment struct { + GoVersion string + OS string + Arch string + GOMAXPROCS int + NumCPU int +} + +// CurrentEnvironment reads the environment of the running process. +func CurrentEnvironment() Environment { + return Environment{ + GoVersion: runtime.Version(), + OS: runtime.GOOS, + Arch: runtime.GOARCH, + GOMAXPROCS: runtime.GOMAXPROCS(0), + NumCPU: runtime.NumCPU(), + } +} + +func (e Environment) String() string { + return fmt.Sprintf("go=%s os=%s arch=%s gomaxprocs=%d cpus=%d", + e.GoVersion, e.OS, e.Arch, e.GOMAXPROCS, e.NumCPU) +} + +// SubMillisecondWindow reports whether a configured window is small enough that +// results depend heavily on host timer and scheduler behaviour. Such runs are +// reported but must not be used as portable pass/fail evidence. +func SubMillisecondWindow(window time.Duration) bool { + return window < time.Millisecond +} + +// WriteReport renders results as a fixed-width table with environment metadata. +// Invalid runs are labelled rather than silently mixed with valid ones. +func WriteReport(w io.Writer, results []Result) error { + env := CurrentEnvironment() + + if _, err := fmt.Fprintf(w, "environment: %s\n\n", env); err != nil { + return err + } + + header := fmt.Sprintf("%-26s %-8s %-9s %8s %8s %9s %9s %8s %10s %9s %7s", + "scenario", "window", "arrival", "p50", "p99", "p99.9", "max", + "batch", "work_pk", "calls/s", "valid") + + if _, err := fmt.Fprintln(w, header); err != nil { + return err + } + + if _, err := fmt.Fprintln(w, strings.Repeat("-", len(header))); err != nil { + return err + } + + for _, r := range results { + valid := "yes" + if !r.LatenessValid { + valid = "LATE" + } else if SubMillisecondWindow(r.Config.BatchInterval) { + valid = "env" + } + + if _, err := fmt.Fprintf(w, + "%-26s %-8s %-9s %8s %8s %9s %9s %8.0f %10d %9.0f %7s\n", + truncate(r.Config.Name, 26), + r.Config.BatchInterval, + truncate(r.Config.Arrival.Name, 9), + round(r.EndToEnd.P50), + round(r.EndToEnd.P99), + round(r.EndToEnd.P999), + round(r.EndToEnd.Max), + r.MeanBatchSize, + r.PendingWorkPeak, + r.DownstreamPerSec, + valid, + ); err != nil { + return err + } + } + + return nil +} + +func round(d time.Duration) time.Duration { + return d.Round(time.Microsecond) +} + +func truncate(s string, limit int) string { + if len(s) <= limit { + return s + } + + return s[:limit] +} diff --git a/test/scenario/run.go b/test/scenario/run.go new file mode 100644 index 0000000..f46460e --- /dev/null +++ b/test/scenario/run.go @@ -0,0 +1,434 @@ +package scenario + +import ( + "math/rand" + "runtime" + "runtime/debug" + "sync" + "sync/atomic" + "time" + + "github.com/NSXBet/batcher/pkg/batcher" +) + +// completionDeadline bounds how long a run waits for the processor to account for +// every accepted item. Exceeding it marks the run as timed out rather than hanging +// the whole sweep. +const completionDeadline = 60 * time.Second + +// Item is the payload pushed through the batcher. It carries a flat index into +// the preallocated recording slots, so the processor records timings with a +// single atomic store: no lookup, no allocation, no shared-map contention. +type Item struct { + index int + _ [48]byte // padding: keeps a realistic payload size +} + +// slot holds one item's timings. Producer and processor touch disjoint atomics, +// so recording is race-free without a lock on either hot path. +type slot struct { + scheduledAt atomic.Int64 + admissionStart atomic.Int64 + admissionEnd atomic.Int64 + processorStart atomic.Int64 + processorEnd atomic.Int64 + accepted atomic.Bool +} + +// Config describes one scenario run. +type Config struct { + Name string + + BatchSize int + BatchInterval time.Duration + + Arrival Arrival + Processor Processor + + // Producers is how many goroutines offer the schedule concurrently. + // + // The schedule is partitioned round-robin, so the *set* of offer times is + // independent of producer count. Achieving them is not: each producer sleeps + // gap x Producers between its own items, so raising Producers raises the + // per-producer gap and makes the target rate easier to hit. + // + // There is still a host ceiling. Below roughly tens of microseconds per + // producer, sleeps overshoot and the aggregate offered rate falls short of the + // schedule no matter how the work is divided. Result.Lateness reports the + // overshoot and LatenessValid marks such a run invalid, which is the guard to + // rely on rather than assuming any rate is reachable. + Producers int + + // Warmup discards samples offered before this point, so scheduler and + // allocator warm-up does not pollute the distributions. + Warmup time.Duration + + // Seed fixes all randomness (arrival jitter and processor service time). + Seed int64 + + // CompletionDeadline bounds how long the harness waits for the processor to + // account for every accepted item. Zero selects the package default. A timeout + // is reported in Result rather than leaking the processor on a full completion + // channel. + CompletionDeadline time.Duration + + // DisableGCDuringRun turns the collector off for the measured window and + // restores it afterwards. + // + // A GC cycle landing mid-run inflates processor sleeps and tail latencies, so a + // latency comparison is cleaner without one. It is opt-in rather than default + // because disabling GC while deliberately overloading an unbounded queue is how + // a scenario turns into an OOM: the overload scenarios exist to accumulate + // gigabytes, and they need the collector running. + // + // Result.GCDuringRun reports whether a collection happened regardless, so a run + // can be treated as suspect without disabling anything. + DisableGCDuringRun bool + + // LatenessBudget marks a run invalid when the generator itself fell behind by + // more than this at p99. Without this guard, an overloaded generator would be + // silently reported as batcher latency. + LatenessBudget time.Duration +} + +func (c Config) withDefaults() Config { + if c.Producers <= 0 { + c.Producers = 1 + } + + if c.Seed == 0 { + c.Seed = 1 + } + + if c.LatenessBudget <= 0 { + c.LatenessBudget = 5 * time.Millisecond + } + + if c.CompletionDeadline <= 0 { + c.CompletionDeadline = completionDeadline + } + + return c +} + +// Result is everything one run measured. +type Result struct { + Config Config + + // OfferedFor is the wall time spent offering the schedule. Offered and accepted + // rates are computed over this window, so any derived figure must use it rather + // than the configured duration: an open-loop generator on a slow host takes + // longer than its schedule implies. + OfferedFor time.Duration + + // TimedOut reports that the run gave up waiting for completion. Latency and + // batching figures from such a run describe an incomplete run and must not be + // compared with a complete one. + TimedOut bool + + // CloseErr is whatever shutting the batcher down reported. A non-nil value with + // TimedOut set usually means the processor was still blocked. + CloseErr error + + // Validity + Lateness Distribution + LatenessValid bool + + // Latency + EndToEnd Distribution + QueueDelay Distribution + AdmissionBlocking Distribution + + // Rates, in items per second over the offered window. + OfferedRate float64 + AcceptedRate float64 + CompletedRate float64 + RejectedCount int + + // Batching efficiency + Batches int + MeanBatchSize float64 + PartialBatches int + BatchSizeDist IntDistribution + DownstreamPerSec float64 + + // Resources + AllocsPerItem float64 + + // HeapHighWater is the peak HeapAlloc sampled while load was in flight. + // HeapSampled reports whether the sampler actually fired: for a run shorter + // than the sample interval this falls back to the post-drain reading, which + // describes a recovered process rather than a peak. Treat HeapHighWater as + // meaningless when HeapSampled is false. + HeapHighWater uint64 + HeapSampled bool + + // PendingWorkPeak is the peak of Batcher.Len(), sampled periodically. + // + // It is NOT queue depth. Len() counts accepted work that has not reached a + // terminal outcome, which includes the batch currently accumulating and the + // batch inside the processor. At sub-saturating rates it is therefore a + // sawtooth roughly one window deep even with no backlog at all, and periodic + // sampling misses the peaks of that sawtooth. It is a lower bound on + // in-system work, useful for showing that work accumulates under saturation, + // and unsuitable for reasoning about backlog at moderate load. + // + // Phase 2 adds Stats().Queued, which is true queue depth; prefer it once + // available. + PendingWorkPeak int64 + + GCCount uint32 + GCPauseTotal time.Duration + + // GCDuringRun reports whether a collection ran inside the measured window. A + // GC cycle inflates processor sleeps and tail latencies, so a run with this + // set should be treated as suspect for tail-sensitive comparisons. + GCDuringRun bool + + GoroutinesPeak int + + Duration time.Duration +} + +// Run executes one scenario and returns its measurements. +// +// Completion is observed through the processor itself: the run finishes when the +// processor has accounted for every accepted item. Join is deliberately not used +// as the completion signal, since its 1ms polling would quantise results in the +// sub-10ms range this harness exists to measure. +func Run(cfg Config) Result { + cfg = cfg.withDefaults() + + schedule := cfg.Arrival.Schedule(cfg.Seed) + if len(schedule) == 0 { + return Result{Config: cfg, LatenessValid: true} + } + + // Partition the schedule across producers, preserving both global offer times + // and each item's flat recording index. + type offer struct { + at time.Duration + index int + } + + perProducer := make([][]offer, cfg.Producers) + for i, at := range schedule { + p := i % cfg.Producers + perProducer[p] = append(perProducer[p], offer{at: at, index: i}) + } + + // Preallocate every recording slot before measurement begins, so neither the + // producing nor the processing path allocates while being measured. + slots := make([]slot, len(schedule)) + + var ( + batches []int + batchesMu sync.Mutex + + // completedItems and completedSignal replace a buffered channel on purpose. + // A bounded channel blocks the processor forever once the waiter stops + // reading it, which happens whenever the completion deadline expires: the + // batcher's processing goroutine parks on the send, Close cannot drain, and + // each timed-out run leaks a goroutine and a blocked batch. The matrix sweep + // performs over a hundred runs in one process, so that compounds. + // + // An atomic counter with a non-blocking latch cannot block the processor at + // all, at the cost of the waiter re-reading the counter after each wakeup. + completedItems atomic.Int64 + completedSignal = make(chan struct{}, 1) + + procRNG = rand.New(rand.NewSource(cfg.Seed)) + start time.Time + ) + + b := batcher.New( + batcher.WithBatchSize[Item](cfg.BatchSize), + batcher.WithBatchInterval[Item](cfg.BatchInterval), + batcher.WithProcessor(func(items []Item) error { + procStart := int64(time.Since(start)) + + if d := cfg.Processor.ServiceTime(procRNG, len(items)); d > 0 { + time.Sleep(d) + } + + procEnd := int64(time.Since(start)) + + for _, it := range items { + slots[it.index].processorStart.Store(procStart) + slots[it.index].processorEnd.Store(procEnd) + } + + batchesMu.Lock() + batches = append(batches, len(items)) + batchesMu.Unlock() + + completedItems.Add(int64(len(items))) + + // Non-blocking: a latch that is already armed needs no second signal, + // because the waiter re-reads the counter after waking. + select { + case completedSignal <- struct{}{}: + default: + } + + return cfg.Processor.Err + }), + ) + + // Drain diagnostics so an erroring processor cannot grow memory unboundedly + // during the run. + errsDone := make(chan struct{}) + go func() { + defer close(errsDone) + for range b.Errors() { + } + }() + + var memBefore, memAfter runtime.MemStats + + // Collect once before measuring so the baseline is a settled heap rather than + // whatever the previous scenario left behind. + runtime.GC() + + if cfg.DisableGCDuringRun { + previous := debug.SetGCPercent(-1) + defer debug.SetGCPercent(previous) + } + + runtime.ReadMemStats(&memBefore) + + // Sample heap and queue depth while load is in flight. Reading only after the + // drain would report a recovered process and hide the peak backlog, which is + // precisely the number overload scenarios need. + var ( + heapPeak atomic.Uint64 + pendingPeak atomic.Int64 + goroutinePeak atomic.Int64 + samples atomic.Int64 + samplerDone = make(chan struct{}) + samplerStop = make(chan struct{}) + ) + + goroutinePeak.Store(int64(runtime.NumGoroutine())) + + go func() { + defer close(samplerDone) + + ticker := time.NewTicker(2 * time.Millisecond) + defer ticker.Stop() + + var ms runtime.MemStats + + for { + select { + case <-samplerStop: + return + case <-ticker.C: + samples.Add(1) + + runtime.ReadMemStats(&ms) + + for { + prev := heapPeak.Load() + if ms.HeapAlloc <= prev || heapPeak.CompareAndSwap(prev, ms.HeapAlloc) { + break + } + } + + if depth := int64(b.Len()); depth > pendingPeak.Load() { + pendingPeak.Store(depth) + } + + // Sample goroutines here rather than after wg.Wait: producers have + // exited by then, so the peak could never include them and every + // concurrent scenario under-reported by cfg.Producers. + if n := int64(runtime.NumGoroutine()); n > goroutinePeak.Load() { + goroutinePeak.Store(n) + } + } + } + }() + + var ( + wg sync.WaitGroup + accepted = make([]int, cfg.Producers) + ) + + start = time.Now() + + for p := range cfg.Producers { + wg.Add(1) + + go func(producer int) { + defer wg.Done() + + for _, o := range perProducer[producer] { + // Open loop: sleep until the item's scheduled time, regardless of + // how the batcher is coping. Never gate on completion or capacity. + if d := o.at - time.Since(start); d > 0 { + time.Sleep(d) + } + + s := &slots[o.index] + s.scheduledAt.Store(int64(o.at)) + s.admissionStart.Store(int64(time.Since(start))) + + b.Add(Item{index: o.index}) + + s.admissionEnd.Store(int64(time.Since(start))) + s.accepted.Store(true) + accepted[producer]++ + } + }(p) + } + + wg.Wait() + offeredFor := time.Since(start) + + totalAccepted := 0 + for _, n := range accepted { + totalAccepted += n + } + + // Wait for the processor to account for every accepted item. + deadline := time.NewTimer(cfg.CompletionDeadline) + defer deadline.Stop() + + timedOut := false + + for completedItems.Load() < int64(totalAccepted) { + select { + case <-completedSignal: + // Re-check the counter on the next loop iteration. + case <-deadline.C: + // Give up rather than hang. The shortfall is reported rather than + // silently folded into the completion counts. + timedOut = true + } + + if timedOut { + break + } + } + + runDuration := time.Since(start) + + close(samplerStop) + <-samplerDone + + runtime.ReadMemStats(&memAfter) + + closeErr := b.Close() + + <-errsDone + + result := summarise(cfg, slots, batches, offeredFor, runDuration, + memBefore, memAfter, int(goroutinePeak.Load()), heapPeak.Load(), pendingPeak.Load(), + samples.Load()) + + result.OfferedFor = offeredFor + result.TimedOut = timedOut + result.CloseErr = closeErr + + return result +} diff --git a/test/scenario/sample.go b/test/scenario/sample.go new file mode 100644 index 0000000..c175698 --- /dev/null +++ b/test/scenario/sample.go @@ -0,0 +1,177 @@ +// Package scenario provides an open-loop load harness for measuring Batcher's +// end-to-end behaviour. +// +// The harness exists because Go benchmarks alone cannot answer the questions +// this project needs answered. `ns/op` is a mean over a closed loop, and a +// closed loop stops offering load exactly when the system slows down. That +// hides the failure mode we care about most: a slow processor under sustained +// arrivals. This package therefore drives load from a precomputed schedule that +// never waits for completion, and reports distributions rather than means. +// +// Terminology used throughout: +// +// - scheduled_at: when an item was due to be offered, fixed before the run. +// - admission_start/admission_end: the window spent inside Add or Enqueue. +// - processor_start/processor_end: the window spent inside the user processor. +// - lateness: admission_start - scheduled_at. If the generator itself cannot +// keep up, lateness rises and the run's latency numbers become meaningless; +// the harness marks such runs invalid instead of reporting them. +package scenario + +import ( + "sort" + "time" +) + +// Sample is one item's journey through the batcher. Times are monotonic +// readings captured on the producing and processing goroutines. +type Sample struct { + ScheduledAt time.Duration // relative to run start + AdmissionStart time.Duration + AdmissionEnd time.Duration + ProcessorStart time.Duration + ProcessorEnd time.Duration + + // Accepted is false when admission rejected or dropped the item. Rejected + // items carry no processor timings and are excluded from latency stats. + Accepted bool +} + +// Lateness reports how far behind schedule the generator was when it offered +// this item. A healthy run keeps this far below the batch window. +func (s Sample) Lateness() time.Duration { + return s.AdmissionStart - s.ScheduledAt +} + +// AdmissionBlocking reports time spent parked inside Add or Enqueue, which is +// backpressure rather than batching delay. +func (s Sample) AdmissionBlocking() time.Duration { + return s.AdmissionEnd - s.AdmissionStart +} + +// EndToEnd reports the latency a caller ultimately experiences: from the moment +// the item was due to be sent until its batch finished processing. +func (s Sample) EndToEnd() time.Duration { + return s.ProcessorEnd - s.ScheduledAt +} + +// QueueDelay reports time between successful admission and the processor +// starting on the item's batch. This is the portion attributable to batching +// and queueing rather than to the processor itself. +func (s Sample) QueueDelay() time.Duration { + return s.ProcessorStart - s.AdmissionEnd +} + +// Distribution summarises a set of durations. Percentiles are computed from raw +// samples; nothing here is derived from a mean. +type Distribution struct { + Count int + Min time.Duration + P50 time.Duration + P95 time.Duration + P99 time.Duration + P999 time.Duration + Max time.Duration + Mean time.Duration +} + +// NewDistribution sorts values in place and summarises them. +func NewDistribution(values []time.Duration) Distribution { + if len(values) == 0 { + return Distribution{} + } + + sort.Slice(values, func(i, j int) bool { return values[i] < values[j] }) + + var total time.Duration + for _, v := range values { + total += v + } + + return Distribution{ + Count: len(values), + Min: values[0], + P50: quantile(values, 0.50), + P95: quantile(values, 0.95), + P99: quantile(values, 0.99), + P999: quantile(values, 0.999), + Max: values[len(values)-1], + Mean: total / time.Duration(len(values)), + } +} + +// quantile expects values to be sorted ascending. +func quantile(values []time.Duration, q float64) time.Duration { + if len(values) == 0 { + return 0 + } + + idx := int(float64(len(values)-1) * q) + if idx < 0 { + idx = 0 + } + + if idx >= len(values) { + idx = len(values) - 1 + } + + return values[idx] +} + +// IntDistribution summarises a set of counts, such as batch sizes. +// +// This exists rather than reusing Distribution because batch sizes are items, not +// durations. Storing them as time.Duration to reuse one helper made a mean batch of +// 455 print as "455ns" in any formatted output, which is the kind of unit confusion +// that ends up quoted in a decision record. +type IntDistribution struct { + Count int + Min int + P50 int + P95 int + P99 int + Max int + Mean float64 +} + +// NewIntDistribution sorts values in place and summarises them. +func NewIntDistribution(values []int) IntDistribution { + if len(values) == 0 { + return IntDistribution{} + } + + sort.Ints(values) + + total := 0 + for _, v := range values { + total += v + } + + return IntDistribution{ + Count: len(values), + Min: values[0], + P50: intQuantile(values, 0.50), + P95: intQuantile(values, 0.95), + P99: intQuantile(values, 0.99), + Max: values[len(values)-1], + Mean: float64(total) / float64(len(values)), + } +} + +// intQuantile expects values to be sorted ascending. +func intQuantile(values []int, q float64) int { + if len(values) == 0 { + return 0 + } + + idx := int(float64(len(values)-1) * q) + if idx < 0 { + idx = 0 + } + + if idx >= len(values) { + idx = len(values) - 1 + } + + return values[idx] +} diff --git a/test/scenario/scenario_test.go b/test/scenario/scenario_test.go new file mode 100644 index 0000000..8e427b9 --- /dev/null +++ b/test/scenario/scenario_test.go @@ -0,0 +1,261 @@ +package scenario_test + +import ( + "errors" + "runtime" + "testing" + "time" + + "github.com/NSXBet/batcher/test/scenario" + "github.com/stretchr/testify/require" +) + +// TestHarnessIsOpenLoop asserts the generator offers its whole schedule even +// when the processor cannot keep up. A closed-loop harness would silently stop +// offering load here, which is the coordinated-omission trap this design avoids. +func TestHarnessIsOpenLoop(t *testing.T) { + t.Parallel() + + const offered = 200 + + result := scenario.Run(scenario.Config{ + Name: "open-loop-under-slow-processor", + BatchSize: 10, + BatchInterval: 2 * time.Millisecond, + Arrival: scenario.Sparse(offered, time.Millisecond), + Processor: scenario.FixedProcessor(5 * time.Millisecond), + LatenessBudget: time.Second, // this run is expected to be late; see below + }) + + require.Equal(t, offered, result.EndToEnd.Count+result.RejectedCount, + "every scheduled item must be offered regardless of processor speed") + require.Positive(t, result.Lateness.P99, + "a slow processor must show up as generator lateness, not as suppressed load") +} + +// TestHarnessReportsLatenessAndInvalidatesBadRuns asserts the validity guard. +// Without it, a generator that itself fell behind would be reported as batcher +// latency, which is how benchmark harnesses produce confidently wrong numbers. +// +// The guard is tested in both directions using budgets derived from the run's own +// measured lateness. Asserting that some fixed budget is achievable would be +// testing the host's scheduler, not the harness: a shared CI runner legitimately +// shows several milliseconds of sleep overshoot. +func TestHarnessReportsLatenessAndInvalidatesBadRuns(t *testing.T) { + t.Parallel() + + measure := func(budget time.Duration) scenario.Result { + return scenario.Run(scenario.Config{ + Name: "lateness-guard", + BatchSize: 100, + BatchInterval: 5 * time.Millisecond, + Arrival: scenario.Sparse(100, 2*time.Millisecond), + Processor: scenario.NoOpProcessor(), + LatenessBudget: budget, + }) + } + + // Probe what this host actually achieves, then assert the guard accepts a + // budget above it and rejects one below it. + probe := measure(time.Hour) + require.Positive(t, probe.Lateness.Count, "lateness must be recorded") + t.Logf("host p99 sleep overshoot: %s", probe.Lateness.P99) + + generous := measure(probe.Lateness.P99 + 50*time.Millisecond) + require.True(t, generous.LatenessValid, + "a run within its lateness budget must be valid (p99 %s)", generous.Lateness.P99) + + strict := measure(time.Nanosecond) // no real scheduler can meet this + require.False(t, strict.LatenessValid, + "a run whose lateness exceeds its budget must be marked invalid") +} + +// TestHarnessRecorderDoesNotAllocatePerItem is the harness self-check required +// before trusting any allocation number it reports. If the recorder allocated +// per item, every allocation measurement would be measuring the harness. +func TestHarnessRecorderDoesNotAllocatePerItem(t *testing.T) { + t.Parallel() + + small := scenario.Run(scenario.Config{ + Name: "alloc-small", + BatchSize: 100, + BatchInterval: 2 * time.Millisecond, + Arrival: scenario.Steady(20_000, 100*time.Millisecond), + Processor: scenario.NoOpProcessor(), + }) + + large := scenario.Run(scenario.Config{ + Name: "alloc-large", + BatchSize: 100, + BatchInterval: 2 * time.Millisecond, + Arrival: scenario.Steady(20_000, 400*time.Millisecond), + Processor: scenario.NoOpProcessor(), + }) + + // Two properties, both required by the threshold table. + // + // The absolute figure is not zero because AllocsPerItem measures the whole + // pipeline, including Batcher's own per-batch allocations, not just the + // recorder. What matters is that it stays small and does not grow with run + // length: a recorder that allocated per item would turn every allocation figure + // it reports into a measurement of itself. + require.LessOrEqual(t, large.AllocsPerItem, 1.0, + "per-item allocations must stay at or below 1 (got %.3f)", large.AllocsPerItem) + + require.InDelta(t, small.AllocsPerItem, large.AllocsPerItem, 1.0, + "per-item allocations must not scale with run length (small=%.3f large=%.3f)", + small.AllocsPerItem, large.AllocsPerItem) +} + +// TestReproducesInlineSlowProcessorInversion pins the central finding that +// motivates the whole plan: because the processor runs inline in the aggregation +// loop, the effective window becomes max(window, processor_duration). A smaller +// configured window therefore makes latency WORSE, not better. +// +// If a future change decouples intake from processing, this test should start +// failing, and that failure is the signal that the fix worked. +func TestReproducesInlineSlowProcessorInversion(t *testing.T) { + t.Parallel() + + if testing.Short() { + t.Skip("timing-sensitive; runs in the full suite only") + } + + const ( + rate = 10_000 + duration = 1500 * time.Millisecond + serviceTime = 50 * time.Millisecond + ) + + small := scenario.Run(scenario.Config{ + Name: "window=5ms", + BatchSize: 100_000, // never size-triggered: isolate the timer + BatchInterval: 5 * time.Millisecond, + Arrival: scenario.Steady(rate, duration), + Processor: scenario.FixedProcessor(serviceTime), + LatenessBudget: time.Second, + }) + + large := scenario.Run(scenario.Config{ + Name: "window=100ms", + BatchSize: 100_000, + BatchInterval: 100 * time.Millisecond, + Arrival: scenario.Steady(rate, duration), + Processor: scenario.FixedProcessor(serviceTime), + LatenessBudget: time.Second, + }) + + t.Logf("5ms window: p50=%s p99=%s mean_batch=%.0f downstream/s=%.0f", + small.EndToEnd.P50, small.EndToEnd.P99, small.MeanBatchSize, small.DownstreamPerSec) + t.Logf("100ms window: p50=%s p99=%s mean_batch=%.0f downstream/s=%.0f", + large.EndToEnd.P50, large.EndToEnd.P99, large.MeanBatchSize, large.DownstreamPerSec) + + require.Greater(t, small.EndToEnd.P50, large.EndToEnd.P50, + "with a %s processor the 5ms window should be SLOWER than 100ms, "+ + "because the inline processor bounds the effective window", serviceTime) +} + +// TestSmallWindowGivesNoOverloadProtection pins the second baseline finding: +// shrinking the batch window does not bound queued work. Overload protection +// requires bounded admission, which the library does not yet offer. +// +// The assertion is expressed relative to what the run actually achieved, not as +// an absolute item count. A shared CI runner cannot offer the same rate as a +// developer machine, and an absolute threshold would simply encode the host it +// was written on. +func TestSmallWindowGivesNoOverloadProtection(t *testing.T) { + t.Parallel() + + if testing.Short() { + t.Skip("allocates aggressively; runs in the full suite only") + } + + const ( + // Offered far above what a 2ms-per-batch processor can retire, so the + // backlog is created by the capacity deficit rather than by any timing + // assumption about the host. + rate = 500_000 + duration = 300 * time.Millisecond + serviceTime = 2 * time.Millisecond + ) + + for _, window := range []time.Duration{100 * time.Millisecond, 1 * time.Millisecond} { + result := scenario.Run(scenario.Config{ + Name: "overload", + BatchSize: 500, + BatchInterval: window, + Arrival: scenario.Steady(rate, duration), + Processor: scenario.FixedProcessor(serviceTime), + // One goroutine cannot offer this rate: the inter-arrival gap is 2µs and + // real sleep granularity is tens of microseconds, so a single producer + // would stretch the run to seconds and collapse the offered rate toward + // the service rate. Spreading the schedule across CPUs keeps the offered + // rate near the intended one on any host. + Producers: runtime.NumCPU(), + LatenessBudget: time.Second, + }) + + t.Logf("window=%-6s offered_for=%s offered/s=%.0f accepted/s=%.0f "+ + "completed/s=%.0f work_peak=%d heap_peak=%dMB heap_sampled=%v", + window, result.OfferedFor.Round(time.Millisecond), + result.OfferedRate, result.AcceptedRate, result.CompletedRate, + result.PendingWorkPeak, result.HeapHighWater/(1<<20), result.HeapSampled) + + require.False(t, result.TimedOut, + "window=%s: the run must complete rather than time out", window) + + require.Zero(t, result.RejectedCount, + "window=%s: an unbounded queue must accept everything, which is exactly "+ + "why a smaller window cannot protect the process", window) + + // The rate assertion is the load-bearing one: work is accepted faster than + // it completes, so the difference accumulates inside the batcher no matter + // how small the window is. + require.Greater(t, result.AcceptedRate, result.CompletedRate, + "window=%s: accepted rate must exceed completed rate under overload", window) + + // PendingWorkPeak is a periodically sampled Len(), which counts in-flight + // work as well as queued work, so it is a lower bound on accumulation rather + // than a queue-depth measurement. Asserting it clears half the deficit is + // therefore a deliberately weak check on a deliberately weak metric: it + // confirms accumulation is real without pretending the number is backlog. + // True queue depth arrives with Stats().Queued in Phase 2. + deficit := (result.AcceptedRate - result.CompletedRate) * result.OfferedFor.Seconds() + require.Greater(t, float64(result.PendingWorkPeak), deficit*0.5, + "window=%s: in-system work must reflect the capacity deficit "+ + "(work_peak=%d, deficit≈%.0f); a smaller window does not bound it", + window, result.PendingWorkPeak, deficit) + } +} + +// TestErroringProcessorDoesNotStallTheHarness keeps the diagnostics path covered, +// so scenarios can model failing downstreams without hanging the run. +func TestErroringProcessorDoesNotStallTheHarness(t *testing.T) { + t.Parallel() + + result := scenario.Run(scenario.Config{ + Name: "erroring", + BatchSize: 10, + BatchInterval: 2 * time.Millisecond, + Arrival: scenario.Sparse(50, time.Millisecond), + Processor: scenario.ErroringProcessor(errors.New("downstream unavailable")), + }) + + require.Positive(t, result.Batches, "batches must still be processed when the processor errors") + require.Positive(t, result.EndToEnd.Count, "latency must still be recorded for failed batches") +} + +// TestDeterministicSchedules asserts fixed seeds produce identical schedules, so +// a reported result can be reproduced exactly. +func TestDeterministicSchedules(t *testing.T) { + t.Parallel() + + arrival := scenario.Poisson(5_000, 200*time.Millisecond) + + first := arrival.Schedule(42) + second := arrival.Schedule(42) + different := arrival.Schedule(43) + + require.Equal(t, first, second, "same seed must produce the same schedule") + require.NotEqual(t, first, different, "different seeds must produce different schedules") +} diff --git a/test/scenario/summarise.go b/test/scenario/summarise.go new file mode 100644 index 0000000..7b3b7c9 --- /dev/null +++ b/test/scenario/summarise.go @@ -0,0 +1,141 @@ +package scenario + +import ( + "runtime" + "time" +) + +// summarise converts raw per-item samples into the reported metrics. +// +// Samples offered during the warmup window are excluded from latency +// distributions but still counted in rates, since they were real load. +func summarise( + cfg Config, + slots []slot, + batches []int, + offeredFor time.Duration, + runDuration time.Duration, + memBefore, memAfter runtime.MemStats, + goroutinesPeak int, + heapPeak uint64, + pendingPeak int64, + heapSamples int64, +) Result { + var ( + lateness []time.Duration + endToEnd []time.Duration + queueDelay []time.Duration + admission []time.Duration + + offered int + accepted int + completed int + rejected int + ) + + for i := range slots { + s := Sample{ + ScheduledAt: time.Duration(slots[i].scheduledAt.Load()), + AdmissionStart: time.Duration(slots[i].admissionStart.Load()), + AdmissionEnd: time.Duration(slots[i].admissionEnd.Load()), + ProcessorStart: time.Duration(slots[i].processorStart.Load()), + ProcessorEnd: time.Duration(slots[i].processorEnd.Load()), + Accepted: slots[i].accepted.Load(), + } + + offered++ + + if !s.Accepted { + rejected++ + + continue + } + + accepted++ + + // A zero ProcessorEnd means the item never reached the processor, which + // is a shortfall worth surfacing rather than averaging away. + if s.ProcessorEnd > 0 { + completed++ + } + + if s.ScheduledAt < cfg.Warmup { + continue + } + + lateness = append(lateness, s.Lateness()) + admission = append(admission, s.AdmissionBlocking()) + + if s.ProcessorEnd > 0 { + endToEnd = append(endToEnd, s.EndToEnd()) + queueDelay = append(queueDelay, s.QueueDelay()) + } + } + + latenessDist := NewDistribution(lateness) + + batchSizes := make([]int, 0, len(batches)) + partial := 0 + totalBatched := 0 + + for _, size := range batches { + batchSizes = append(batchSizes, size) + totalBatched += size + + if size < cfg.BatchSize { + partial++ + } + } + + meanBatch := 0.0 + if len(batches) > 0 { + meanBatch = float64(totalBatched) / float64(len(batches)) + } + + seconds := offeredFor.Seconds() + if seconds <= 0 { + seconds = runDuration.Seconds() + } + + // Prefer the in-flight peak. Fall back to the post-drain reading only if the + // sampler never fired, and report that via HeapSampled so a caller can tell a + // real peak from a recovered-process number. + heapSampled := heapSamples > 0 && heapPeak > 0 + + heapHighWater := heapPeak + if !heapSampled { + heapHighWater = memAfter.HeapAlloc + } + + allocsPerItem := 0.0 + if accepted > 0 { + allocsPerItem = float64(memAfter.Mallocs-memBefore.Mallocs) / float64(accepted) + } + + return Result{ + Config: cfg, + Lateness: latenessDist, + LatenessValid: latenessDist.Count == 0 || latenessDist.P99 <= cfg.LatenessBudget, + EndToEnd: NewDistribution(endToEnd), + QueueDelay: NewDistribution(queueDelay), + AdmissionBlocking: NewDistribution(admission), + OfferedRate: float64(offered) / seconds, + AcceptedRate: float64(accepted) / seconds, + CompletedRate: float64(completed) / runDuration.Seconds(), + RejectedCount: rejected, + Batches: len(batches), + MeanBatchSize: meanBatch, + PartialBatches: partial, + BatchSizeDist: NewIntDistribution(batchSizes), + DownstreamPerSec: float64(len(batches)) / runDuration.Seconds(), + AllocsPerItem: allocsPerItem, + HeapHighWater: heapHighWater, + HeapSampled: heapSampled, + PendingWorkPeak: pendingPeak, + GCCount: memAfter.NumGC - memBefore.NumGC, + GCPauseTotal: time.Duration(memAfter.PauseTotalNs - memBefore.PauseTotalNs), + GCDuringRun: memAfter.NumGC > memBefore.NumGC, + GoroutinesPeak: goroutinesPeak, + Duration: runDuration, + } +}