Repository navigation
Shard projection drains by owner, push wakes, fresh first, head deadline - #24731
Git-on-my-level wants to merge 2 commits into
Conversation
Owner-hash shard leases in Redis give each owner one drain writer across the fleet. Merge commits push wakes, and the maintenance leader follows the committed tail and runs a discovery-only backstop. The per-instance sweep is no longer needed while shards are engaged. Holders fill drain slots fresh first and keep a reserve for backlog. A stuck owner yields its lane place. Every Redis error, and every existing drain kill switch, returns that process to the legacy sweep. The ordering fences are unchanged. Eligible never-passing ProjectionBlocked heads get a deadline and take the existing held disposition (audio kept, redrivable). This is behind BATCH_PROJECTION_BLOCKED_HEAD_DEADLINE_ENABLED, default off. All new switches default to current behavior. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017CeWy13948MM5eAUJE7zqo
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a30a19ff4f
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| drains = projector.ensure_drains() | ||
| free = drains.free_slots | ||
| if free <= 0: | ||
| return 0 |
There was a problem hiding this comment.
With shards enabled, once all drain slots are occupied by owners with long chains, this check prevents every other owner from being admitted. OwnerDrains.finished() immediately requeues productive continuations, so free_slots remains zero even at the 120-second turn boundary. Unlike the previous sweep, which queued other owners behind active turns, the shard pump cannot add waiting owners to that FIFO. Fresh captures and other backlog owners can therefore wait until an initially admitted owner's entire chain finishes, defeating both fresh-first scheduling and per-owner turn fairness. Allow bounded admission of waiting owners between continuations rather than requiring an empty execution slot.
Useful? React with 👍 / 👎.
| def blocked_head_type(error: BaseException) -> str: | ||
| """Bounded, log-safe episode token for a deadline-eligible refusal.""" | ||
| return f'ProjectionBlocked:{projection_block_reason(error)}' |
There was a problem hiding this comment.
When the blocked-head deadline is enabled, all six message-based eligible refusals produce the same ProjectionBlocked:other token because their exceptions do not carry block_reason. If one refusal is resolved and a different eligible refusal is encountered—for example, a recovery handoff lands and assignment then discovers missing VAD accounting—resolve_blocked_head() reuses the old timestamp and can immediately quarantine the row instead of restarting the new refusal's deadline. Give each allowlisted message a distinct bounded episode token so the deadline measures time spent blocked for the same reason.
Useful? React with 👍 / 👎.
| except Exception as error: | ||
| self.healthy = False | ||
| self.held = set() | ||
| self.unheld = set(range(self.count)) | ||
| logger.warning( |
There was a problem hiding this comment.
When Redis refresh fails, this branch marks sharding inactive and serving resumes the legacy per-instance sweep, but it only emits a warning. Neither this path nor the legacy-sweep caller invokes the required shared record_fallback helper, so operators cannot observe this mode change through omi_fallback_total even though it restores fleet-wide duplicate discovery and claim contention. Instrument the fail-open transition with the shared helper and an appropriate outcome, rather than relying solely on the shard warning.
AGENTS.md reference: backend/AGENTS.md:L322-L322
Useful? React with 👍 / 👎.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
19 issues found across 20 files
Confidence score: 2/5
- In
projection_shard_pump.py, a productive owner can keep requeueing itself afterfree_slotsreaches zero, preventing other owners from getting a turn. Admit a bounded waiting set between continuations. - In
batch_pipeline.py, the documentedBATCH_PROJECTION_SHARDS_ENABLED=1andonvalues do not enable shards in normal serving mode. Parse the documented truthy values before passing the legacy value through. - In
projection.py, different eligible refusal types all becomeProjectionBlocked:other, so changing refusal type does not restart the deadline. Encode each allowlisted refusal type distinctly. - In
projection_shard_pump.py,pickcan select a backlog wake before its retry time after a process restart, causing a stuck owner to retry immediately. Skip backlog wakes whose scores are still in the future.
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="backend/tests/unit/test_batch_projection_shards.py">
<violation number="1" location="backend/tests/unit/test_batch_projection_shards.py:23">
P2: These tests do not isolate all kill switches that control the behavior they assert. Clear `BATCH_PIPELINE_OWNER_DRAIN_ENABLED` and `BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` in the autouse fixture (or set every required switch explicitly) so CI/deployment environment flags cannot change the test outcome.</violation>
</file>
<file name="backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-owner-shards.md">
<violation number="1" location="backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-owner-shards.md:21">
P2: The 30-page bound excludes the five-page tail and the finish page. State the total as up to 35 discovery pages plus one finish page, or reduce the component budgets to preserve the advertised read bound.</violation>
<violation number="2" location="backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-owner-shards.md:25">
P2: This heading overstates the guarantee: several non-terminal head refusal classes explicitly remain waiting, and the blocked-head deadline is disabled by default. Rename it to limit the exit guarantee to eligible failing or blocked heads.</violation>
</file>
<file name="backend/tests/unit/test_batch_projection_shard_drain.py">
<violation number="1" location="backend/tests/unit/test_batch_projection_shard_drain.py:35">
P2: This fixture does not force the shard prerequisites on, so an inherited false value makes `engaged` false and these tests stop exercising shard routing. Set all required shard/owner-chain switches to `true` (and keep prefetch separately unset) to make the tests environment-independent.</violation>
</file>
<file name="backend/docs/batch-pipeline-serving.md">
<violation number="1" location="backend/docs/batch-pipeline-serving.md:1265">
P3: `Each owner has one drain writer fleet-wide` is only true while its shard is held and Redis is healthy. Unheld shards and local fallback requests can drain on another instance.</violation>
</file>
<file name="backend/utils/batch_pipeline/projection_shard_pump.py">
<violation number="1" location="backend/utils/batch_pipeline/projection_shard_pump.py:129">
P2: Future-dated captures are admitted to the fresh lane and scored with their future timestamp, so clock-skewed audio can outrank genuinely current work. Reject future ages or clamp the capture score before assigning the fresh lane, matching the sync lane’s future-skew handling.</violation>
<violation number="2" location="backend/utils/batch_pipeline/projection_shard_pump.py:232">
P1: `free_slots` reaching zero stops the pump from admitting more owners, while productive turns can immediately requeue the same owner. Admit a bounded waiting set between continuations so other owners can reach the drain FIFO.</violation>
<violation number="3" location="backend/utils/batch_pipeline/projection_shard_pump.py:269">
P2: Deferred backlog wakes are still eligible before their retry time after a process restart because `pick` never checks the wake score. Filter future backlog scores here so a stuck owner cannot be retried immediately and monopolize a drain slot.</violation>
</file>
<file name="backend/config/batch_pipeline.py">
<violation number="1" location="backend/config/batch_pipeline.py:921">
P2: `BATCH_PROJECTION_SHARDS_ENABLED=1` or `on` does not enable shards in the normal serving mode, despite the documented `true/1/on` contract. Parse the documented truthy values before passing the legacy value to `rollout_default_off`.</violation>
</file>
<file name="backend/docs/listen_pusher_pipeline.mdx">
<violation number="1" location="backend/docs/listen_pusher_pipeline.mdx:9366">
P3: The documented enablement set omits `yes`, although the shared rollout parser accepts it. Add `yes` to the contract or narrow the parser so operators do not get a surprising rollout value.</violation>
<violation number="2" location="backend/docs/listen_pusher_pipeline.mdx:9397">
P2: Turn and step-budget continuations do not cool or defer their wakes; `OwnerDrains` immediately requeues them at the FIFO tail. Document cooldown only for blocked, error, or hold-not-landed stops, or change the continuation behavior.</violation>
<violation number="3" location="backend/docs/listen_pusher_pipeline.mdx:9420">
P2: The backstop can route the same owner twice per tick because tail and whole-log discovery use separate deduplication sets. Share a per-tick owner set or document the two-pass behavior; otherwise duplicate wake pushes add Redis work and advance wake sequences unnecessarily.</violation>
<violation number="4" location="backend/docs/listen_pusher_pipeline.mdx:9432">
P2: Wake pushes and routed drain requests call Redis synchronously outside `db_executor`, so they can occupy the caller's executor for the Redis timeout. Route these operations through `db_executor` or narrow this statement to the pump's tick and depth operations.</violation>
</file>
<file name="backend/utils/batch_pipeline/projection.py">
<violation number="1" location="backend/utils/batch_pipeline/projection.py:210">
P2: `blocked_head_type` collapses every message-based eligible refusal to `ProjectionBlocked:other`, so changing from a handoff refusal to a VAD or manifest refusal does not restart the deadline. Encode each allowlisted message in the episode token so the current refusal receives its own deadline.</violation>
<violation number="2" location="backend/utils/batch_pipeline/projection.py:564">
P2: `discover_tail` routes an owner from its first row, so an older capture can force a mixed owner into BACKLOG and hide a newer fresh capture. Aggregate each owner’s rows and choose the fresh/latest lane before routing.</violation>
</file>
<file name="backend/utils/batch_pipeline/projection_shards.py">
<violation number="1" location="backend/utils/batch_pipeline/projection_shards.py:141">
P2: A stuck owner present in both lanes is not fully deferred: the fresh wake increments the backlog sequence before the already-taken backlog wake is processed, so that second defer returns 0 and the old backlog score remains at the head. Defer both lane entries atomically, or avoid changing the target sequence until the existing backlog wake has been handled, so stuck owners cannot keep crowding take pages.</violation>
<violation number="2" location="backend/utils/batch_pipeline/projection_shards.py:209">
P2: Record the transition with the shared fallback helper; otherwise a Redis refresh failure switches this instance to the legacy sweep without incrementing `omi_fallback_total`.</violation>
<violation number="3" location="backend/utils/batch_pipeline/projection_shards.py:400">
P2: `settle()` and `defer()` do not mark the coordinator inactive when their Redis calls fail, so the process can remain engaged instead of returning to the legacy sweep. Mark the shard coordinator unhealthy on these errors, consistently with `take()`, before returning failure.</violation>
</file>
<file name="backend/config/flag_budget.json">
<violation number="1" location="backend/config/flag_budget.json:27">
P3: The budget metadata is stale: `as_of` remains `2026-10-29` while this change adds five names from 2026-11-05. Update `as_of` to the actual inventory date so the count has an accurate audit timestamp.</violation>
</file>
Reply with feedback, questions, or to request a fix.
View guided diff | Turn on auto-fix | Re-trigger cubic
| return 0 | ||
| drains = projector.ensure_drains() | ||
| free = drains.free_slots | ||
| if free <= 0: |
There was a problem hiding this comment.
P1: free_slots reaching zero stops the pump from admitting more owners, while productive turns can immediately requeue the same owner. Admit a bounded waiting set between continuations so other owners can reach the drain FIFO.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/utils/batch_pipeline/projection_shard_pump.py, line 232:
<comment>`free_slots` reaching zero stops the pump from admitting more owners, while productive turns can immediately requeue the same owner. Admit a bounded waiting set between continuations so other owners can reach the drain FIFO.</comment>
<file context>
@@ -0,0 +1,294 @@
+ return 0
+ drains = projector.ensure_drains()
+ free = drains.free_slots
+ if free <= 0:
+ return 0
+ now = self.monotonic()
</file context>
| from utils.batch_pipeline.scheduler import owner_key | ||
| from utils.batch_pipeline.serving import Serving | ||
|
|
||
| ENV = ( |
There was a problem hiding this comment.
P2: These tests do not isolate all kill switches that control the behavior they assert. Clear BATCH_PIPELINE_OWNER_DRAIN_ENABLED and BATCH_PIPELINE_SERVING_PROJECTION_SWEEP in the autouse fixture (or set every required switch explicitly) so CI/deployment environment flags cannot change the test outcome.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/tests/unit/test_batch_projection_shards.py, line 23:
<comment>These tests do not isolate all kill switches that control the behavior they assert. Clear `BATCH_PIPELINE_OWNER_DRAIN_ENABLED` and `BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` in the autouse fixture (or set every required switch explicitly) so CI/deployment environment flags cannot change the test outcome.</comment>
<file context>
@@ -0,0 +1,671 @@
+from utils.batch_pipeline.scheduler import owner_key
+from utils.batch_pipeline.serving import Serving
+
+ENV = (
+ 'BATCH_PROJECTION_SHARDS_ENABLED',
+ 'BATCH_PROJECTION_BLOCKED_HEAD_DEADLINE_ENABLED',
</file context>
| 3. **Fresh first.** Commits of audio captured under 6 h ago take the `fresh` | ||
| lane, newest capture first. Two drain slots per process stay reserved for | ||
| backlog while backlog waits. A stuck owner moves behind due owners. | ||
| 4. **Every non-terminal head has an exit.** |
There was a problem hiding this comment.
P2: This heading overstates the guarantee: several non-terminal head refusal classes explicitly remain waiting, and the blocked-head deadline is disabled by default. Rename it to limit the exit guarantee to eligible failing or blocked heads.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-owner-shards.md, line 25:
<comment>This heading overstates the guarantee: several non-terminal head refusal classes explicitly remain waiting, and the blocked-head deadline is disabled by default. Rename it to limit the exit guarantee to eligible failing or blocked heads.</comment>
<file context>
@@ -0,0 +1,49 @@
+3. **Fresh first.** Commits of audio captured under 6 h ago take the `fresh`
+ lane, newest capture first. Two drain slots per process stay reserved for
+ backlog while backlog waits. A stuck owner moves behind due owners.
+4. **Every non-terminal head has an exit.**
+ - Failing heads already took the held disposition after 24 h. That bound
+ is now `BATCH_PROJECTION_HEAD_DEADLINE_SECONDS`.
</file context>
| 4. **Every non-terminal head has an exit.** | |
| 4. **Eligible failing and blocked heads have an exit.** |
| monkeypatch.setattr(projection_drain, 'projection_claim_held', lambda namespace, key: key in state['held']) | ||
| monkeypatch.setattr(projection, 'projection_claim', claim) | ||
| monkeypatch.setattr(projection_followup, 'projection_claim', claim) | ||
| for name in ('BATCH_PIPELINE_OWNER_DRAIN_ENABLED', 'BATCH_PIPELINE_PROJECTION_PREFETCH'): |
There was a problem hiding this comment.
P2: This fixture does not force the shard prerequisites on, so an inherited false value makes engaged false and these tests stop exercising shard routing. Set all required shard/owner-chain switches to true (and keep prefetch separately unset) to make the tests environment-independent.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/tests/unit/test_batch_projection_shard_drain.py, line 35:
<comment>This fixture does not force the shard prerequisites on, so an inherited false value makes `engaged` false and these tests stop exercising shard routing. Set all required shard/owner-chain switches to `true` (and keep prefetch separately unset) to make the tests environment-independent.</comment>
<file context>
@@ -0,0 +1,122 @@
+ monkeypatch.setattr(projection_drain, 'projection_claim_held', lambda namespace, key: key in state['held'])
+ monkeypatch.setattr(projection, 'projection_claim', claim)
+ monkeypatch.setattr(projection_followup, 'projection_claim', claim)
+ for name in ('BATCH_PIPELINE_OWNER_DRAIN_ENABLED', 'BATCH_PIPELINE_PROJECTION_PREFETCH'):
+ monkeypatch.delenv(name, raising=False)
+
</file context>
| if not self.shards.covered(key): | ||
| return False | ||
| now = self.shards.clock() | ||
| if capture_at is not None and now - capture_at <= FRESH_CAPTURE_SECONDS: |
There was a problem hiding this comment.
P2: Future-dated captures are admitted to the fresh lane and scored with their future timestamp, so clock-skewed audio can outrank genuinely current work. Reject future ages or clamp the capture score before assigning the fresh lane, matching the sync lane’s future-skew handling.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/utils/batch_pipeline/projection_shard_pump.py, line 129:
<comment>Future-dated captures are admitted to the fresh lane and scored with their future timestamp, so clock-skewed audio can outrank genuinely current work. Reject future ages or clamp the capture score before assigning the fresh lane, matching the sync lane’s future-skew handling.</comment>
<file context>
@@ -0,0 +1,294 @@
+ if not self.shards.covered(key):
+ return False
+ now = self.shards.clock()
+ if capture_at is not None and now - capture_at <= FRESH_CAPTURE_SECONDS:
+ lane, score = FRESH, float(capture_at)
+ else:
</file context>
| if capture_at is not None and now - capture_at <= FRESH_CAPTURE_SECONDS: | |
| if capture_at is not None and 0 <= now - capture_at <= FRESH_CAPTURE_SECONDS: |
| only if no push landed since the take. So a commit during a drain always gets | ||
| another drain. | ||
|
|
||
| A drain that stops short (blocked, error, turn or step budget, hold not |
There was a problem hiding this comment.
P2: Turn and step-budget continuations do not cool or defer their wakes; OwnerDrains immediately requeues them at the FIFO tail. Document cooldown only for blocked, error, or hold-not-landed stops, or change the continuation behavior.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/docs/listen_pusher_pipeline.mdx, line 9397:
<comment>Turn and step-budget continuations do not cool or defer their wakes; `OwnerDrains` immediately requeues them at the FIFO tail. Document cooldown only for blocked, error, or hold-not-landed stops, or change the continuation behavior.</comment>
<file context>
@@ -9351,3 +9351,144 @@ owner-drain admissions use a read-only Redis claim GET and fail open on errors.
+only if no push landed since the take. So a commit during a drain always gets
+another drain.
+
+A drain that stops short (blocked, error, turn or step budget, hold not
+landed) cools the owner for about 120 s in process memory. It also moves the
+wake to `backlog` at its retry time, so stuck owners never fill the take page
</file context>
| self.members = 0 | ||
| # False until a lease pass succeeds, and after any Redis error: the | ||
| # caller then runs the legacy sweep on this instance. | ||
| self.healthy = False |
There was a problem hiding this comment.
P2: Record the transition with the shared fallback helper; otherwise a Redis refresh failure switches this instance to the legacy sweep without incrementing omi_fallback_total.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/utils/batch_pipeline/projection_shards.py, line 209:
<comment>Record the transition with the shared fallback helper; otherwise a Redis refresh failure switches this instance to the legacy sweep without incrementing `omi_fallback_total`.</comment>
<file context>
@@ -0,0 +1,466 @@
+ self.members = 0
+ # False until a lease pass succeeds, and after any Redis error: the
+ # caller then runs the legacy sweep on this instance.
+ self.healthy = False
+ self.refreshed_at: float | None = None
+ self._lock = threading.Lock()
</file context>
| ### Owner shards and push wakes (dark, 2026-11-05) | ||
|
|
||
| `BATCH_PROJECTION_SHARDS_ENABLED` (default off) replaces the per-instance | ||
| sweep's owner discovery with owner-hash shard leases. Each owner has one |
There was a problem hiding this comment.
P3: Each owner has one drain writer fleet-wide is only true while its shard is held and Redis is healthy. Unheld shards and local fallback requests can drain on another instance.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/docs/batch-pipeline-serving.md, line 1265:
<comment>`Each owner has one drain writer fleet-wide` is only true while its shard is held and Redis is healthy. Unheld shards and local fallback requests can drain on another instance.</comment>
<file context>
@@ -1259,6 +1259,22 @@ below have their own switches; all three off is the pre-drain behaviour):
+### Owner shards and push wakes (dark, 2026-11-05)
+
+`BATCH_PROJECTION_SHARDS_ENABLED` (default off) replaces the per-instance
+sweep's owner discovery with owner-hash shard leases. Each owner has one
+drain writer fleet-wide, and that writer is woken by merge commits rather than
+by a polling keyset.
</file context>
|
|
||
| | Env | Default | Bounds and fallback | | ||
| | --- | --- | --- | | ||
| | `BATCH_PROJECTION_SHARDS_ENABLED` | off | Only `true`/`1`/`on` enables it. Off is the per-instance sweep. | |
There was a problem hiding this comment.
P3: The documented enablement set omits yes, although the shared rollout parser accepts it. Add yes to the contract or narrow the parser so operators do not get a surprising rollout value.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/docs/listen_pusher_pipeline.mdx, line 9366:
<comment>The documented enablement set omits `yes`, although the shared rollout parser accepts it. Add `yes` to the contract or narrow the parser so operators do not get a surprising rollout value.</comment>
<file context>
@@ -9351,3 +9351,144 @@ owner-drain admissions use a read-only Redis claim GET and fail open on errors.
+
+| Env | Default | Bounds and fallback |
+| --- | --- | --- |
+| `BATCH_PROJECTION_SHARDS_ENABLED` | off | Only `true`/`1`/`on` enables it. Off is the per-instance sweep. |
+| `BATCH_PROJECTION_SHARD_MEMBER` | true | `false` keeps the process out of shard leases. It still routes and pushes wakes. |
+| `BATCH_PROJECTION_BACKLOG_RESERVE` | 2 | 0..16, never more than the drain slots (`BATCH_PIPELINE_MAX_OWNER_DRAINS`). Malformed values keep 2. |
</file context>
| | `BATCH_PROJECTION_SHARDS_ENABLED` | off | Only `true`/`1`/`on` enables it. Off is the per-instance sweep. | | |
| | `BATCH_PROJECTION_SHARDS_ENABLED` | off | Only `true`/`1`/`on`/`yes` enables it. Off is the per-instance sweep. | |
| "seed_review_by": "2026-11-15", | ||
| "budget": 169, | ||
| "count": 259, | ||
| "count": 264, |
There was a problem hiding this comment.
P3: The budget metadata is stale: as_of remains 2026-10-29 while this change adds five names from 2026-11-05. Update as_of to the actual inventory date so the count has an accurate audit timestamp.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/config/flag_budget.json, line 27:
<comment>The budget metadata is stale: `as_of` remains `2026-10-29` while this change adds five names from 2026-11-05. Update `as_of` to the actual inventory date so the count has an accurate audit timestamp.</comment>
<file context>
@@ -24,7 +24,7 @@
"seed_review_by": "2026-11-15",
"budget": 169,
- "count": 259,
+ "count": 264,
"names": {
"BATCH_PIPELINE_SERVING_PROJECTION_SWEEP": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"},
</file context>
A member taken off the sharded path by a drain kill switch kept renewing its leases, because tick() refreshed before the engagement gate. Other members then routed wakes into lanes nobody read, and those owners stalled silently. A switched-off member now releases its leases, leaves the members set, and skips lease passes until the switches are back on. It rejoins on the next pass. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017CeWy13948MM5eAUJE7zqo
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 38305edb17
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
|
||
| fresh_ok = [w for w in fresh if eligible(w)] | ||
| backlog_ok = [w for w in backlog if eligible(w)] | ||
| reserve = min(projection_backlog_reserve(), free) if backlog_ok else 0 |
There was a problem hiding this comment.
When owners finish one or two at a time while both lanes have waiting work, the default reserve of two assigns every newly available slot to backlog, even if other backlog owners already occupy the reserved capacity. After the initial fresh owners finish, all slots can become backlog slots, and fresh captures then starve until the backlog empties or at least three slots happen to become free together. Calculate the reservation against backlog drains already in flight rather than reserving two additional slots on every admission.
Useful? React with 👍 / 👎.
| and projection_blocked_head_deadline_enabled() | ||
| and blocked_head_deadline_eligible(error) | ||
| and self.resolve_blocked_head(segment_id, job, entry, binding, error) |
There was a problem hiding this comment.
With the blocked-head deadline enabled but sharding disabled, an owner's sole pending result can repeatedly raise an eligible refusal without ever reaching this handler. sweep() catches ProjectionBlocked and continues without scheduling an owner drain or calling resolve_blocked_head(), so it never stamps projection_blocked_since and the result remains blocked indefinitely. This also affects dedicated projection-pump processes, which retain the plain sweep. Route eligible blocked heads into the walker, or apply the deadline from the sweep after verifying that the result is the chain head.
Useful? React with 👍 / 👎.
Requested by David · project thread
What changed and why
Today every backend-integration instance pages the same global
committed_atkeyset each 20 s and races the others for each owner's advisory claim. Most drain turns endclaim_busyor 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.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.projection_shard_pump.py,serving.py).Serving.drain_ownerpush a wake for the owner. Requests routed throughProjector.request_drainbecome wakes for the shard holder.freshlane (capture under 6 h old, newest first) and abackloglane (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.projection.py).BATCH_PROJECTION_HEAD_DEADLINE_SECONDS, with the same default.BATCH_PROJECTION_BLOCKED_HEAD_DEADLINE_ENABLED, a never-passingProjectionBlockedhead on a closed allowlist takes the existing held disposition after the deadline: aquarantine_heldreceipt, 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:
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:
BATCH_PROJECTION_SHARDS_ENABLED=trueon staging backend-integration.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. WithBATCH_PIPELINE_SERVING_PROJECTION_SWEEP=off, the leader skips the tail and the backstop, but commit wakes still flow.Out of scope:
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.shexited 0 after rebasing onf8edd96cd. After rebasing again onto5b8265c50(which merged feat(batch): add opt-in dedicated projection drain pump #24712), the same set plustest_batch_pipeline_projection_pump.pyandtest_flag_budget.pypassed. 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.database/batch_pipeline_store.pychanged. CI runs that.python3 scripts/check_feature_flag_registry.pyreported OK.scripts/scan_async_blockers.py --diff-base origin/mainreported 0 findings.scripts/backend-python-format(black 26.5.1) was applied.Needs a credentialed staging check before the prod flip:
EVALon the serving Redis.omi_batch_projection_shards_unheldis 0).Tests
backend/tests/unit/test_batch_projection_shards.py(38 tests), against fakeredis with real Lua:endstop after a hold.backend/tests/unit/test_batch_projection_shard_drain.py: two serving processes with realOwnerDrains/Projectoron one fakeredis.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