Shard projection drains by owner, push wakes, fresh first, head deadline

GI
5h ago
+2322-27

Requested by David · project thread

What changed and why

Today every backend-integration instance pages the same global committed_at keyset each 20 s and races the others for each owner's advisory claim. Most drain turns end claim_busy or find nothing, the GPUs sit near 20% busy, and backlog churn drives Firestore reads. This PR makes four changes, all behind default-off or default-unchanged switches.

  1. Owner-hash shard leases (utils/batch_pipeline/projection_shards.py). Each shard has a 30 s Redis lease held by one serving process, and the leases balance across members. Only the holder drains a shard's owners, so each owner has one drain writer.
  2. Push wakes and Firestore cuts (projection_shard_pump.py, serving.py).
    • Commits from Serving.drain_owner push a wake for the owner. Requests routed through Projector.request_drain become wakes for the shard holder.
    • The GPU worker's own merge commits push no wake. Instead, the maintenance leader follows the tail of recently committed rows.
    • Only the maintenance leader pages the result log. It does so discovery-only, from a resuming cursor, and runs the finish page. Followers skip the sweep.
  3. Fresh-first lanes. Each shard has a fresh lane (capture under 6 h old, newest first) and a backlog lane (oldest commit first). Two drain slots stay reserved for backlog while backlog waits. A stuck owner moves behind due owners, so it cannot fill the take page.
  4. Every non-terminal head has an exit (projection.py).
    • The 24 h bound on failing heads is now BATCH_PROJECTION_HEAD_DEADLINE_SECONDS, with the same default.
    • With BATCH_PROJECTION_BLOCKED_HEAD_DEADLINE_ENABLED, a never-passing ProjectionBlocked head on a closed allowlist takes the existing held disposition after the deadline: a quarantine_held receipt, audio and result kept, owned by the release scan and reviewed redrive. Nothing is deleted.

Ordering is unchanged. The transactional predecessor fence and the advisory owner claim still decide every assignment.

Each process fails open to the legacy sweep in any of these cases:

  • A lease pass fails.
  • A wake take fails.
  • One of the existing kill switches is off: owner chain, sweep owner chain, owner drain, or pending page.

Contract: backend/docs/listen_pusher_pipeline.mdx#batch-projection-owner-shards. Change note: backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-owner-shards.md.

Rollout. All new switches default to today's behavior. The intended order:

  1. BATCH_PROJECTION_SHARDS_ENABLED=true on staging backend-integration.
  2. The same on every prod backend-integration serving process in one config train.

The clean metrics are in the flag registry entry. Rollback is false, and the sweep returns on the next tick.

For BATCH_PROJECTION_BLOCKED_HEAD_DEADLINE_ENABLED, David answered "Yes unblock" in the project chat at 2026-11-05 18:46Z, and the coordinator relayed that as approval to enable dead-lettering of held rows in prod. Confirm with David before the prod flip. He asked for a nightly redrive and an alert on dead-letter volume alongside it. Those two are not in this PR.

This builds on the drain pump from #24712. It reuses that PR's BATCH_PIPELINE_MAX_OWNER_DRAINS, which stays at 8. Shards apply only to the serving loops. A dedicated projection pump process keeps its plain sweep. With BATCH_PIPELINE_SERVING_PROJECTION_SWEEP=off, the leader skips the tail and the backstop, but commit wakes still flow.

Out of scope:

  • Transactional multi-row assignment batching. It touches the core receipt transaction and needs its own review.
  • Fresh-first ordering for provider-stt and enrichment, which is #24714.

Line-count exception

Line-Count-Exception: backend/utils/batch_pipeline/projection.py | 1702 -> 1909 | Blocked-head deadline branch and resolver, shard router hooks, and tail/backstop discovery on Projector; new shard logic already lives in projection_shards.py and projection_shard_pump.py. Follow-up split: move Projector discovery (sweep, discover_owners, discover_tail, sweep_finish_only) into utils/batch_pipeline/projection_discovery.py, owned by omi-pipeline.

Product invariants affected

none

How it was verified

Hermetic only. Nothing was run against staging or prod.

  • BACKEND_UNIT_TEST_FILE_LIST=<184 files> bash backend/test.sh exited 0 after rebasing on f8edd96cd. After rebasing again onto 5b8265c50 (which merged #24712), the same set plus test_batch_pipeline_projection_pump.py and test_flag_budget.py passed. The 184 files are every selected backend unit test whose name matches batch_pipeline, projection, serving, firestore, flag, ratchet, gmp, registry, batch_stage, executor, async_block or redis.
  • The full selector requires the whole backend unit suite because database/batch_pipeline_store.py changed. CI runs that.
  • python3 scripts/check_feature_flag_registry.py reported OK.
  • scripts/scan_async_blockers.py --diff-base origin/main reported 0 findings.
  • scripts/backend-python-format (black 26.5.1) was applied.
  • An independent review pass found 8 issues, and all of them are fixed here:
    • GPU commits pushed no wake.
    • Kill switches were bypassed.
    • A failed take did not fail open.
    • A Redis call ran on the loop when deferral overflowed.
    • A settle could reset the sequence and lose a wake.
    • A failed drain task was treated as the chain end.
    • The stop token was wrong after a hold.
    • The deadline clock was shared with transient failures.

Needs a credentialed staging check before the prod flip:

  • Redis Lua EVAL on the serving Redis.
  • Lease balance across the real instance count (omi_batch_projection_shards_unheld is 0).
  • Wake depth and drain throughput against the pre-switch hour.

Tests

  • backend/tests/unit/test_batch_projection_shards.py (38 tests), against fakeredis with real Lua:
    • Leases: cover, balance, handover, join, observer, fail-open.
    • Wakes: lane ordering, the settle sequence race, deferral, the sequence never resetting.
    • Pump: fresh first, backlog reserve, stuck owners not filling the take page, claim-busy keeping its lane position, kill switches, local fallback for an unheld shard.
    • Discovery: tail discovery of GPU commits, the backstop.
    • Serving wiring.
    • Blocked-head deadline: off, on, ineligible messages, clock restart, end stop after a hold.
    • Drain-task error reporting.
  • backend/tests/unit/test_batch_projection_shard_drain.py: two serving processes with real OwnerDrains/Projector on one fakeredis.
    • Every owner drains once, in chain order, by its shard holder only.
    • Zero claim overlaps, and all wakes settle.
    • A foreign drain request is routed to its holder.

UI evidence

UI-Evidence: none -- backend only

Failure class (fixes)

Failure-Class: none

🤖 Generated with Claude Code

https://claude.ai/code/session_017CeWy13948MM5eAUJE7zqo

🤖 Generated with Claude Code