fix(ingestion-consumer): defer unroutable keys instead of dropping them

[jose-sequeira]

Problem

PR #67447 fixed a drain-locality bug the test suite structurally could not see, which prompted a broader audit of failure scenarios the ingestion-consumer tests never exercise. The audit found one live routing bug plus a set of untested failure paths, some with zero e2e coverage (the reaper loop never ran in any e2e test).

The bug: a fresh (unpinned) key arriving while no worker is routable, for example when the whole pool drains during a deploy overlap, was silently dropped from assignment. The batch then fails its accepted-count check and the process exits. Pinned keys already got the defer-and-wait treatment, so a transient full-pool drain turned into a restart storm for fresh keys only.

Changes

• Unroutable fresh keys are now stashed like pinned deferrals and flushed once a worker returns, bounded by the existing deferred-flush timeout. A key with stashed groups but no pin keeps deferring, so newer messages cannot race ahead of its stashed ones.
• The e2e harness now runs the production reaper loop, can model pod restarts (crash, then a fresh consumer stack in the same group), and the fake worker can simulate lost ACKs, poison 4xx rejections, and partial acceptance.
• New scenario coverage: consumer crash before commit, forced reap at drain timeout with in-flight work, lost-ACK replay, poison batches, partial acceptance, crash mid-drain, full-pool loss at max_in_flight > 1, full fleet redeploy via reconcile_membership (now pub for tests), a second consumer joining the group, static-membership fencing, headerless and non-UTF-8 messages, and post-churn asserts that pins, in-flight load, and stash all drain back to zero.
• A crate README documents the remaining follow-up work: sub-batch size cap vs the worker's 20 MB body limit, commit-error observability, a poison-message DLQ, and revoke-aware partition handoff.

How did you test this code?

TDD. The regression tests for the fix were written first and confirmed red: test_unroutable_fresh_key_defers_until_a_worker_returns (unit) and full_pool_drain_holds_fresh_keys_until_a_worker_returns (e2e) both failed on the unfixed dispatcher, then went green with the fix. test_fresh_key_queues_behind_its_own_deferred_groups catches the second-order ordering hazard the fix introduces (a deferring key with no pin must keep deferring).

Each new scenario test catches a regression nothing covered before, notably redelivery-without-loss after a consumer crash, reaper removal racing a live request, at-least-once behavior on ambiguous ACKs, fail-safe behavior (no commit) on poison and under-acknowledged batches, and bookkeeping leaks after churn.

Full crate suite passes with Kafka running: 151 tests across 6 suites (108 lib unit, 6 dispatcher integration, 27 e2e, 10 transport). cargo fmt clean, cargo clippy clean (the workspace patch notice is pre-existing), cargo shear clean. No manual or production testing.

Automatic notifications

• Publish to changelog?
• Alert Sales and Marketing teams?

Docs update

None needed. The crate README added here is the documentation.

🤖 Agent context

Autonomy: Human-driven (agent-assisted)

Written by Claude (Claude Code), directed by me. I asked for an analysis of failure scenarios the tests do not cover, picked the relevant ones, and had the tests written first to confirm the red state before any fix. Two candidate code changes were rejected after closer analysis: a saturating guard in Stash::completed (underflow is unreachable since entries are removed at zero, a no-op regression test documents the invariant) and a bytes-preserving wire format for non-UTF-8 payloads (capture guarantees UTF-8 JSON, so that class goes to the follow-up DLQ design instead). The two-consumer test asserts no loss only; cross-pod ordering during rebalance is a documented follow-up (revoke-aware handoff). No repo skills were invoked.

[greptile-apps]

Comments Outside Diff (1)

1. rust/ingestion-consumer/src/dispatcher.rs, line 94-106 (link)

   Stash-ordering deferrals counted under reason=drain

   The new None if table.stash.is_deferring arm increments deferred_count, which is emitted with "reason" => "drain" (line ~313). However, this path fires not because a worker is draining, but because the key has outstanding stash entries from a prior unroutable event and newer messages must queue behind them. Monitoring a spike in reason=drain would be ambiguous — it could mean workers are leaving the pool or that messages are being held for stash-ordering. Emitting a separate counter variable (e.g. stash_ordering_deferred_count) with its own reason label — perhaps "reason" => "stash_ordering" — would keep the drain signal clean.

   Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

_{Reviews (1): Last reviewed commit: "chore(ingestion-consumer): document foll..." | Re-trigger Greptile}

[deployment-status-posthog]

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-07-03 09:04 UTC	Run
prod-us	✅ Deployed	2026-07-03 09:26 UTC	Run
prod-eu	✅ Deployed	2026-07-03 09:27 UTC	Run