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

backend/config/batch_pipeline.py

@@ -892,3 +892,64 @@ def recovery_wake_lease_seconds() -> int:
     if value == 0:
         return 0
     return value if 120 <= value <= RECOVERY_WAKE_LEASE_MAX else RECOVERY_WAKE_LEASE_DEFAULT
+
+
+# Owner-hash projection shards (utils/batch_pipeline/projection_shards.py).
+PROJECTION_BACKLOG_RESERVE_DEFAULT = 2
+PROJECTION_BACKLOG_RESERVE_MAX = 16  # max_owner_drains() caps slots at 16
+PROJECTION_HEAD_DEADLINE_DEFAULT = 24 * 3600
+PROJECTION_HEAD_DEADLINE_MIN = 3600
+PROJECTION_HEAD_DEADLINE_MAX = 7 * 24 * 3600
+
+
+def _bounded_int(raw: str, default: int, low: int, high: int) -> int:
+    """Malformed or out-of-range values keep the default."""
+    raw = raw.strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    return value if low <= value <= high else default
+
+
+def projection_shards_enabled() -> bool:
+    """Default off: one shard holder drains each owner; off restores the per-instance sweep."""
+    return rollout_default_off(
+        'BATCH_PROJECTION_SHARDS_ENABLED',
+        os.getenv('BATCH_PROJECTION_SHARDS_ENABLED', 'false').strip().lower() == 'true',
+    )
+
+
+def projection_shard_member() -> bool:
+    """Whether this process holds shards when shards are on. ``false`` makes it serve only."""
+    return os.getenv('BATCH_PROJECTION_SHARD_MEMBER', 'true').strip().lower() != 'false'
+
+
+def projection_backlog_reserve() -> int:
+    """Drain slots a shard holder keeps for backlog wakes while fresh wakes wait."""
+    return _bounded_int(
+        os.getenv('BATCH_PROJECTION_BACKLOG_RESERVE', ''),
+        PROJECTION_BACKLOG_RESERVE_DEFAULT,
+        0,
+        PROJECTION_BACKLOG_RESERVE_MAX,
+    )
+
+
+def projection_head_deadline_seconds() -> int:
+    """How long a failing or blocked chain head may retry before it is held (default 24h)."""
+    return _bounded_int(
+        os.getenv('BATCH_PROJECTION_HEAD_DEADLINE_SECONDS', ''),
+        PROJECTION_HEAD_DEADLINE_DEFAULT,
+        PROJECTION_HEAD_DEADLINE_MIN,
+        PROJECTION_HEAD_DEADLINE_MAX,
+    )
+
+
+def projection_blocked_head_deadline_enabled() -> bool:
+    """Default off: a chain head blocked past the deadline is held (audio kept, redrivable)."""
+    return rollout_default_off(
+        'BATCH_PROJECTION_BLOCKED_HEAD_DEADLINE_ENABLED',
+        os.getenv('BATCH_PROJECTION_BLOCKED_HEAD_DEADLINE_ENABLED', 'false').strip().lower() == 'true',
+    )

backend/config/flag_budget.json

@@ -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"},
     "BATCH_PIPELINE_PROJECTION_PUMP_ENABLED": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"},
