refactor(ingest): per-table batching + pipelined flushes; unblock per-table-per-test e2e isolation

[EricAndrechek]

Context

PR #190 (sdk-updates) surfaced a CI flake in tests/e2e/sdk/batching.test.ts — "flushes immediately when hitting the 500-item batch limit". Investigation on the throwaway ci-flake-dbg branch (instrumented worker + uploaded wavehouse-cov.log artifacts on runs 26578745441 and 26580071857) showed the batcher itself is healthy — the size trigger fires correctly at batch_len=500 in ~2.9s. The flake is caused by the worker batching across all tables:

// internal/ingest/worker.go:60-65
cons, err := js.CreateOrUpdateConsumer(ctx, mq.StreamName(), jetstream.ConsumerConfig{
    Durable:       BufferConsumerName,
    FilterSubject: "ingest.>",     // every ingest subject
    ...
})
// ...:146-147
var batch []jetstream.Msg          // one slice for all tables
// ...:165
if len(batch) >= maxBatch {        // size trigger ignores table

On CI, streaming.test.ts publishes 2 events to clicks ~340ms before batching.test.ts starts. They open batch 1 first, so the size trigger flushes 2 streaming + 498 batching, leaving the last 2 batching events in batch 2. Batch 2 fires by maxWait=5s timer — but the test's waitForCondition budget is also 5s. The test sees count=498 and times out a few ms before batch 2 lands. (Attempt 2 on the original failing run passed at 4996ms — 4 ms of luck.) Switching streaming.test.ts to a different table does NOT fix it: the maxBatch check counts msgs regardless of table.

Goals

A single PR that addresses the root cause, in this order:

1. Deterministic repro test (do this first)

Before touching worker.go, land a test that always reproduces the contamination. Publish N events to table A, then immediately N events to table B (where A+B totals ≥ maxBatch); assert table B's count reaches N within maxWait − epsilon. With today's code this fails deterministically; with the per-table fix it passes.

This protects us against regressions and gives every commit in the refactor a green-vs-red signal.

2. Refactor the batcher

Several optimizations on the table — design discussion welcome:

• Per-table batches. Each table gets its own batch, batchOpenedAt, and timer. Low-volume tables (events, users, dlq) stop blocking high-volume ones (clicks). Smallest correctness win.
• Pipelined flushes per table. Today flush() blocks runLoop until CH POST returns. Under sustained load on a single hot table, the next batch's accumulation pauses for the CH RTT. Allow a flush goroutine to run while the next batch fills — bounded (e.g., max 2 in-flight per table) to avoid unbounded memory growth.
• ClickHouse "too many parts" guardrail. Naively pipelining = lots of small parts when a hot table hits maxBatch repeatedly per second. Options worth prototyping + benchmarking:
  • Adaptive maxBatch based on observed rate per table (bigger batches when hot)
  • Min interval between flushes per table (the user notes this feels janky — probably not the right primitive)
  • Coalesce the in-flight batch with the next batch if a flush is already running
  • Some combination — needs measurement against Performance Benchmarks #39

3. E2E test isolation

• Each e2e test (or test file) publishes to a unique table name generated at setup. Fixture SQL becomes templated (clicks_${suite} etc.), policy bootstrap covers the generated names. Cross-test contamination becomes structurally impossible.
• Once tables are isolated, drop singleFork: true from tests/e2e/sdk/vitest.config.ts and let test files run in parallel. One WH + one CH still serves everything; concurrent files just write to different tables. Expected wall-time drop on CI is significant.

Related issues

Could potentially fold into this PR — debatable:

• NATS Robustness & Backoff #49 NATS robustness & backoff — reconnect strategy. Touches the same consumer setup; worth doing alongside if the batcher refactor moves consumer config around anyway.
• feat(ingest): transient-error retry buffer with exponential backoff (DLQ shipped) #50 transient-error retry buffer with exponential backoff — pipelined batches need to think about how a retry interacts with the next-in-line batch on the same table, so the design pressure overlaps.

Related but should stay separate PRs:

• refactor: move DLQ settings to policy (per-table/role) and general DLQ + NATS/JetStream cleanup #188 DLQ settings → policy (per-table/role) — policy-driven DLQ behavior, orthogonal to the batching mechanic.
• docs(deploy): JetStream SyncAlways=true durability vs throughput on commodity storage #84 JetStream SyncAlways durability vs throughput docs — different layer (durability config).
• Performance Benchmarks #39 Performance benchmarks — useful for measuring this refactor's effect, but the benchmark harness itself should land independently.
• Configurable Stream Names #36 / feat(mq): make NATS stream name configurable (Re-Iteration) #92 configurable stream names — naming/config, not batching.
• mq: validate WH_MQ_MAX_BYTES_GB upper bound to prevent disk over-reservation #138 validate WH_MQ_MAX_BYTES_GB upper bound — config validation, unrelated layer.

Acceptance criteria

• Deterministic Go (or e2e) test that reproduces the cross-table contamination on main and passes after the refactor.
• Per-table batching implemented; existing e2e suite passes without bumping waitForCondition timeouts.
• Pipelined flushes per table with a documented bound on in-flight CH POSTs.
• Decision recorded (in the PR description or an ADR) on how we avoid CH "too many parts" — adaptive batch, coalescing, or other.
• e2e test files use per-test (or per-file) generated table names; singleFork: true removed; CI wall time recorded before/after.

[coderabbitai]

🔗 Related PRs

#122 - fix(ingest): break infinite-Nak loop on permanent delete errors [merged]
#124 - chore(api)!: drop hub wildcard fan-out; SSE/WS use ?table= (closes #100) [merged]
#125 - fix(boot): non-fatal schema discovery, /health 503 with diagnostic [merged]
#164 - refactor(ingest)!: insert-only pipeline; mutations via /v1/query [merged]
#182 - refactor: full api --> ingest --> clickhouse --> dlq refactor [merged]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!