From c2d1bdea8918bc6be9cb0e4e78f7ff347ac90ebd Mon Sep 17 00:00:00 2001 From: Ray Liu Date: Sat, 26 Sep 2026 02:22:43 -0400 Subject: [PATCH] [1/n][worker][docs] plan internal consumer sharding --- docs/plans/0005-internal-consumer-sharding.md | 265 ++++++++++++++++++ 1 file changed, 265 insertions(+) create mode 100644 docs/plans/0005-internal-consumer-sharding.md diff --git a/docs/plans/0005-internal-consumer-sharding.md b/docs/plans/0005-internal-consumer-sharding.md new file mode 100644 index 0000000..3b72dce --- /dev/null +++ b/docs/plans/0005-internal-consumer-sharding.md @@ -0,0 +1,265 @@ +# 0005: Internal Consumer Sharding + +## Objective + +Let one `kq.Worker` run several independent share-group consumers, called +shards, so a single worker process can supply many small poll generations and +reach Redis-level queue time at north-star throughput. + +This keeps the fixed-generation model from [0003](0003-worker-concurrency.md) +inside each shard. It does not add continuous refill. + +## Motivation + +The 2026-09-23 runs showed that queue time is dominated by the generation +wait, and that many small generations remove most of it: + +- [Queue-latency diagnosis](../performance/runs/2026-09-23-adhoc-queue-latency-diagnosis-33c95bf/analysis.md): + with zero execution time, queue p99 was 6 ms. With north-star execution at + concurrency 1000, it was 1,105 ms at only 2,000 RPS. A new record waits for + the in-progress generation, whose length is set by its slowest task. +- [Share group with 1000 members](../performance/runs/2026-09-23-adhoc-share-group-1000-members-33c95bf/analysis.md): + 1000 members x concurrency 10 sustained 20,000 RPS with 14 / 104 / 203 ms + queue p50 / p95 / p99 at 8 partitions, and collapsed at 2 and 4. +- [Shard-20 ramp](../performance/runs/2026-09-23-adhoc-shard-20-ramp-33c95bf/analysis.md): + 1000 members x concurrency 20 sustained 34,000 RPS at 16 partitions with + 22 / 66 / 95 ms, below the Redis reference of 36 / 77 / 100 ms. + +Those runs emulated shards with 1000 separate `kq.Worker` instances across 10 +processes. Each instance created its own share consumer and its own producer, +so 1000 members also meant 1000 idle producer clients. Applications should not +need to construct and supervise hundreds of workers to get this behavior. + +A single consumer cannot refill freed slots because franz-go finalizes +unresolved share records when the next poll begins. Separate consumers each +poll independently, so shards are the smallest change that exploits the +measured effect. + +## Scope + +This plan covers: + +- A shard count in `WorkerConfig`. +- One share consumer and one fixed handler pool per shard. +- One producer shared by every shard in a worker. +- Independent poll generations, acknowledgement flushing, and shutdown per + shard. +- Worker-wide error propagation and cancellation. +- Clear errors when the share group is full. +- Operator guidance for member budgets, shard concurrency, and partitions. +- Unit and Kafka integration coverage. +- A performance comparison against the emulated-shard runs. + +This plan does not cover: + +- Continuous refill or explicit acknowledgement across polls. +- Acquisition renewal or bounded shutdown. +- Changing shard count while a worker runs, or autoscaling. +- Sharding the retry mover. +- Changes to franz-go. +- Metrics. Queue time and completion rate should be exposed before a + production pilot, in a separate plan. + +## Configuration + +```go +type WorkerConfig struct { + Config + Shards int // share consumers in this worker + ShardConcurrency int // records per shard generation and handlers per shard +} +``` + +Both fields are required. Like `Concurrency` in 0003, neither has a hidden +default; callers choose them explicitly: + +```text +share-group members from this worker = Shards +handler slots in this worker = Shards x ShardConcurrency +``` + +`ShardConcurrency` replaces 0003's `Concurrency` and keeps its meaning within +a shard: the generation size and handler pool size. `Shards: 1` with +`ShardConcurrency: X` behaves exactly as `Concurrency: X` does today. + +Validation: + +- `Shards` must be between 1 and 1000, the broker maximum for + `group.share.max.size`. +- `ShardConcurrency` must be between 1 and `math.MaxInt32`, the limit of + franz-go's `ShareMaxRecords`. +- `Shards x ShardConcurrency` must not overflow `int`. + +Removing `Concurrency` is a breaking API change. Update every caller in the +same change: the harness worker role, integration tests, and examples in +`docs/`. The harness topology renames `worker_concurrency` to +`worker_shard_concurrency` and adds `worker_shards`; resolved runs record both, +and published runs keep their original field names. + +## Execution Model + +```mermaid +flowchart LR + Kafka[(Ready topic)] + subgraph Worker + S1[Shard 1: consumer + pool] + S2[Shard 2: consumer + pool] + SN[Shard N: consumer + pool] + Producer[Shared producer] + end + Kafka --> S1 + Kafka --> S2 + Kafka --> SN + S1 --> Producer + S2 --> Producer + SN --> Producer + Producer --> Retry[(Retry and DLQ topics)] +``` + +Each shard runs the existing 0003 loop unchanged: + +```text +poll up to ShardConcurrency records +-> execute them on the shard's pool +-> resolve and flush the shard's acknowledgements +-> poll the next generation +``` + +Shards share nothing except the handler, the queue configuration, and the +producer. A slow task delays only its own shard's next poll. The franz-go +producer client is safe for concurrent use, so retry and DLQ writes from every +shard go through one client. + +`Worker.Run` starts every shard and returns when all shards have returned. + +## Record Lifecycle + +The per-record lifecycle and write-before-source-resolution rule from 0003 +apply within each shard unchanged. A record is polled, handled, and +acknowledged by exactly one shard. Kafka's share-group delivery is the only +coordination between shards; KQ adds no cross-shard state. + +## Errors + +A worker-fatal error in any shard cancels every other shard. The worker waits +for every shard to drain under the 0003 shutdown rules, then returns all +causes with `errors.Join`. Errors identify their shard index. + +A shard's consumer failing to join because the share group is at +`group.share.max.size` is worker-fatal. The returned error names the broker +limit and the configured shard count so the misconfiguration is obvious. + +## Shutdown + +On context cancellation, every shard follows the 0003 shutdown sequence +independently: + +```text +stop polling +let dispatched handlers observe cancellation +wait for the shard's pool +release unsuccessful records +flush final acknowledgements +``` + +After every shard has returned, the worker returns. `Close` closes every shard +consumer and then the shared producer, so no shard can write a retry or DLQ +record after the producer closes. + +## Resource Cost + +Each shard is one franz-go client. It holds connections to the brokers that +lead its fetched partitions and one share session per broker. Clients cannot be +shared between shards because a franz-go share-group client is exactly one +group member. + +The runs used 100 members per process. This plan documents that shape as +tested and does not recommend larger per-process shard counts until measured. + +## Operator Guidance + +Sizing relations, Kafka limits, and measured behavior live in +[`docs/scaling.md`](../scaling.md). Update it with shard terminology: + +```text +members per queue = sum of Shards across every worker process +slots per queue = members x ShardConcurrency +``` + +Link it from [`docs/architecture/worker.md`](../architecture/worker.md), and +replace its emulated-shard measurements with this plan's sharded validation +run once published. + +## Test Plan + +Unit tests with a fake consumer factory must verify: + +- `Shards` and `ShardConcurrency` outside their ranges, including 0, are + rejected; `Shards: 1` creates one consumer. +- A worker with `Shards: N` creates exactly N consumers and one producer. +- Each shard executes at most `ShardConcurrency` handlers at once. +- Total concurrent handlers reach `Shards x ShardConcurrency` when work is + available. +- A slow task in one shard does not block polling in another. +- Each shard never begins a new generation before its current one is resolved + and flushed. +- A fatal error in one shard cancels the others and returns joined errors + naming each shard. +- Cancellation drains every shard and does not leak goroutines. +- `Close` closes all consumers before the shared producer. +- The share-group-full error is surfaced with the configured shard count. + +Kafka integration coverage must run a worker with `Shards: 4` and +`ShardConcurrency: 2`, block handlers at a barrier, observe 8 concurrent handlers, +release them, and account for every task exactly once. Existing integration tests must pass after migrating to `Shards: 1`. Run tests with the race detector. + +## Performance Validation + +This plan depends on [0004](0004-scalable-harness.md) for measurement. Add +`worker_shards` to the harness topology, then compare real shards with the +emulated-shard run 6: + +```text +broker: group.share.max.size=1000, record locks 4000 +scenario: north-star-steady-v1 (same as run 6) +emulated (reference): 10 processes x 100 workers x Shards 1 x ShardConcurrency 20 +sharded (candidate): 10 processes x 1 worker x Shards 100 x ShardConcurrency 20 +partitions: 16 +ramp: 24,000 to 36,000 RPS +``` + +The comparison must report the highest passing rate, queue percentiles, Kafka +CPU, worker CPU and memory, and Kafka client connections per process. + +Expected results: + +- The same highest passing rate as run 6, one ramp step either way. +- Queue percentiles within the tolerance used in 0004 validation. +- Lower worker memory and fewer client connections, since 1000 producer + clients become 10. + +Then repeat the headline point with `north-star-steady-v2`. + +## Acceptance Criteria + +The plan is complete when: + +- Callers must set `Shards` and `ShardConcurrency` explicitly; `Shards: 1` + reproduces current behavior. +- Every existing `Concurrency` caller is migrated. +- One worker runs that many independent share consumers with one shared + producer. +- Per-shard concurrency bounds and generation ordering hold. +- Each record is accepted or released exactly once. +- A fatal shard error stops the worker and preserves every cause. +- Cancellation drains every shard; `Close` never closes the producer before + the shards. +- A full share group produces an explicit, actionable error. +- Unit and integration tests pass under the race detector. +- `docs/scaling.md` uses shard terminology and cites the sharded run. +- A sharded run matching run 6's topology is published and fully accounted for. + +## Open Questions + +- Should the worker spread shard startup over time to avoid 100 simultaneous + group joins per process?