@@ -37,6 +37,11 @@
     "PROJECTION_DRAIN_CLAIM_AWARE": {"owner": "two-lane lead", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"},
     "BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED": {"owner": "omi-pipeline", "review_by": "2026-12-06", "registry": "config/feature-flags.yaml"},
     "K_REVISION": {"owner": "omi-pipeline", "review_by": "2026-12-06"},
+    "BATCH_PROJECTION_SHARDS_ENABLED": {"owner": "two-lane lead", "review_by": "2026-12-05"},
+    "BATCH_PROJECTION_SHARD_MEMBER": {"owner": "two-lane lead", "review_by": "2026-12-05"},
+    "BATCH_PROJECTION_BACKLOG_RESERVE": {"owner": "two-lane lead", "review_by": "2026-12-05"},
+    "BATCH_PROJECTION_HEAD_DEADLINE_SECONDS": {"owner": "two-lane lead", "review_by": "2026-12-05"},
+    "BATCH_PROJECTION_BLOCKED_HEAD_DEADLINE_ENABLED": {"owner": "two-lane lead", "review_by": "2026-12-05"},
     "PARAKEET_SHAPE_CANARY_FILE": {"owner": "two-lane lead", "review_by": "2026-12-04"},
     "PARAKEET_LID_MODEL": {"owner": "planscost", "review_by": "2026-11-30"},
     "PARAKEET_LID_ONLY": {"owner": "planscost", "review_by": "2026-11-30"},

backend/database/batch_pipeline_store.py

@@ -359,6 +359,11 @@ def projection_page(
     def projection_successor(self, owner_key: str, previous: str | None) -> str | None:
         return projection_successor(self.client, self.namespace, owner_key, previous)
 
+    def projection_page_owners(
+        self, limit: int = 100, *, after=None, states: tuple[str, ...] = PROJECTION_DISCOVERY_STATES
+    ) -> tuple[list[tuple[str, str, float, float | None]], Any]:
+        return projection_page_owners(self.client, self.namespace, limit, after=after, states=states)
+
 
 def projection_pending(client: Any, namespace: str, limit: int = 100) -> list[str]:
     return projection_page(client, namespace, limit)[0]
@@ -396,6 +401,45 @@ def projection_page(
     return [doc.id for doc in docs], cursor
 
 
+def projection_page_owners(
+    client: Any,
+    namespace: str,
+    limit: int = 100,
+    *,
+    after=None,
+    states: tuple[str, ...] = PROJECTION_DISCOVERY_STATES,
+) -> tuple[list[tuple[str, str, float, float | None]], Any]:
+    """``projection_page`` returning ``(segment_id, owner_id, committed_at, capture_at)`` per row.
+
+    ``capture_at`` is None when the row lacks a numeric one.
+
+    The same query, index and cursor as ``projection_page``; the owner comes
+    from the committed result row itself, so routing a wake costs no extra read.
+    """
+    if not states or not set(states) <= set(PROJECTION_OBLIGATION_STATES):
+        raise ValueError('projection discovery states must be drain states')
+    collection = client.collection(f'{namespace}_results')
+    query = collection.where('projection_state', 'in', list(states)).order_by('committed_at').order_by('__name__')
+    if after is not None and after[1] == '':
+        query = query.start_at({'committed_at': after[0]})
+    elif after is not None:
+        query = query.start_after({'committed_at': after[0], '__name__': collection.document(after[1])})
+    size = max(1, min(limit, 100))
+    docs = list(query.limit(size).stream())
+    rows = []
+    for doc in docs:
+        data = doc.to_dict() or {}
+        owner_id = data.get('owner_id')
+        if isinstance(owner_id, str) and owner_id:
+            capture_at = data.get('capture_at')
+            if not isinstance(capture_at, (int, float)) or isinstance(capture_at, bool):
+                capture_at = None
+            rows.append((doc.id, owner_id, float(data.get('committed_at') or 0.0), capture_at))
+    # Same wrap rule as projection_page: a short page restarts at the head.
+    cursor = (docs[-1].to_dict()['committed_at'], docs[-1].id) if len(docs) == size else None
+    return rows, cursor
+
+
 def projection_successor(client: Any, namespace: str, owner_key: str, previous: str | None) -> str | None:
     """The owner's result committed directly after ``previous`` in its log.
 

backend/deploy/cloud_run_gmp_sync.yaml

@@ -43,7 +43,7 @@ spec:
         - action: keep
           sourceLabels:
             - __name__
-          regex: 'omi_(finalization_pair_.*|enrichment_summary_repair_.*|batch_record_pending_.*|record_seal_nojob_.*|enrichment_obligation_.*|enrichment_early_abort_total|enrichment_supersede_streak(_bucket|_count|_sum)?|finalization_byok_.*|listen_server_batch_suppression_total|listen_preview_batch_custody_total|delivery_outcome_total|people_scan_dedupe_total|pinecone_.*|playback_writes_total|sync_.*|journey_.*|client_journey_.*|owner_recognition_conversations_total|conversation_duration_seconds(_.*)?|conversation_relevance_decision_total|conversation_summary_vector_upserts_total|conversation_relevance_decision_total|firestore_documents_per_operation(_.*)?|capture_finalization_.*|queue_oldest_ready_.*|batch_pipeline_.*|batch_projection_drains_skipped_total|batch_projection_claim_busy_backoffs_total|batch_projection_page_collapsed_rows_total|batch_projection_drain_head_requeues_total|batch_intake_outcome_total|batch_admission_sample_window_total|batch_ramp_heavy_owner_excluded_total|batch_shadow_.*|accept_to_summary_seconds(_.*)?|summary_terminal_total|summary_debt_.*|live_session_transcript_outcome_total|fallback_total|parakeet_request_seconds(_.*)?|firestore_billed_reads_by_(caller|origin)_total|firestore_caller_label_overflow_total|firestore_billed_writes(_by_caller)?_total|firestore_write_caller_label_overflow_total|firestore_ambiguous_commit_writes_total|firestore_noop_writes_by_caller_total|memory_released_projection_total|fsr_leader_lease_total|held_content_.*|held_recovery_pacing_.*|stt_cap_recovery_reconciliation_total|stt_cap_recovery_cohort_narrowed|stt_cap_cohort_.*|stt_cap_launch_config_errors.*|stt_cap_dispatch_decisions_total|stt_cap_provider_audio_seconds_total|legacy_sync_.*|legacy_dark_.*|jobstore_.*|capture_coverage_.*|capture_dedupe_shadow_total|capture_retry_cap_total|speaker_evidence_.*|speaker_evidence_confident_owner_total|lifecycle_refactor_shadow_total|executor_task_(active|queued)|clients_refactor_shadow_total|notes_refactor_shadow_total|notes_refactor_results_total|stt_language_gate_.*|prerecorded_vendor_.*|episode_compose_.*|transcription_final_pass_total|transcription_final_pass_(total|unit_skips_total)|transcription_final_pass_(total|unit_skips_total|(agreement|latency_seconds|owner_delta_ratio)_(bucket|sum|count))|transcription_final_pass_(total|unit_skips_total|retry_terminal_(total|observed_at_seconds)|(agreement|latency_seconds|owner_delta_ratio)_(bucket|sum|count)))|conversation_llm_input_dedupe_.*|privacy_(checks|check_calls|check_reads|decision_mismatch|shadow_checks|memo_requests|memo_ineligible|ttl_cache)_total'
+          regex: 'omi_(finalization_pair_.*|enrichment_summary_repair_.*|batch_record_pending_.*|record_seal_nojob_.*|enrichment_obligation_.*|enrichment_early_abort_total|enrichment_supersede_streak(_bucket|_count|_sum)?|finalization_byok_.*|listen_server_batch_suppression_total|listen_preview_batch_custody_total|delivery_outcome_total|people_scan_dedupe_total|pinecone_.*|playback_writes_total|sync_.*|journey_.*|client_journey_.*|owner_recognition_conversations_total|conversation_duration_seconds(_.*)?|conversation_relevance_decision_total|conversation_summary_vector_upserts_total|conversation_relevance_decision_total|firestore_documents_per_operation(_.*)?|capture_finalization_.*|queue_oldest_ready_.*|batch_pipeline_.*|batch_projection_drains_skipped_total|batch_projection_claim_busy_backoffs_total|batch_projection_page_collapsed_rows_total|batch_projection_drain_head_requeues_total|batch_projection_blocked_head_holds_total|batch_projection_shards_(held|unheld)|batch_projection_shard_wakes_total|batch_projection_shard_wake_depth|batch_intake_outcome_total|batch_admission_sample_window_total|batch_ramp_heavy_owner_excluded_total|batch_shadow_.*|accept_to_summary_seconds(_.*)?|summary_terminal_total|summary_debt_.*|live_session_transcript_outcome_total|fallback_total|parakeet_request_seconds(_.*)?|firestore_billed_reads_by_(caller|origin)_total|firestore_caller_label_overflow_total|firestore_billed_writes(_by_caller)?_total|firestore_write_caller_label_overflow_total|firestore_ambiguous_commit_writes_total|firestore_noop_writes_by_caller_total|memory_released_projection_total|fsr_leader_lease_total|held_content_.*|held_recovery_pacing_.*|stt_cap_recovery_reconciliation_total|stt_cap_recovery_cohort_narrowed|stt_cap_cohort_.*|stt_cap_launch_config_errors.*|stt_cap_dispatch_decisions_total|stt_cap_provider_audio_seconds_total|legacy_sync_.*|legacy_dark_.*|jobstore_.*|capture_coverage_.*|capture_dedupe_shadow_total|capture_retry_cap_total|speaker_evidence_.*|speaker_evidence_confident_owner_total|lifecycle_refactor_shadow_total|executor_task_(active|queued)|clients_refactor_shadow_total|notes_refactor_shadow_total|notes_refactor_results_total|stt_language_gate_.*|prerecorded_vendor_.*|episode_compose_.*|transcription_final_pass_total|transcription_final_pass_(total|unit_skips_total)|transcription_final_pass_(total|unit_skips_total|(agreement|latency_seconds|owner_delta_ratio)_(bucket|sum|count))|transcription_final_pass_(total|unit_skips_total|retry_terminal_(total|observed_at_seconds)|(agreement|latency_seconds|owner_delta_ratio)_(bucket|sum|count)))|conversation_llm_input_dedupe_.*|privacy_(checks|check_calls|check_reads|decision_mismatch|shadow_checks|memo_requests|memo_ineligible|ttl_cache)_total'
         # These eagerly initialized live-STT combinations dominate empty series
         # on sync workers. Keep fallbacks from sync/summary/storage components.
         - action: drop

backend/docs/batch-pipeline-serving.md

@@ -1259,6 +1259,22 @@ below have their own switches; all three off is the pre-drain behaviour):
   advanced= more= elapsed_ms=` per turn, and `action=owner_chain
   outcome=claim_busy`.
 
+### 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.
+
+Only the maintenance leader pages the durable log, discovery-only, as a
+backstop. Follower instances skip the sweep and keep their async follow-up
+discovery. Holders fill drain slots fresh first, keeping
+`BATCH_PROJECTION_BACKLOG_RESERVE` slots for backlog.
+
+Any Redis failure returns that instance to the sweep. The contract, env
+bounds, cutover order and rollback are in
+[Batch projection owner shards](listen_pusher_pipeline.mdx#batch-projection-owner-shards).
+
 ### Follow-up desync (2026-11-05)
 
 Every serving process runs its own async follow-up discovery over the same