From 6108a6bb7a7da5a794948869bd0f8565ea3f1246 Mon Sep 17 00:00:00 2001 From: David Zhang Date: Thu, 5 Nov 2026 23:39:19 +0700 Subject: [PATCH 1/5] feat(batch): add opt-in dedicated projection drain pump --- backend/config/batch_pipeline.py | 26 ++ backend/config/flag_budget.json | 7 +- backend/config/policy/kill_switches.py | 10 +- backend/docs/feature-flag-registry.md | 10 + backend/docs/listen_pusher_pipeline.mdx | 18 ++ .../changes/2026-11-05-projection-pump.md | 19 ++ .../test_batch_pipeline_projection_pump.py | 235 ++++++++++++++++++ backend/tests/unit/test_batch_unit_policy.py | 2 +- backend/tests/unit/test_flag_budget.py | 13 +- .../utils/batch_pipeline/projection_drain.py | 3 +- backend/utils/batch_pipeline/serving.py | 48 +++- config/feature-flags.yaml | 5 + 12 files changed, 386 insertions(+), 10 deletions(-) create mode 100644 backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md create mode 100644 backend/tests/unit/test_batch_pipeline_projection_pump.py diff --git a/backend/config/batch_pipeline.py b/backend/config/batch_pipeline.py index df10e9bf34b..e0dff5afdb3 100644 --- a/backend/config/batch_pipeline.py +++ b/backend/config/batch_pipeline.py @@ -573,6 +573,32 @@ def owner_drain_enabled() -> bool: return kill_switch('BATCH_PIPELINE_OWNER_DRAIN_ENABLED') +def serving_projection_sweep_enabled() -> bool: + """Only ``off`` disables serving-instance discovery; owner wakes still run.""" + from config.policy import kill_switch + + return kill_switch('BATCH_PIPELINE_SERVING_PROJECTION_SWEEP') + + +def _pump_bound(raw: str, default: int, maximum: int) -> int: + try: + return max(1, min(int(raw), maximum)) + except ValueError: + return default + + +def projection_pump_tick_seconds() -> int: + return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_TICK_SECONDS', '20'), 20, 60) + + +def projection_pump_sweep_pages() -> int: + return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '1'), 1, 10) + + +def max_owner_drains() -> int: + return _pump_bound(os.getenv('BATCH_PIPELINE_MAX_OWNER_DRAINS', '8'), 8, 32) + + def projection_index_deferred_enabled() -> bool: """Kill switch: walks converge search/people-stats after the chain, once per conversation.""" from config.policy import kill_switch diff --git a/backend/config/flag_budget.json b/backend/config/flag_budget.json index 5a7b0686a0b..e4f584bd09b 100644 --- a/backend/config/flag_budget.json +++ b/backend/config/flag_budget.json @@ -24,8 +24,13 @@ }, "seed_review_by": "2026-11-15", "budget": 169, - "count": 248, + "count": 253, "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"}, + "BATCH_PIPELINE_PUMP_SWEEP_PAGES": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"}, + "BATCH_PIPELINE_PUMP_TICK_SECONDS": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"}, + "BATCH_PIPELINE_MAX_OWNER_DRAINS": {"owner": "omi-pipeline / lead2", "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"}, diff --git a/backend/config/policy/kill_switches.py b/backend/config/policy/kill_switches.py index b3a3a9dbc3c..7e53e1cc897 100644 --- a/backend/config/policy/kill_switches.py +++ b/backend/config/policy/kill_switches.py @@ -19,6 +19,7 @@ 'BATCH_PIPELINE_PROJECTION_PENDING_PAGE_ENABLED': True, 'BATCH_PIPELINE_PROJECTION_FOLLOWUP_PAGE_ENABLED': True, 'BATCH_PIPELINE_OWNER_DRAIN_ENABLED': True, + 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP': True, 'BATCH_PIPELINE_PROJECTION_INDEX_DEFERRED_ENABLED': True, 'PARAKEET_WINDOW_STRANDED_DRIFT_SETTLE_ENABLED': True, 'BATCH_PIPELINE_HELD_TERMINAL_STATUS_ENABLED': True, @@ -34,12 +35,17 @@ def kill_switch(name: str, *, policy_id: str = 'control', flags=None) -> bool: raise ValueError('unregistered live batch switch') env = os.environ if flags is None else flags value = env.get(name, str(KILL_SWITCHES[name])).strip().lower() - if name == 'PARAKEET_WINDOW_STRANDED_DRIFT_SETTLE_ENABLED': + if name == 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP': + enabled = value != 'off' + elif name == 'PARAKEET_WINDOW_STRANDED_DRIFT_SETTLE_ENABLED': enabled = value not in {'false', '0', 'off'} else: enabled = value != 'false' if name in DISABLE_ONLY_SWITCHES else value == 'true' from config.refactor_switches import default_on - enabled = default_on(name, enabled, environ=env) + # This instance selector deliberately accepts only "off", even during the + # shared boolean-parser migration; false/0/malformed keep today's sweep. + if name != 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP': + enabled = default_on(name, enabled, environ=env) logger.info('event=batch_policy_kill_switch policy_id=%s switch=%s enabled=%s count=1', policy_id, name, enabled) return enabled diff --git a/backend/docs/feature-flag-registry.md b/backend/docs/feature-flag-registry.md index 92f2b2e8f40..610c259d577 100644 --- a/backend/docs/feature-flag-registry.md +++ b/backend/docs/feature-flag-registry.md @@ -528,6 +528,7 @@ and an explicit empty literal renders as `''`. | `BATCH_PIPELINE_LANGUAGE_INHERITANCE_ENABLED` | Default on unless false; batch pieces inherit a known bound conversation language, else the user language, before per-piece LID | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_LEADER_LEASE_ENABLED` | Default on unless false; one batch worker or serving replica per Firestore lease runs periodic recovery, drain, cleanup and quarantine discovery | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_LONG_SPEECH_HANDOFF_ENABLED` | Default on unless false; a zero-word batch unit with >=5 s worker VAD speech whose intake STT language is unknown, multi or outside Parakeet TDT v3 skips the padded Parakeet retry and goes straight to quarantine recovery | backend | env | open | — | — | — | — | keep | keep | — | dazheng | +| `BATCH_PIPELINE_MAX_OWNER_DRAINS` | Default 8; concurrent owner drains per process, clamped to 1-32 at import; malformed keeps 8 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | | `BATCH_PIPELINE_MERGE_CATCH_UP_DEPTH` | Default 16; fresh batch uploads (capture ended at most 10 min before admission) commit ahead of the owner's pending catch-up rows, and catch-up commits wait while the owner already has this many committed rows awaiting projection | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_MERGE_HEAD_WAIT_SECONDS` | Default 60; once a batch result has waited this long behind an unready earlier-captured merge head, the owner's ready results commit in capture order and the head appends later as late_capture | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_OWNER_CHAIN_PROJECTION_ENABLED` | Default on; a settled quarantine recovery walks that owner's result log from its projection watermark in chain order and runs its contents' follow-ups first | backend | env | open | — | — | — | — | keep | keep | — | dazheng | @@ -542,7 +543,10 @@ and an explicit empty literal renders as `''`. | `BATCH_PIPELINE_PROJECTION_PENDING_PAGE_ENABLED` | Default on; the projection sweep's assignment page holds only pending results and a separate persisted/ready page finishes synchronous-policy receipts | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_PROJECTION_PREDECESSOR_REPLAY_ENABLED` | Default on; a blocked projection successor schedules its owner's missing predecessor through the existing ordered walk | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_PROJECTION_PREFETCH` | Default 4; an owner-chain walk reads the next N rows' successor links, job bindings, result payloads and audio manifests ahead of the chain, and runs the candidate lookup beside context and build | backend | env | open | — | — | — | — | keep | keep | — | dazheng | +| `BATCH_PIPELINE_PROJECTION_PUMP_ENABLED` | Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | | `BATCH_PIPELINE_PROJECTION_SOURCE_DELETED_ENABLED` | Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication | backend | env | open | — | — | — | — | keep | keep | — | cl2-cost / lead2 | +| `BATCH_PIPELINE_PUMP_SWEEP_PAGES` | Default 1; projection pump sweep pages per tick, clamped to 1-10; malformed keeps 1 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | +| `BATCH_PIPELINE_PUMP_TICK_SECONDS` | Default 20 seconds; projection pump tick sleep, clamped to 1-60; malformed keeps 20 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | | `BATCH_PIPELINE_RAMP_EXCLUDE_HEAVY_OWNERS` | Ship-on partial-ramp admission isolation; only off disables; owners over 30 open jobs or six hours pending audio use legacy | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED` | Default on unless false; the post-recovery owner wake commits the recovered unit, then its content's ready siblings, at once past an unready merge head (the unit itself also past the catch-up depth hold), so the same wake's chain walk projects them | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_RECOVERY_DISPATCH_INTERVAL_SECONDS` | Default 5; batch serving dispatches quarantine recovery on its own loop at this interval, on the maintenance leader only | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | @@ -552,6 +556,7 @@ and an explicit empty literal renders as `''`. | `BATCH_PIPELINE_RECOVERY_WAKE_LEASE_SECONDS` | One outstanding quarantine-recovery delivery task per paid intent; lease before a lost wake is repaired; default 600, 0 = legacy 120s wake on every sweep | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_SECOND_MULTILINGUAL_ENABLED` | Ship-on Soniox second chance after valid-empty Modulate held recovery; false restores Modulate-only ordinary recovery; independent language-guard recovery retains its separate switch and cap | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_SERVING_METRICS_MIN_REFRESH_SECONDS` | Default 30; each batch serving instance refreshes its queue gauges at most once per interval | backend | env | closed | — | — | 60 | — | keep | keep | — | dazheng | +| `BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` | Default on; only off skips Serving.reconcile projection sweep, preserving owner wakes, held drains, metrics and leader-leased maintenance | backend | env | open | — | — | — | — | keep | keep | — | lead2 | | `BATCH_PIPELINE_SHARED_EMBEDDING_TRUNK_ENABLED` | Default on unless false; the serial batch GPU chain runs the WeSpeaker embedding trunk once per 10 s chunk of a unit and pools it per local speaker, instead of once per (chunk, speaker) pair | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_SWEEP_OWNER_CHAIN_ENABLED` | Default on; when the periodic projection sweep advances an owner's chain head it keeps walking that owner's result log in chain order instead of waiting for its cursor to wrap | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_VAD_SEGMENTATION_REUSE_ENABLED` | Default on unless false; on the serial batch GPU chain a unit diarized at full length takes the raw segmentation scores its own VAD pass computed instead of a second segmentation forward | backend | env | open | — | — | — | — | keep | keep | — | dazheng | @@ -1084,6 +1089,7 @@ their code default (`fail` tells you which way a missing value resolves). - `BATCH_PIPELINE_LANGUAGE_INHERITANCE_ENABLED` — Default on unless false; batch pieces inherit a known bound conversation language, else the user language, before per-piece LID (fail: open) - `BATCH_PIPELINE_LEADER_LEASE_ENABLED` — Default on unless false; one batch worker or serving replica per Firestore lease runs periodic recovery, drain, cleanup and quarantine discovery (fail: open) - `BATCH_PIPELINE_LONG_SPEECH_HANDOFF_ENABLED` — Default on unless false; a zero-word batch unit with >=5 s worker VAD speech whose intake STT language is unknown, multi or outside Parakeet TDT v3 skips the padded Parakeet retry and goes straight to quarantine recovery (fail: open) +- `BATCH_PIPELINE_MAX_OWNER_DRAINS` — Default 8; concurrent owner drains per process, clamped to 1-32 at import; malformed keeps 8 (fail: closed) - `BATCH_PIPELINE_MERGE_CATCH_UP_DEPTH` — Default 16; fresh batch uploads (capture ended at most 10 min before admission) commit ahead of the owner's pending catch-up rows, and catch-up commits wait while the owner already has this many committed rows awaiting projection (fail: closed) - `BATCH_PIPELINE_MERGE_HEAD_WAIT_SECONDS` — Default 60; once a batch result has waited this long behind an unready earlier-captured merge head, the owner's ready results commit in capture order and the head appends later as late_capture (fail: closed) - `BATCH_PIPELINE_MODEL_BATCH_SIZE` — GPU chain maximum model window batch size; default 8; hard ceiling 32 (fail: closed) @@ -1101,8 +1107,11 @@ their code default (`fail` tells you which way a missing value resolves). - `BATCH_PIPELINE_PROJECTION_PENDING_PAGE_ENABLED` — Default on; the projection sweep's assignment page holds only pending results and a separate persisted/ready page finishes synchronous-policy receipts (fail: open) - `BATCH_PIPELINE_PROJECTION_PREDECESSOR_REPLAY_ENABLED` — Default on; a blocked projection successor schedules its owner's missing predecessor through the existing ordered walk (fail: open) - `BATCH_PIPELINE_PROJECTION_PREFETCH` — Default 4; an owner-chain walk reads the next N rows' successor links, job bindings, result payloads and audio manifests ahead of the chain, and runs the candidate lookup beside context and build (fail: open) +- `BATCH_PIPELINE_PROJECTION_PUMP_ENABLED` — Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED (fail: closed) - `BATCH_PIPELINE_PROJECTION_SOURCE_DELETED_ENABLED` — Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication (fail: open) - `BATCH_PIPELINE_PUBLISH_CONCURRENCY` — GPU worker post-flight publication concurrency: bounded fenced transitions and one ordered drain per owner per flight; default 4, 1 = legacy serial kill switch, max 16 (fail: closed) +- `BATCH_PIPELINE_PUMP_SWEEP_PAGES` — Default 1; projection pump sweep pages per tick, clamped to 1-10; malformed keeps 1 (fail: closed) +- `BATCH_PIPELINE_PUMP_TICK_SECONDS` — Default 20 seconds; projection pump tick sleep, clamped to 1-60; malformed keeps 20 (fail: closed) - `BATCH_PIPELINE_RAMP_EXCLUDE_HEAVY_OWNERS` — Ship-on partial-ramp admission isolation; only off disables; owners over 30 open jobs or six hours pending audio use legacy (fail: closed) - `BATCH_PIPELINE_RECORD_PROTOCOL_READY` — One-time native record protocol cutover; default closed retains record_pending obligations without expiry (fail: closed) - `BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED` — Default on unless false; the post-recovery owner wake commits the recovered unit, then its content's ready siblings, at once past an unready merge head (the unit itself also past the catch-up depth hold), so the same wake's chain walk projects them (fail: open) @@ -1113,6 +1122,7 @@ their code default (`fail` tells you which way a missing value resolves). - `BATCH_PIPELINE_RECOVERY_WAKE_LEASE_SECONDS` — One outstanding quarantine-recovery delivery task per paid intent; lease before a lost wake is repaired; default 600, 0 = legacy 120s wake on every sweep (fail: closed) - `BATCH_PIPELINE_REUSE_LIVE_ENABLED` — Default-off GPU-chain pre-STT reuse; requires existing lineage gates and full-span words-of-record quality proof (fail: closed) - `BATCH_PIPELINE_SECOND_MULTILINGUAL_ENABLED` — Ship-on Soniox second chance after valid-empty Modulate held recovery; false restores Modulate-only ordinary recovery; independent language-guard recovery retains its separate switch and cap (fail: open) +- `BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` — Default on; only off skips Serving.reconcile projection sweep, preserving owner wakes, held drains, metrics and leader-leased maintenance (fail: open) - `BATCH_PIPELINE_SHADOW_LANGUAGE_ONLY` — Default false; classify one canonical batch unit per selected legacy upload instead of entering the idle GPU shadow; no route or publication (fail: closed) - `BATCH_PIPELINE_SHADOW_LANGUAGE_PROBES_PER_HOUR` — Default and maximum 500 fleet-wide pair attempts per UTC hour; zero or malformed denies; minute pacing and two 120-second slots also apply (fail: closed) - `BATCH_PIPELINE_SHADOW_RECOVERY_API_URL` — Explicit HTTPS existing record-recovery CPU role for shadow recovery; URL is also its OIDC audience; blank fails closed (fail: closed) diff --git a/backend/docs/listen_pusher_pipeline.mdx b/backend/docs/listen_pusher_pipeline.mdx index 2a71e486aa3..3ccb527d131 100644 --- a/backend/docs/listen_pusher_pipeline.mdx +++ b/backend/docs/listen_pusher_pipeline.mdx @@ -9299,3 +9299,21 @@ frames, one automatic notes-only refresh is claimed. A user title prevents both the claim and a concurrent refresh commit. An all-rejected pass may still refresh late OCR/calendar context, but rejected frame bytes never reach notes. This refresh does not run derived memory, app, embedding, or speech-learning jobs. + +### Dedicated projection pump + +`BATCH_PIPELINE_PROJECTION_PUMP_ENABLED` defaults to `false`; only `true` +selects one projection-only loop in place of the serving runtime's three loops, +even when `BATCH_PIPELINE_SERVING_ENABLED` is unset/false. It uses the same +stage/store/artifact construction and never calls reconcile, maintenance or +recovery dispatch. `BATCH_PIPELINE_PUMP_SWEEP_PAGES` defaults to 1 (clamped +1-10) and `BATCH_PIPELINE_PUMP_TICK_SECONDS` to 20 seconds (clamped 1-60). +Malformed numeric values use those defaults. Each tick sweeps pages through +`sync_executor`, starts owed owner drains on the event loop and sleeps. + +`BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` defaults on; only `off` skips the +serving reconcile sweep. Owner wakes, held drains, metrics refresh, maintenance +and recovery dispatch keep their existing behavior. The pump ignores this +serving-only switch. `BATCH_PIPELINE_MAX_OWNER_DRAINS` sets the process's +owner drain capacity at import (default 8, clamped 1-32; malformed uses 8). +[Change and ramp](listen_pusher_pipeline/changes/2026-11-05-projection-pump.md). diff --git a/backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md b/backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md new file mode 100644 index 00000000000..6206f8f36d2 --- /dev/null +++ b/backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md @@ -0,0 +1,19 @@ +# Dedicated projection discovery pump + +The [serving projection contract](../../listen_pusher_pipeline.mdx#dedicated-projection-pump) +can select one projection-only loop without starting serving maintenance, +record admission or quarantine dispatch. Unset env preserves the existing three +serving loops, sweep cadence and eight owner drains. No deploy configuration is +changed here; service shape remains a separate decision. + +The dedicated loop offloads each sweep to `sync_executor`, starts owed owner +drains on the event loop, and emits a heartbeat with successful sweep count, +requested owners and elapsed time on every tick, including failed ticks. Failures +log their exception type and retry on the next tick. Durable projection receipts +and advisory owner claims are unchanged. + +Ramp owner: lead2, tonight (2026-11-05). Start with a pump service of N=3-4, +verify owner_drain rows/s stays at least 70% of pre-cutover and claim_busy ratio +does not exceed pre-cutover, then disable serving sweeps only after at least +30 minutes stable. Rollback: restore serving sweeps, disable the pump, and return +owner drain capacity to eight. This PR only adds code and unit coverage. diff --git a/backend/tests/unit/test_batch_pipeline_projection_pump.py b/backend/tests/unit/test_batch_pipeline_projection_pump.py new file mode 100644 index 00000000000..b6ae77a4760 --- /dev/null +++ b/backend/tests/unit/test_batch_pipeline_projection_pump.py @@ -0,0 +1,235 @@ +"""Dedicated pump selection, bounded capacity, failure containment and loop handoff.""" + +import asyncio +import importlib +import logging +import threading +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from config import batch_pipeline as config +from tests.unit.test_batch_pipeline_scheduler import MemoryStore +from utils.batch_pipeline import projection_drain, serving + + +@pytest.mark.parametrize('mode', ['shadow', 'serve']) +@pytest.mark.parametrize( + 'raw,enabled', + [ + (None, True), + ('off', False), + (' OFF ', False), + ('false', True), + ('0', True), + ('', True), + ('bad', True), + ('on', True), + ], +) +def test_serving_sweep_only_off_disables(monkeypatch, mode, raw, enabled): + monkeypatch.setenv('CAPTURE_COVERAGE_MODE', mode) + name = 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP' + if raw is None: + monkeypatch.delenv(name, raising=False) + else: + monkeypatch.setenv(name, raw) + assert config.serving_projection_sweep_enabled() is enabled + + +@pytest.mark.parametrize('raw,default', [(None, True), ('off', False)]) +def test_serving_sweep_gate_preserves_other_reconcile_work(monkeypatch, raw, default): + if raw is None: + monkeypatch.delenv('BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', raising=False) + else: + monkeypatch.setenv('BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', raw) + service = serving.Serving(MemoryStore(), projector=Mock(), metrics=Mock()) + service._maintain = Mock() + service._reconcile_owner = Mock(return_value=['followup']) + service.merge.held_owners.add('held') + service.drain_owner = Mock() + monkeypatch.setattr(serving, 'owner_chain_projection_enabled', lambda: True) + assert service.reconcile(periodic=True, owner_id='wake', recovered='unit') == ['followup'] + assert service.projector.sweep.call_count == int(default) + service._maintain.assert_called_once_with(recover_quarantine=False) + service._reconcile_owner.assert_called_once_with('wake', 'unit') + service.drain_owner.assert_called_once_with('held') + service.metrics.refresh.assert_called_once() + + +@pytest.mark.parametrize( + 'name,reader,default,maximum', + [ + ('BATCH_PIPELINE_PUMP_SWEEP_PAGES', config.projection_pump_sweep_pages, 1, 10), + ('BATCH_PIPELINE_PUMP_TICK_SECONDS', config.projection_pump_tick_seconds, 20, 60), + ('BATCH_PIPELINE_MAX_OWNER_DRAINS', config.max_owner_drains, 8, 32), + ], +) +@pytest.mark.parametrize('raw', [None, '', 'bad', '2.5', '-10', '0', '3', '999']) +def test_pump_numeric_env_bounds(monkeypatch, name, reader, default, maximum, raw): + if raw is None: + monkeypatch.delenv(name, raising=False) + else: + monkeypatch.setenv(name, raw) + expected = default if raw in (None, '', 'bad', '2.5') else max(1, min(int(raw), maximum)) + assert reader() == expected + + +@pytest.mark.parametrize('raw,expected', [(None, 8), ('bad', 8), ('', 8), ('0', 1), ('999', 32), ('16', 16)]) +def test_owner_drain_module_capacity_from_env(monkeypatch, raw, expected): + try: + with monkeypatch.context() as patch: + if raw is None: + patch.delenv('BATCH_PIPELINE_MAX_OWNER_DRAINS', raising=False) + else: + patch.setenv('BATCH_PIPELINE_MAX_OWNER_DRAINS', raw) + importlib.reload(projection_drain) + assert projection_drain.MAX_OWNER_DRAINS == expected + finally: + importlib.reload(projection_drain) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('serving_enabled', ['false', 'true']) +@pytest.mark.parametrize( + 'pages,tick,expected_pages,expected_tick', + [(None, None, 1, 20), ('3', '4', 3, 4), ('999', '999', 10, 60), ('0', '0', 1, 1), ('bad', 'bad', 1, 20)], +) +async def test_pump_only_sweeps_and_hands_drains_to_loop( + monkeypatch, caplog, serving_enabled, pages, tick, expected_pages, expected_tick +): + monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'true') + monkeypatch.setenv('BATCH_PIPELINE_SERVING_ENABLED', serving_enabled) + # Pump discovery is independent of the serving-instance sweep kill switch. + monkeypatch.setenv('BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', 'off') + for name, raw in [('BATCH_PIPELINE_PUMP_SWEEP_PAGES', pages), ('BATCH_PIPELINE_PUMP_TICK_SECONDS', tick)]: + if raw is None: + monkeypatch.delenv(name, raising=False) + else: + monkeypatch.setenv(name, raw) + loop_thread = threading.get_ident() + projector = SimpleNamespace(handoff=threading.Lock(), drain_requests={}, owes_drains=False) + sweep_calls = [] + + def sweep(): + assert threading.get_ident() != loop_thread + sweep_calls.append(1) + projector.drain_requests['owner'] = None + projector.owes_drains = True + + def start_drains(): + assert threading.get_ident() == loop_thread + assert len(sweep_calls) == expected_pages + projector.drain_requests.clear() + projector.owes_drains = False + + projector.sweep = Mock(side_effect=sweep) + projector.start_drains = Mock(side_effect=start_drains) + service = SimpleNamespace( + projector=projector, reconcile=Mock(), reconcile_async=Mock(), _maintain=Mock(), dispatch_recovery=Mock() + ) + monkeypatch.setattr(serving, 'current_stage', lambda: 'dev') + factory = Mock(return_value=service) + monkeypatch.setattr(serving, 'serving_for_stage', factory) + record_drain = Mock() + monkeypatch.setattr(serving, 'drain_recordings', record_drain) + timeouts = [] + + async def wait_for(awaitable, *, timeout): + timeouts.append(timeout) + runtime[0].set() + return await awaitable + + monkeypatch.setattr( + serving, 'asyncio', SimpleNamespace(Event=asyncio.Event, wait_for=wait_for, TimeoutError=asyncio.TimeoutError) + ) + caplog.set_level(logging.INFO, logger=serving.__name__) + runtime = serving.start_serving_from_env() + await asyncio.wait_for(runtime[1], timeout=5) + assert timeouts == [expected_tick] + factory.assert_called_once_with('dev') + assert projector.sweep.call_count == expected_pages + projector.start_drains.assert_called_once() + for forbidden in ( + service.reconcile, + service.reconcile_async, + service._maintain, + service.dispatch_recovery, + record_drain, + ): + forbidden.assert_not_called() + assert f'outcome=tick sweeps={expected_pages} owners_requested=1 elapsed_seconds=' in caplog.text + + +@pytest.mark.asyncio +@pytest.mark.parametrize('failure', ['build', 'sweep', 'drains']) +async def test_pump_logs_failure_and_runs_next_tick(monkeypatch, caplog, failure): + monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'true') + monkeypatch.setenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '1') + projector = SimpleNamespace( + handoff=threading.Lock(), drain_requests={}, owes_drains=True, sweep=Mock(), start_drains=Mock() + ) + service = SimpleNamespace(projector=projector) + factory = Mock(return_value=service) + target = {'build': factory, 'sweep': projector.sweep, 'drains': projector.start_drains}[failure] + target.side_effect = [RuntimeError('synthetic'), None if failure != 'build' else service] + monkeypatch.setattr(serving, 'serving_for_stage', factory) + monkeypatch.setattr(serving, 'current_stage', lambda: 'dev') + ticks = [] + + async def wait_for(awaitable, *, timeout): + ticks.append(timeout) + if len(ticks) == 2: + runtime[0].set() + return await awaitable + awaitable.close() + raise asyncio.TimeoutError + + monkeypatch.setattr( + serving, 'asyncio', SimpleNamespace(Event=asyncio.Event, wait_for=wait_for, TimeoutError=asyncio.TimeoutError) + ) + caplog.set_level(logging.INFO, logger=serving.__name__) + runtime = serving.start_serving_from_env() + await asyncio.wait_for(runtime[1], timeout=5) + assert len(ticks) == 2 + assert factory.call_count == 2 + assert 'event=batch_projection_pump outcome=retry exception_type=RuntimeError' in caplog.text + assert caplog.text.count('event=batch_projection_pump outcome=tick') == 2 + assert projector.start_drains.call_count >= 1 + + +@pytest.mark.parametrize('raw', [None, 'false', '', 'bad', '1']) +def test_disabled_pump_does_not_start_without_serving(monkeypatch, raw): + monkeypatch.delenv('BATCH_PIPELINE_SERVING_ENABLED', raising=False) + if raw is None: + monkeypatch.delenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', raising=False) + else: + monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', raw) + factory = Mock() + monkeypatch.setattr(serving, 'serving_for_stage', factory) + assert serving.start_serving_from_env() is None + factory.assert_not_called() + + +@pytest.mark.asyncio +async def test_idle_pump_stops_during_tick_sleep_without_starting_drains(monkeypatch): + monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'true') + monkeypatch.setenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '1') + swept = asyncio.Event() + projector = SimpleNamespace(sweep=Mock(side_effect=swept.set), owes_drains=False, start_drains=Mock()) + monkeypatch.setattr(serving, 'serving_for_stage', lambda stage: SimpleNamespace(projector=projector)) + monkeypatch.setattr(serving, 'current_stage', lambda: 'dev') + + async def run_blocking(executor, fn, *args): + return fn(*args) + + monkeypatch.setattr(serving, 'run_blocking', run_blocking) + runtime = serving.start_serving_from_env() + try: + await asyncio.wait_for(swept.wait(), timeout=5) + finally: + await serving.stop_serving(runtime) + projector.sweep.assert_called_once() + projector.start_drains.assert_not_called() + assert runtime[1].done() diff --git a/backend/tests/unit/test_batch_unit_policy.py b/backend/tests/unit/test_batch_unit_policy.py index e7f23d66e59..991bcae3d83 100644 --- a/backend/tests/unit/test_batch_unit_policy.py +++ b/backend/tests/unit/test_batch_unit_policy.py @@ -87,7 +87,7 @@ def test_cohort_registry_and_live_switch_contract(): for key in KILL_SWITCHES: assert by_key[key]['lifecycle'] == 'ops_kill' assert kill_switch(key, flags={}) - assert not kill_switch(key, flags={key: 'false'}) + assert not kill_switch(key, flags={key: 'off' if key == 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP' else 'false'}) assert 'nested' in by_key['BATCH_LIVE_STT_RETIRE_PERCENT']['notes'].lower() with pytest.raises(ValueError): kill_switch('BATCH_PIPELINE_ENABLED') diff --git a/backend/tests/unit/test_flag_budget.py b/backend/tests/unit/test_flag_budget.py index 9d6d61212c8..70086751a13 100644 --- a/backend/tests/unit/test_flag_budget.py +++ b/backend/tests/unit/test_flag_budget.py @@ -188,7 +188,11 @@ def test_helper_kill_switch_budget_entries_match_the_default_on_registry(): # This rollout is explicitly registered although config/server_batch.py is # outside the pipeline scanner's scope. Keep helper kill switches separate. rollout_name = 'SERVER_BATCH_ONLY_WEB_PERCENT' - assert helper_names == HELPER_KILL_SWITCHES | {'SERVER_BATCH_UNSUPPORTED_LANGUAGES', rollout_name} + assert helper_names == HELPER_KILL_SWITCHES | { + 'SERVER_BATCH_UNSUPPORTED_LANGUAGES', + rollout_name, + 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', + } # Keep this inventory check inside the fast-unit CPU budget; the C safe # loader has the same safe semantics and parses the full registry faster. @@ -196,6 +200,13 @@ def test_helper_kill_switch_budget_entries_match_the_default_on_registry(): FEATURE_FLAGS_PATH.read_text(encoding='utf-8'), Loader=getattr(yaml, 'CSafeLoader', yaml.SafeLoader) ) flags = {flag['key']: flag for flag in registry['flags']} + sweep_name = 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP' + assert names[sweep_name]['owner'] == 'omi-pipeline / lead2' + assert names[sweep_name]['review_by'] == '2026-12-05' + assert flags[sweep_name]['owner'] == 'lead2' + assert flags[sweep_name]['kind'] == 'env' + assert flags[sweep_name]['lifecycle'] == 'ops_kill' + assert flags[sweep_name]['fail'] == 'open' unsupported = names['SERVER_BATCH_UNSUPPORTED_LANGUAGES'] assert unsupported['owner'] == 'planscost' assert unsupported['review_by'] == '2026-11-30' diff --git a/backend/utils/batch_pipeline/projection_drain.py b/backend/utils/batch_pipeline/projection_drain.py index c065fbb4cae..2a395acf78b 100644 --- a/backend/utils/batch_pipeline/projection_drain.py +++ b/backend/utils/batch_pipeline/projection_drain.py @@ -16,6 +16,7 @@ import time from typing import TYPE_CHECKING +from config.batch_pipeline import max_owner_drains from utils.executors import run_blocking, start_background_task, sync_executor if TYPE_CHECKING: @@ -23,7 +24,7 @@ logger = logging.getLogger(__name__) # Owners draining at once in one process. -MAX_OWNER_DRAINS = 8 +MAX_OWNER_DRAINS = max_owner_drains() # One owner's turn; with more rows it re-queues behind every waiting owner. DRAIN_TURN_SECONDS = 120.0 # One blocking step: bounds a sync-pool slot and the owner-claim hold. diff --git a/backend/utils/batch_pipeline/serving.py b/backend/utils/batch_pipeline/serving.py index f65408bd986..b9c91323c98 100644 --- a/backend/utils/batch_pipeline/serving.py +++ b/backend/utils/batch_pipeline/serving.py @@ -15,6 +15,9 @@ recovery_dispatch_interval_seconds, projection_hold_recovery_enabled, serving_metrics_min_refresh_seconds, + serving_projection_sweep_enabled, + projection_pump_sweep_pages, + projection_pump_tick_seconds, ) from database.outbox import OutboxKind, outbox_for from database.batch_privacy_attempt import with_batch_privacy_attempt @@ -189,7 +192,7 @@ def reconcile( # Projection and the local gauges stay per instance: projection is # throughput (ordered and fenced in its own transactions) and every # instance exports its own queue sample. - if self.projector: + if self.projector and serving_projection_sweep_enabled(): self.projector.sweep() # Catch-up rows this instance held for projection depth may now fit. for held_owner in list(self.merge.held_owners): @@ -364,10 +367,42 @@ def _metrics() -> QueueMetrics: def start_serving_from_env(): """Separate drain switch stays on when BATCH_PIPELINE_ENABLED rolls back.""" - if os.getenv('BATCH_PIPELINE_SERVING_ENABLED', 'false').strip().lower() != 'true': + pump_enabled = os.getenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'false').strip().lower() == 'true' + if not pump_enabled and os.getenv('BATCH_PIPELINE_SERVING_ENABLED', 'false').strip().lower() != 'true': return None stop = asyncio.Event() + async def projection_pump(): + # Dedicated instances only discover and drain owners. They never enter + # reconcile or either leader-leased maintenance/recovery path. + while not stop.is_set(): + started = time.monotonic() + sweeps = owners_requested = 0 + try: + service = await run_blocking(sync_executor, serving_for_stage, current_stage()) + for _ in range(projection_pump_sweep_pages()): + await run_blocking(sync_executor, service.projector.sweep) + sweeps += 1 + if service.projector.owes_drains: + with service.projector.handoff: + owners_requested = len(service.projector.drain_requests) + service.projector.start_drains() + except Exception as error: + logging.getLogger(__name__).error( + 'event=batch_projection_pump outcome=retry exception_type=%s', type(error).__name__ + ) + finally: + logging.getLogger(__name__).info( + 'event=batch_projection_pump outcome=tick sweeps=%d owners_requested=%d elapsed_seconds=%.3f', + sweeps, + owners_requested, + time.monotonic() - started, + ) + try: + await asyncio.wait_for(stop.wait(), timeout=projection_pump_tick_seconds()) + except asyncio.TimeoutError: + pass + async def supervise(): while not stop.is_set(): try: @@ -427,9 +462,14 @@ async def dispatch_recovery(): pass async def run(): - await asyncio.gather(supervise(), drain(), dispatch_recovery()) + if pump_enabled: + await projection_pump() + else: + await asyncio.gather(supervise(), drain(), dispatch_recovery()) - return stop, start_background_task(run(), name='batch_pipeline_serving') + return stop, start_background_task( + run(), name='batch_projection_pump' if pump_enabled else 'batch_pipeline_serving' + ) async def stop_serving(runtime): diff --git a/config/feature-flags.yaml b/config/feature-flags.yaml index 1705fb624d2..8e5d44329e0 100644 --- a/config/feature-flags.yaml +++ b/config/feature-flags.yaml @@ -205,6 +205,11 @@ flags: - {"key":"BATCH_PIPELINE_RECOVERY_OWNER_ONLY_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; post-recovery requests drain only the recovered owner and leave global scans/retries to periodic serving; false restores global reconciliation","fail":"open","owner":"two-lane rollout lead","created":"2026-11-03","terminal":"keep","decision":"keep","notes":"Periodic serving supervision retains global maintenance, projection discovery and queue sampling. The owner request retains fenced merge, projection, targeted durable follow-ups and owner-scoped index work. Pending jobs and result receipts remain the durable retry owner after process loss. Watch stage=post_recovery latency, maintenance aborts, prepared/held age and queue slope. See backend/docs/batch-pipeline-serving.md."} - {"key":"BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on unless false; the post-recovery owner wake commits the recovered unit, then its content's ready siblings, at once past an unready merge head (the unit itself also past the catch-up depth hold), so the same wake's chain walk projects them","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","ramp":{"cohorts":["5% run 19","25%","50%","100%"],"clean_metric":"hash-cohort quarantine_projection_p95 <= 300 s and event=batch_merge outcome=recovered_release present with no new privacy or chain errors","expander":"two-lane lead session","done_by":"2026-10-24"},"notes":"5% run 18 (2026-10-20): 46 of 54 post-recovery owner wakes advanced no row because the recovered unit, whose ready_at is its settle time, still sat behind a leased head for the 60 s head wait plus a maintenance tick (settled -> commit 65-86 s); projection then waited for the 20 s serving sweep. Only the wake passes a release set (the recovered unit plus up to 32 content siblings); each row must still be the owner's pending, ready job inside the commit transaction, and the commit body, late marking, log chain and projection order are unchanged. false restores the head wait and depth hold for recovered units. Log event=batch_merge outcome=recovered_release (closed labels); results carry recovered_release: true. See backend/docs/batch-pipeline-serving.md."} - {"key":"BATCH_PIPELINE_SWEEP_OWNER_CHAIN_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; when the periodic projection sweep advances an owner's chain head it keeps walking that owner's result log in chain order instead of waiting for its cursor to wrap","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","notes":"False restores one attempt per page row. One walk per owner per sweep, 64 rows / 30 s each and 30 s per sweep across owners. Every row still takes the owner claim and transactional predecessor, privacy and receipt fences; no new per-owner concurrency. See backend/docs/batch-pipeline-serving.md."} + - {"key":"BATCH_PIPELINE_SERVING_PROJECTION_SWEEP","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; only off skips Serving.reconcile projection sweep, preserving owner wakes, held drains, metrics and leader-leased maintenance","fail":"open","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} + - {"key":"BATCH_PIPELINE_PROJECTION_PUMP_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} + - {"key":"BATCH_PIPELINE_PUMP_SWEEP_PAGES","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 1; projection pump sweep pages per tick, clamped to 1-10; malformed keeps 1","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} + - {"key":"BATCH_PIPELINE_PUMP_TICK_SECONDS","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 20 seconds; projection pump tick sleep, clamped to 1-60; malformed keeps 20","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} + - {"key":"BATCH_PIPELINE_MAX_OWNER_DRAINS","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 8; concurrent owner drains per process, clamped to 1-32 at import; malformed keeps 8","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} - {"key":"BATCH_PIPELINE_OWNER_DRAIN_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; a per-process owner drain walks each discovered owner's result log continuously (up to 8 owners at once, 120 s turns, FIFO re-queue) instead of the sweep-shared 30 s walk budget","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","notes":"False restores the sweep owner-chain walk with every pre-check; read-ahead (BATCH_PIPELINE_PROJECTION_PREFETCH=0) and the deferred index (BATCH_PIPELINE_PROJECTION_INDEX_DEFERRED_ENABLED=false) have their own switches, and all three off is the pre-drain behaviour. On: the sweep only discovers owners; each walk holds the advisory owner claim for its 15 s step and skips the predecessor/per-part pre-check transactions for rows it reached as the watermark's successor. The assignment transaction's predecessor, privacy and receipt fences stay authoritative; one owner never runs two chains. See backend/docs/batch-pipeline-serving.md."} - {"key":"BATCH_PIPELINE_HELD_TERMINAL_STATUS_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; terminally held batch uploads retain phone custody as processing/batch_unit_held with a five-minute poll hint","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","ramp":{"cohorts":["all batch uploads at deploy (default on)"],"clean_metric":"Zero completed status responses for uploads with unresolved held units; projected siblings remain visible and held polls are paced at five minutes","expander":"dazheng (two-lane lead)","done_by":"2026-10-27"},"notes":"Read-only status projection; no ledger, job or retention write. Phones keep their independent WAL copy until actual completion; server artifact retention alone never permits acknowledgement. Open operator intents or in-flight siblings also remain processing. False restores failed/batch_worker_quarantined and still never acknowledges custody. See backend/docs/batch-pipeline-serving.md."} - {"key":"BATCH_PIPELINE_RECOVERY_POLICY_REFRESH_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; a quarantined held unit whose recorded recovery policy is disabled and which has no recovery intent (no paid call authorized) re-resolves the recovery setting at sweep or dispatch, and adopts it only when it is now enabled","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","ramp":{"cohorts":["all held units at deploy (default on)"],"clean_metric":"recovery sweep outcome=disabled count falls to zero and paid recovery starts stay within the per-owner allowance","expander":"dazheng (two-lane lead)","done_by":"2026-10-27"},"notes":"Spends at most the one at-most-once paid recovery the unit would have had if admitted while recovery was on; the existing allowance/cap gates and RecoveryStore.start fence still apply. An enabled snapshot is never re-resolved or downgraded. False restores the frozen snapshot (outcome=disabled forever). See backend/docs/batch-pipeline-serving.md."} From 85b564880f3ba519a2a62156baadad6810ad26c4 Mon Sep 17 00:00:00 2001 From: David Zhang Date: Thu, 5 Nov 2026 23:47:49 +0700 Subject: [PATCH 2/5] refactor(batch): narrow projector before running pump tick --- backend/utils/batch_pipeline/serving.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/backend/utils/batch_pipeline/serving.py b/backend/utils/batch_pipeline/serving.py index b9c91323c98..d14f11ff0cb 100644 --- a/backend/utils/batch_pipeline/serving.py +++ b/backend/utils/batch_pipeline/serving.py @@ -380,13 +380,16 @@ async def projection_pump(): sweeps = owners_requested = 0 try: service = await run_blocking(sync_executor, serving_for_stage, current_stage()) + projector = service.projector + if projector is None: + raise RuntimeError('projection pump requires a projector') for _ in range(projection_pump_sweep_pages()): - await run_blocking(sync_executor, service.projector.sweep) + await run_blocking(sync_executor, projector.sweep) sweeps += 1 - if service.projector.owes_drains: - with service.projector.handoff: - owners_requested = len(service.projector.drain_requests) - service.projector.start_drains() + if projector.owes_drains: + with projector.handoff: + owners_requested = len(projector.drain_requests) + projector.start_drains() except Exception as error: logging.getLogger(__name__).error( 'event=batch_projection_pump outcome=retry exception_type=%s', type(error).__name__ From 36db876ba63559f9182675a56b1eb6436df8bff1 Mon Sep 17 00:00:00 2001 From: David Zhang Date: Fri, 6 Nov 2026 00:04:01 +0700 Subject: [PATCH 3/5] refactor(batch): resolve pump role through policy config --- backend/config/batch_pipeline.py | 7 ++++++ backend/config/flag_budget.json | 2 +- backend/config/policy/kill_switches.py | 8 ++++--- .../test_batch_pipeline_projection_pump.py | 24 +++++++++++++++++++ backend/tests/unit/test_batch_unit_policy.py | 2 +- backend/tests/unit/test_flag_budget.py | 8 +++++++ backend/utils/batch_pipeline/serving.py | 3 ++- 7 files changed, 48 insertions(+), 6 deletions(-) diff --git a/backend/config/batch_pipeline.py b/backend/config/batch_pipeline.py index e0dff5afdb3..0ce2ba91459 100644 --- a/backend/config/batch_pipeline.py +++ b/backend/config/batch_pipeline.py @@ -587,6 +587,13 @@ def _pump_bound(raw: str, default: int, maximum: int) -> int: return default +def projection_pump_enabled() -> bool: + """Process-role opt-in: only ``true`` starts the otherwise inert pump.""" + from config.policy import kill_switch + + return kill_switch('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED') + + def projection_pump_tick_seconds() -> int: return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_TICK_SECONDS', '20'), 20, 60) diff --git a/backend/config/flag_budget.json b/backend/config/flag_budget.json index e4f584bd09b..1ee8a7e9f93 100644 --- a/backend/config/flag_budget.json +++ b/backend/config/flag_budget.json @@ -27,7 +27,7 @@ "count": 253, "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"}, + "BATCH_PIPELINE_PROJECTION_PUMP_ENABLED": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"}, "BATCH_PIPELINE_PUMP_SWEEP_PAGES": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"}, "BATCH_PIPELINE_PUMP_TICK_SECONDS": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"}, "BATCH_PIPELINE_MAX_OWNER_DRAINS": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"}, diff --git a/backend/config/policy/kill_switches.py b/backend/config/policy/kill_switches.py index 7e53e1cc897..94c539c810d 100644 --- a/backend/config/policy/kill_switches.py +++ b/backend/config/policy/kill_switches.py @@ -20,6 +20,7 @@ 'BATCH_PIPELINE_PROJECTION_FOLLOWUP_PAGE_ENABLED': True, 'BATCH_PIPELINE_OWNER_DRAIN_ENABLED': True, 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP': True, + 'BATCH_PIPELINE_PROJECTION_PUMP_ENABLED': False, 'BATCH_PIPELINE_PROJECTION_INDEX_DEFERRED_ENABLED': True, 'PARAKEET_WINDOW_STRANDED_DRIFT_SETTLE_ENABLED': True, 'BATCH_PIPELINE_HELD_TERMINAL_STATUS_ENABLED': True, @@ -43,9 +44,10 @@ def kill_switch(name: str, *, policy_id: str = 'control', flags=None) -> bool: enabled = value != 'false' if name in DISABLE_ONLY_SWITCHES else value == 'true' from config.refactor_switches import default_on - # This instance selector deliberately accepts only "off", even during the - # shared boolean-parser migration; false/0/malformed keep today's sweep. - if name != 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP': + # Instance selectors retain their explicit token grammar even during the + # shared boolean-parser migration: only off skips serving sweeps, and only + # true opts a process into the otherwise inert dedicated pump. + if name not in {'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', 'BATCH_PIPELINE_PROJECTION_PUMP_ENABLED'}: enabled = default_on(name, enabled, environ=env) logger.info('event=batch_policy_kill_switch policy_id=%s switch=%s enabled=%s count=1', policy_id, name, enabled) return enabled diff --git a/backend/tests/unit/test_batch_pipeline_projection_pump.py b/backend/tests/unit/test_batch_pipeline_projection_pump.py index b6ae77a4760..e3f45725494 100644 --- a/backend/tests/unit/test_batch_pipeline_projection_pump.py +++ b/backend/tests/unit/test_batch_pipeline_projection_pump.py @@ -38,6 +38,30 @@ def test_serving_sweep_only_off_disables(monkeypatch, mode, raw, enabled): assert config.serving_projection_sweep_enabled() is enabled +@pytest.mark.parametrize('mode', ['shadow', 'serve']) +@pytest.mark.parametrize( + 'raw,enabled', + [ + (None, False), + ('true', True), + (' TRUE ', True), + ('off', False), + ('false', False), + ('0', False), + ('', False), + ('bad', False), + ('1', False), + ], +) +def test_projection_pump_policy_opt_in(monkeypatch, mode, raw, enabled): + monkeypatch.setenv('CAPTURE_COVERAGE_MODE', mode) + if raw is None: + monkeypatch.delenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', raising=False) + else: + monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', raw) + assert config.projection_pump_enabled() is enabled + + @pytest.mark.parametrize('raw,default', [(None, True), ('off', False)]) def test_serving_sweep_gate_preserves_other_reconcile_work(monkeypatch, raw, default): if raw is None: diff --git a/backend/tests/unit/test_batch_unit_policy.py b/backend/tests/unit/test_batch_unit_policy.py index 991bcae3d83..4616d561a42 100644 --- a/backend/tests/unit/test_batch_unit_policy.py +++ b/backend/tests/unit/test_batch_unit_policy.py @@ -86,7 +86,7 @@ def test_cohort_registry_and_live_switch_contract(): by_key = {row['key']: row for row in rows} for key in KILL_SWITCHES: assert by_key[key]['lifecycle'] == 'ops_kill' - assert kill_switch(key, flags={}) + assert kill_switch(key, flags={}) is KILL_SWITCHES[key] assert not kill_switch(key, flags={key: 'off' if key == 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP' else 'false'}) assert 'nested' in by_key['BATCH_LIVE_STT_RETIRE_PERCENT']['notes'].lower() with pytest.raises(ValueError): diff --git a/backend/tests/unit/test_flag_budget.py b/backend/tests/unit/test_flag_budget.py index 70086751a13..c210335043b 100644 --- a/backend/tests/unit/test_flag_budget.py +++ b/backend/tests/unit/test_flag_budget.py @@ -192,6 +192,7 @@ def test_helper_kill_switch_budget_entries_match_the_default_on_registry(): 'SERVER_BATCH_UNSUPPORTED_LANGUAGES', rollout_name, 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', + 'BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', } # Keep this inventory check inside the fast-unit CPU budget; the C safe @@ -200,6 +201,13 @@ def test_helper_kill_switch_budget_entries_match_the_default_on_registry(): FEATURE_FLAGS_PATH.read_text(encoding='utf-8'), Loader=getattr(yaml, 'CSafeLoader', yaml.SafeLoader) ) flags = {flag['key']: flag for flag in registry['flags']} + pump_name = 'BATCH_PIPELINE_PROJECTION_PUMP_ENABLED' + assert names[pump_name]['owner'] == 'omi-pipeline / lead2' + assert names[pump_name]['review_by'] == '2026-12-05' + assert flags[pump_name]['owner'] == 'lead2' + assert flags[pump_name]['kind'] == 'env' + assert flags[pump_name]['lifecycle'] == 'ops_kill' + assert flags[pump_name]['fail'] == 'closed' sweep_name = 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP' assert names[sweep_name]['owner'] == 'omi-pipeline / lead2' assert names[sweep_name]['review_by'] == '2026-12-05' diff --git a/backend/utils/batch_pipeline/serving.py b/backend/utils/batch_pipeline/serving.py index d14f11ff0cb..3b27f53b71e 100644 --- a/backend/utils/batch_pipeline/serving.py +++ b/backend/utils/batch_pipeline/serving.py @@ -16,6 +16,7 @@ projection_hold_recovery_enabled, serving_metrics_min_refresh_seconds, serving_projection_sweep_enabled, + projection_pump_enabled, projection_pump_sweep_pages, projection_pump_tick_seconds, ) @@ -367,7 +368,7 @@ def _metrics() -> QueueMetrics: def start_serving_from_env(): """Separate drain switch stays on when BATCH_PIPELINE_ENABLED rolls back.""" - pump_enabled = os.getenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'false').strip().lower() == 'true' + pump_enabled = projection_pump_enabled() if not pump_enabled and os.getenv('BATCH_PIPELINE_SERVING_ENABLED', 'false').strip().lower() != 'true': return None stop = asyncio.Event() From e1cc45d27b5042de8cabb672a47c2e6f13362786 Mon Sep 17 00:00:00 2001 From: David Zhang Date: Fri, 6 Nov 2026 00:22:27 +0700 Subject: [PATCH 4/5] fix(batch): schedule pump follow-ups and cap discovery load --- backend/config/batch_pipeline.py | 10 +- backend/docs/feature-flag-registry.md | 12 +- backend/docs/listen_pusher_pipeline.mdx | 11 +- .../changes/2026-11-05-projection-pump.md | 29 +++- .../test_batch_pipeline_projection_pump.py | 138 ++++++++++++++++-- backend/utils/batch_pipeline/metrics.py | 5 + backend/utils/batch_pipeline/serving.py | 6 + config/feature-flags.yaml | 10 +- 8 files changed, 180 insertions(+), 41 deletions(-) diff --git a/backend/config/batch_pipeline.py b/backend/config/batch_pipeline.py index 0ce2ba91459..c97a971c425 100644 --- a/backend/config/batch_pipeline.py +++ b/backend/config/batch_pipeline.py @@ -580,9 +580,9 @@ def serving_projection_sweep_enabled() -> bool: return kill_switch('BATCH_PIPELINE_SERVING_PROJECTION_SWEEP') -def _pump_bound(raw: str, default: int, maximum: int) -> int: +def _pump_bound(raw: str, default: int, maximum: int, minimum: int = 1) -> int: try: - return max(1, min(int(raw), maximum)) + return max(minimum, min(int(raw), maximum)) except ValueError: return default @@ -595,15 +595,15 @@ def projection_pump_enabled() -> bool: def projection_pump_tick_seconds() -> int: - return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_TICK_SECONDS', '20'), 20, 60) + return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_TICK_SECONDS', '20'), 20, 60, 2) def projection_pump_sweep_pages() -> int: - return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '1'), 1, 10) + return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '1'), 1, 4) def max_owner_drains() -> int: - return _pump_bound(os.getenv('BATCH_PIPELINE_MAX_OWNER_DRAINS', '8'), 8, 32) + return _pump_bound(os.getenv('BATCH_PIPELINE_MAX_OWNER_DRAINS', '8'), 8, 16) def projection_index_deferred_enabled() -> bool: diff --git a/backend/docs/feature-flag-registry.md b/backend/docs/feature-flag-registry.md index 610c259d577..7d0d7d9f00f 100644 --- a/backend/docs/feature-flag-registry.md +++ b/backend/docs/feature-flag-registry.md @@ -528,7 +528,7 @@ and an explicit empty literal renders as `''`. | `BATCH_PIPELINE_LANGUAGE_INHERITANCE_ENABLED` | Default on unless false; batch pieces inherit a known bound conversation language, else the user language, before per-piece LID | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_LEADER_LEASE_ENABLED` | Default on unless false; one batch worker or serving replica per Firestore lease runs periodic recovery, drain, cleanup and quarantine discovery | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_LONG_SPEECH_HANDOFF_ENABLED` | Default on unless false; a zero-word batch unit with >=5 s worker VAD speech whose intake STT language is unknown, multi or outside Parakeet TDT v3 skips the padded Parakeet retry and goes straight to quarantine recovery | backend | env | open | — | — | — | — | keep | keep | — | dazheng | -| `BATCH_PIPELINE_MAX_OWNER_DRAINS` | Default 8; concurrent owner drains per process, clamped to 1-32 at import; malformed keeps 8 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | +| `BATCH_PIPELINE_MAX_OWNER_DRAINS` | Default 8; concurrent owner drains per process, clamped to 1-16 at import; malformed keeps 8 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | | `BATCH_PIPELINE_MERGE_CATCH_UP_DEPTH` | Default 16; fresh batch uploads (capture ended at most 10 min before admission) commit ahead of the owner's pending catch-up rows, and catch-up commits wait while the owner already has this many committed rows awaiting projection | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_MERGE_HEAD_WAIT_SECONDS` | Default 60; once a batch result has waited this long behind an unready earlier-captured merge head, the owner's ready results commit in capture order and the head appends later as late_capture | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_OWNER_CHAIN_PROJECTION_ENABLED` | Default on; a settled quarantine recovery walks that owner's result log from its projection watermark in chain order and runs its contents' follow-ups first | backend | env | open | — | — | — | — | keep | keep | — | dazheng | @@ -545,8 +545,8 @@ and an explicit empty literal renders as `''`. | `BATCH_PIPELINE_PROJECTION_PREFETCH` | Default 4; an owner-chain walk reads the next N rows' successor links, job bindings, result payloads and audio manifests ahead of the chain, and runs the candidate lookup beside context and build | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_PROJECTION_PUMP_ENABLED` | Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | | `BATCH_PIPELINE_PROJECTION_SOURCE_DELETED_ENABLED` | Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication | backend | env | open | — | — | — | — | keep | keep | — | cl2-cost / lead2 | -| `BATCH_PIPELINE_PUMP_SWEEP_PAGES` | Default 1; projection pump sweep pages per tick, clamped to 1-10; malformed keeps 1 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | -| `BATCH_PIPELINE_PUMP_TICK_SECONDS` | Default 20 seconds; projection pump tick sleep, clamped to 1-60; malformed keeps 20 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | +| `BATCH_PIPELINE_PUMP_SWEEP_PAGES` | Default 1; projection pump sweep pages per tick, clamped to 1-4; malformed keeps 1 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | +| `BATCH_PIPELINE_PUMP_TICK_SECONDS` | Default 20 seconds; projection pump tick sleep, clamped to 2-60; malformed keeps 20 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 | | `BATCH_PIPELINE_RAMP_EXCLUDE_HEAVY_OWNERS` | Ship-on partial-ramp admission isolation; only off disables; owners over 30 open jobs or six hours pending audio use legacy | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED` | Default on unless false; the post-recovery owner wake commits the recovered unit, then its content's ready siblings, at once past an unready merge head (the unit itself also past the catch-up depth hold), so the same wake's chain walk projects them | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_RECOVERY_DISPATCH_INTERVAL_SECONDS` | Default 5; batch serving dispatches quarantine recovery on its own loop at this interval, on the maintenance leader only | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | @@ -1089,7 +1089,7 @@ their code default (`fail` tells you which way a missing value resolves). - `BATCH_PIPELINE_LANGUAGE_INHERITANCE_ENABLED` — Default on unless false; batch pieces inherit a known bound conversation language, else the user language, before per-piece LID (fail: open) - `BATCH_PIPELINE_LEADER_LEASE_ENABLED` — Default on unless false; one batch worker or serving replica per Firestore lease runs periodic recovery, drain, cleanup and quarantine discovery (fail: open) - `BATCH_PIPELINE_LONG_SPEECH_HANDOFF_ENABLED` — Default on unless false; a zero-word batch unit with >=5 s worker VAD speech whose intake STT language is unknown, multi or outside Parakeet TDT v3 skips the padded Parakeet retry and goes straight to quarantine recovery (fail: open) -- `BATCH_PIPELINE_MAX_OWNER_DRAINS` — Default 8; concurrent owner drains per process, clamped to 1-32 at import; malformed keeps 8 (fail: closed) +- `BATCH_PIPELINE_MAX_OWNER_DRAINS` — Default 8; concurrent owner drains per process, clamped to 1-16 at import; malformed keeps 8 (fail: closed) - `BATCH_PIPELINE_MERGE_CATCH_UP_DEPTH` — Default 16; fresh batch uploads (capture ended at most 10 min before admission) commit ahead of the owner's pending catch-up rows, and catch-up commits wait while the owner already has this many committed rows awaiting projection (fail: closed) - `BATCH_PIPELINE_MERGE_HEAD_WAIT_SECONDS` — Default 60; once a batch result has waited this long behind an unready earlier-captured merge head, the owner's ready results commit in capture order and the head appends later as late_capture (fail: closed) - `BATCH_PIPELINE_MODEL_BATCH_SIZE` — GPU chain maximum model window batch size; default 8; hard ceiling 32 (fail: closed) @@ -1110,8 +1110,8 @@ their code default (`fail` tells you which way a missing value resolves). - `BATCH_PIPELINE_PROJECTION_PUMP_ENABLED` — Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED (fail: closed) - `BATCH_PIPELINE_PROJECTION_SOURCE_DELETED_ENABLED` — Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication (fail: open) - `BATCH_PIPELINE_PUBLISH_CONCURRENCY` — GPU worker post-flight publication concurrency: bounded fenced transitions and one ordered drain per owner per flight; default 4, 1 = legacy serial kill switch, max 16 (fail: closed) -- `BATCH_PIPELINE_PUMP_SWEEP_PAGES` — Default 1; projection pump sweep pages per tick, clamped to 1-10; malformed keeps 1 (fail: closed) -- `BATCH_PIPELINE_PUMP_TICK_SECONDS` — Default 20 seconds; projection pump tick sleep, clamped to 1-60; malformed keeps 20 (fail: closed) +- `BATCH_PIPELINE_PUMP_SWEEP_PAGES` — Default 1; projection pump sweep pages per tick, clamped to 1-4; malformed keeps 1 (fail: closed) +- `BATCH_PIPELINE_PUMP_TICK_SECONDS` — Default 20 seconds; projection pump tick sleep, clamped to 2-60; malformed keeps 20 (fail: closed) - `BATCH_PIPELINE_RAMP_EXCLUDE_HEAVY_OWNERS` — Ship-on partial-ramp admission isolation; only off disables; owners over 30 open jobs or six hours pending audio use legacy (fail: closed) - `BATCH_PIPELINE_RECORD_PROTOCOL_READY` — One-time native record protocol cutover; default closed retains record_pending obligations without expiry (fail: closed) - `BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED` — Default on unless false; the post-recovery owner wake commits the recovered unit, then its content's ready siblings, at once past an unready merge head (the unit itself also past the catch-up depth hold), so the same wake's chain walk projects them (fail: open) diff --git a/backend/docs/listen_pusher_pipeline.mdx b/backend/docs/listen_pusher_pipeline.mdx index 3ccb527d131..67e83579d92 100644 --- a/backend/docs/listen_pusher_pipeline.mdx +++ b/backend/docs/listen_pusher_pipeline.mdx @@ -9307,13 +9307,18 @@ selects one projection-only loop in place of the serving runtime's three loops, even when `BATCH_PIPELINE_SERVING_ENABLED` is unset/false. It uses the same stage/store/artifact construction and never calls reconcile, maintenance or recovery dispatch. `BATCH_PIPELINE_PUMP_SWEEP_PAGES` defaults to 1 (clamped -1-10) and `BATCH_PIPELINE_PUMP_TICK_SECONDS` to 20 seconds (clamped 1-60). +1-4) and `BATCH_PIPELINE_PUMP_TICK_SECONDS` to 20 seconds (clamped 2-60). Malformed numeric values use those defaults. Each tick sweeps pages through -`sync_executor`, starts owed owner drains on the event loop and sleeps. +`sync_executor`, schedules discovered or owed projection follow-ups with +`start_followups()` and schedules the owed index flush after each sweep, starts +owed owner drains on the event loop after the pages, and sleeps. Each completed +tick, including an error tick, updates the Unix timestamp gauge +`omi_batch_projection_pump_last_tick_seconds`; zero means never observed. +Scope the missing/stale alert to the dedicated pump service. `BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` defaults on; only `off` skips the serving reconcile sweep. Owner wakes, held drains, metrics refresh, maintenance and recovery dispatch keep their existing behavior. The pump ignores this serving-only switch. `BATCH_PIPELINE_MAX_OWNER_DRAINS` sets the process's -owner drain capacity at import (default 8, clamped 1-32; malformed uses 8). +owner drain capacity at import (default 8, clamped 1-16; malformed uses 8). [Change and ramp](listen_pusher_pipeline/changes/2026-11-05-projection-pump.md). diff --git a/backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md b/backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md index 6206f8f36d2..d4783f13fd6 100644 --- a/backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md +++ b/backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md @@ -6,14 +6,29 @@ record admission or quarantine dispatch. Unset env preserves the existing three serving loops, sweep cadence and eight owner drains. No deploy configuration is changed here; service shape remains a separate decision. -The dedicated loop offloads each sweep to `sync_executor`, starts owed owner -drains on the event loop, and emits a heartbeat with successful sweep count, +The dedicated loop offloads each sweep to `sync_executor`, schedules discovered or owed +projection follow-ups with `start_followups()` and the owed index flush after +each sweep, starts owed owner drains on the event loop, and emits a heartbeat with successful sweep count, requested owners and elapsed time on every tick, including failed ticks. Failures log their exception type and retry on the next tick. Durable projection receipts and advisory owner claims are unchanged. -Ramp owner: lead2, tonight (2026-11-05). Start with a pump service of N=3-4, -verify owner_drain rows/s stays at least 70% of pre-cutover and claim_busy ratio -does not exceed pre-cutover, then disable serving sweeps only after at least -30 minutes stable. Rollback: restore serving sweeps, disable the pump, and return -owner drain capacity to eight. This PR only adds code and unit coverage. +Hard bounds: pages 1-4, tick sleep 2-60 seconds, owner drains 1-16. Defaults +remain pages 1, tick 20 seconds and owner drains 8; malformed env keeps defaults. + +Ramp owner: lead2, after David's deployment-shape decision. Safe start: a +**dedicated service**, N=4 instances, pages 1, tick 2 seconds, MAX_OWNER_DRAINS 8. +Never enable the pump on backend-integration itself: the selector is service-wide. +Serving sweeps stay ON until owner_drain advanced is at least 20 rows/s for +30 minutes. Before switching them off, install and verify a dedicated-service +missing/stale alert on `omi_batch_projection_pump_last_tick_seconds` (Unix time, +zero before the first completed tick; updated even on failed ticks). + +Retain the read-rate alarm above +50% of pre-cutover and the owner_drain rows/s +floor of 70% of pre-cutover. Watch sweep elapsed time and executor saturation. +Evaluate claim_busy share (no higher than pre-cutover) after serving sweeps are +switched off, not during dual-running. If the gate is unmet, serving sweeps stay ON. + +Rollback is env only, through new revisions: restore serving sweeps ON on the +existing service **first**, then set pump=false on the dedicated service and +owner drain capacity back to 8. This PR only adds code and unit coverage. diff --git a/backend/tests/unit/test_batch_pipeline_projection_pump.py b/backend/tests/unit/test_batch_pipeline_projection_pump.py index e3f45725494..283f93933b4 100644 --- a/backend/tests/unit/test_batch_pipeline_projection_pump.py +++ b/backend/tests/unit/test_batch_pipeline_projection_pump.py @@ -5,11 +5,15 @@ import logging import threading from types import SimpleNamespace -from unittest.mock import Mock +from unittest.mock import AsyncMock, Mock import pytest +from prometheus_client import CollectorRegistry + from config import batch_pipeline as config +from config.policy import resolve_stage_policy +from utils.batch_pipeline.metrics import QueueMetrics from tests.unit.test_batch_pipeline_scheduler import MemoryStore from utils.batch_pipeline import projection_drain, serving @@ -83,24 +87,24 @@ def test_serving_sweep_gate_preserves_other_reconcile_work(monkeypatch, raw, def @pytest.mark.parametrize( - 'name,reader,default,maximum', + 'name,reader,default,minimum,maximum', [ - ('BATCH_PIPELINE_PUMP_SWEEP_PAGES', config.projection_pump_sweep_pages, 1, 10), - ('BATCH_PIPELINE_PUMP_TICK_SECONDS', config.projection_pump_tick_seconds, 20, 60), - ('BATCH_PIPELINE_MAX_OWNER_DRAINS', config.max_owner_drains, 8, 32), + ('BATCH_PIPELINE_PUMP_SWEEP_PAGES', config.projection_pump_sweep_pages, 1, 1, 4), + ('BATCH_PIPELINE_PUMP_TICK_SECONDS', config.projection_pump_tick_seconds, 20, 2, 60), + ('BATCH_PIPELINE_MAX_OWNER_DRAINS', config.max_owner_drains, 8, 1, 16), ], ) -@pytest.mark.parametrize('raw', [None, '', 'bad', '2.5', '-10', '0', '3', '999']) -def test_pump_numeric_env_bounds(monkeypatch, name, reader, default, maximum, raw): +@pytest.mark.parametrize('raw', [None, '', 'bad', '2.5', '-10', '0', '1', '3', '999']) +def test_pump_numeric_env_bounds(monkeypatch, name, reader, default, minimum, maximum, raw): if raw is None: monkeypatch.delenv(name, raising=False) else: monkeypatch.setenv(name, raw) - expected = default if raw in (None, '', 'bad', '2.5') else max(1, min(int(raw), maximum)) + expected = default if raw in (None, '', 'bad', '2.5') else max(minimum, min(int(raw), maximum)) assert reader() == expected -@pytest.mark.parametrize('raw,expected', [(None, 8), ('bad', 8), ('', 8), ('0', 1), ('999', 32), ('16', 16)]) +@pytest.mark.parametrize('raw,expected', [(None, 8), ('bad', 8), ('', 8), ('0', 1), ('999', 16), ('16', 16)]) def test_owner_drain_module_capacity_from_env(monkeypatch, raw, expected): try: with monkeypatch.context() as patch: @@ -118,7 +122,7 @@ def test_owner_drain_module_capacity_from_env(monkeypatch, raw, expected): @pytest.mark.parametrize('serving_enabled', ['false', 'true']) @pytest.mark.parametrize( 'pages,tick,expected_pages,expected_tick', - [(None, None, 1, 20), ('3', '4', 3, 4), ('999', '999', 10, 60), ('0', '0', 1, 1), ('bad', 'bad', 1, 20)], + [(None, None, 1, 20), ('3', '4', 3, 4), ('999', '999', 4, 60), ('0', '0', 1, 2), ('bad', 'bad', 1, 20)], ) async def test_pump_only_sweeps_and_hands_drains_to_loop( monkeypatch, caplog, serving_enabled, pages, tick, expected_pages, expected_tick @@ -133,7 +137,14 @@ async def test_pump_only_sweeps_and_hands_drains_to_loop( else: monkeypatch.setenv(name, raw) loop_thread = threading.get_ident() - projector = SimpleNamespace(handoff=threading.Lock(), drain_requests={}, owes_drains=False) + projector = SimpleNamespace( + handoff=threading.Lock(), + drain_requests={}, + owes_drains=False, + has_async_followups=False, + owes_followups=False, + start_followups=AsyncMock(), + ) sweep_calls = [] def sweep(): @@ -175,6 +186,7 @@ async def wait_for(awaitable, *, timeout): factory.assert_called_once_with('dev') assert projector.sweep.call_count == expected_pages projector.start_drains.assert_called_once() + projector.start_followups.assert_not_awaited() for forbidden in ( service.reconcile, service.reconcile_async, @@ -187,16 +199,28 @@ async def wait_for(awaitable, *, timeout): @pytest.mark.asyncio -@pytest.mark.parametrize('failure', ['build', 'sweep', 'drains']) +@pytest.mark.parametrize('failure', ['build', 'sweep', 'followups', 'drains']) async def test_pump_logs_failure_and_runs_next_tick(monkeypatch, caplog, failure): monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'true') monkeypatch.setenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '1') projector = SimpleNamespace( - handoff=threading.Lock(), drain_requests={}, owes_drains=True, sweep=Mock(), start_drains=Mock() + handoff=threading.Lock(), + drain_requests={}, + owes_drains=True, + sweep=Mock(), + start_drains=Mock(), + has_async_followups=False, + owes_followups=failure == 'followups', + start_followups=AsyncMock(), ) service = SimpleNamespace(projector=projector) factory = Mock(return_value=service) - target = {'build': factory, 'sweep': projector.sweep, 'drains': projector.start_drains}[failure] + target = { + 'build': factory, + 'sweep': projector.sweep, + 'followups': projector.start_followups, + 'drains': projector.start_drains, + }[failure] target.side_effect = [RuntimeError('synthetic'), None if failure != 'build' else service] monkeypatch.setattr(serving, 'serving_for_stage', factory) monkeypatch.setattr(serving, 'current_stage', lambda: 'dev') @@ -213,9 +237,12 @@ async def wait_for(awaitable, *, timeout): monkeypatch.setattr( serving, 'asyncio', SimpleNamespace(Event=asyncio.Event, wait_for=wait_for, TimeoutError=asyncio.TimeoutError) ) + metrics = Mock() + monkeypatch.setattr(serving, '_metrics', lambda: metrics) caplog.set_level(logging.INFO, logger=serving.__name__) runtime = serving.start_serving_from_env() await asyncio.wait_for(runtime[1], timeout=5) + assert metrics.projection_pump_last_tick.set.call_count == 2 assert len(ticks) == 2 assert factory.call_count == 2 assert 'event=batch_projection_pump outcome=retry exception_type=RuntimeError' in caplog.text @@ -241,7 +268,14 @@ async def test_idle_pump_stops_during_tick_sleep_without_starting_drains(monkeyp monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'true') monkeypatch.setenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '1') swept = asyncio.Event() - projector = SimpleNamespace(sweep=Mock(side_effect=swept.set), owes_drains=False, start_drains=Mock()) + projector = SimpleNamespace( + sweep=Mock(side_effect=swept.set), + owes_drains=False, + start_drains=Mock(), + has_async_followups=False, + owes_followups=False, + start_followups=AsyncMock(), + ) monkeypatch.setattr(serving, 'serving_for_stage', lambda stage: SimpleNamespace(projector=projector)) monkeypatch.setattr(serving, 'current_stage', lambda: 'dev') @@ -257,3 +291,77 @@ async def run_blocking(executor, fn, *args): projector.sweep.assert_called_once() projector.start_drains.assert_not_called() assert runtime[1].done() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('hint', ['last_page', 'first_page', 'owed']) +async def test_pump_starts_followups_without_serving_sweeps_or_maintenance(monkeypatch, hint): + monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'true') + monkeypatch.setenv('BATCH_PIPELINE_SERVING_ENABLED', 'false') + monkeypatch.setenv('BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', 'off') + monkeypatch.setenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '2') + monkeypatch.setenv('BATCH_PIPELINE_ASYNC_PROJECTION_ENABLED', 'true') + assert resolve_stage_policy('owner', 'projection').async_projection is True + registry = CollectorRegistry() + metrics = QueueMetrics(registry) + monkeypatch.setattr(serving, '_metrics', lambda: metrics) + assert registry.get_sample_value('omi_batch_projection_pump_last_tick_seconds') == 0 + loop_thread = threading.get_ident() + calls = [] + projector = SimpleNamespace(has_async_followups=False, owes_followups=False, owes_drains=False, owes_index=False) + pages = [] + + def sweep(): + assert threading.get_ident() != loop_thread + calls.append('sweep') + pages.append(1) + # The per-page hint is reset just as Projector.sweep resets it. + projector.has_async_followups = (hint == 'first_page' and len(pages) == 1) or ( + hint == 'last_page' and len(pages) == 2 + ) + projector.owes_followups = hint == 'owed' + projector.owes_index = True + + async def start_followups(priority): + assert threading.get_ident() == loop_thread + assert priority == () + assert calls[-1] == 'sweep' + calls.append('followups') + + def schedule_index_flush(): + assert threading.get_ident() == loop_thread + calls.append('index') + projector.owes_index = False + + projector.sweep = Mock(side_effect=sweep) + projector.schedule_index_flush = Mock(side_effect=schedule_index_flush) + projector.start_followups = AsyncMock(side_effect=start_followups) + projector.start_drains = Mock() + service = SimpleNamespace( + projector=projector, reconcile=Mock(), reconcile_async=Mock(), _maintain=Mock(), dispatch_recovery=Mock() + ) + monkeypatch.setattr(serving, 'serving_for_stage', lambda stage: service) + monkeypatch.setattr(serving, 'current_stage', lambda: 'dev') + + async def wait_for(awaitable, *, timeout): + runtime[0].set() + return await awaitable + + monkeypatch.setattr( + serving, 'asyncio', SimpleNamespace(Event=asyncio.Event, wait_for=wait_for, TimeoutError=asyncio.TimeoutError) + ) + runtime = serving.start_serving_from_env() + await asyncio.wait_for(runtime[1], timeout=5) + assert projector.start_followups.await_count == (2 if hint == 'owed' else 1) + assert all(call.args == ((),) for call in projector.start_followups.await_args_list) + assert projector.schedule_index_flush.call_count == 2 + expected = { + 'first_page': ['sweep', 'followups', 'index', 'sweep', 'index'], + 'last_page': ['sweep', 'index', 'sweep', 'followups', 'index'], + 'owed': ['sweep', 'followups', 'index', 'sweep', 'followups', 'index'], + } + assert calls == expected[hint] + assert registry.get_sample_value('omi_batch_projection_pump_last_tick_seconds') > 0 + projector.start_drains.assert_not_called() + for forbidden in (service.reconcile, service.reconcile_async, service._maintain, service.dispatch_recovery): + forbidden.assert_not_called() diff --git a/backend/utils/batch_pipeline/metrics.py b/backend/utils/batch_pipeline/metrics.py index 3e3a4cfcfdf..a7dfb45032f 100644 --- a/backend/utils/batch_pipeline/metrics.py +++ b/backend/utils/batch_pipeline/metrics.py @@ -47,6 +47,11 @@ def __init__( self._refresh_due = 0.0 self._refresh_lock = threading.Lock() self.capacity_sampling = capacity_sampling + self.projection_pump_last_tick = Gauge( + 'omi_batch_projection_pump_last_tick_seconds', + 'Unix timestamp of the last completed projection pump tick, including failed ticks; zero before first tick', + registry=registry, + ) self.projection_age = Gauge( 'omi_batch_projection_obligation_oldest_age_seconds', 'Oldest unfinished committed projection age, including held content; alarm above 900s', diff --git a/backend/utils/batch_pipeline/serving.py b/backend/utils/batch_pipeline/serving.py index 3b27f53b71e..d9b594f76e9 100644 --- a/backend/utils/batch_pipeline/serving.py +++ b/backend/utils/batch_pipeline/serving.py @@ -376,6 +376,7 @@ def start_serving_from_env(): async def projection_pump(): # Dedicated instances only discover and drain owners. They never enter # reconcile or either leader-leased maintenance/recovery path. + metrics = _metrics() while not stop.is_set(): started = time.monotonic() sweeps = owners_requested = 0 @@ -387,6 +388,10 @@ async def projection_pump(): for _ in range(projection_pump_sweep_pages()): await run_blocking(sync_executor, projector.sweep) sweeps += 1 + if projector.has_async_followups or getattr(projector, 'owes_followups', False) is True: + await projector.start_followups(()) + if getattr(projector, 'owes_index', False) is True: + projector.schedule_index_flush() if projector.owes_drains: with projector.handoff: owners_requested = len(projector.drain_requests) @@ -396,6 +401,7 @@ async def projection_pump(): 'event=batch_projection_pump outcome=retry exception_type=%s', type(error).__name__ ) finally: + metrics.projection_pump_last_tick.set(time.time()) logging.getLogger(__name__).info( 'event=batch_projection_pump outcome=tick sweeps=%d owners_requested=%d elapsed_seconds=%.3f', sweeps, diff --git a/config/feature-flags.yaml b/config/feature-flags.yaml index 8e5d44329e0..1e3442fbe6f 100644 --- a/config/feature-flags.yaml +++ b/config/feature-flags.yaml @@ -205,11 +205,11 @@ flags: - {"key":"BATCH_PIPELINE_RECOVERY_OWNER_ONLY_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; post-recovery requests drain only the recovered owner and leave global scans/retries to periodic serving; false restores global reconciliation","fail":"open","owner":"two-lane rollout lead","created":"2026-11-03","terminal":"keep","decision":"keep","notes":"Periodic serving supervision retains global maintenance, projection discovery and queue sampling. The owner request retains fenced merge, projection, targeted durable follow-ups and owner-scoped index work. Pending jobs and result receipts remain the durable retry owner after process loss. Watch stage=post_recovery latency, maintenance aborts, prepared/held age and queue slope. See backend/docs/batch-pipeline-serving.md."} - {"key":"BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on unless false; the post-recovery owner wake commits the recovered unit, then its content's ready siblings, at once past an unready merge head (the unit itself also past the catch-up depth hold), so the same wake's chain walk projects them","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","ramp":{"cohorts":["5% run 19","25%","50%","100%"],"clean_metric":"hash-cohort quarantine_projection_p95 <= 300 s and event=batch_merge outcome=recovered_release present with no new privacy or chain errors","expander":"two-lane lead session","done_by":"2026-10-24"},"notes":"5% run 18 (2026-10-20): 46 of 54 post-recovery owner wakes advanced no row because the recovered unit, whose ready_at is its settle time, still sat behind a leased head for the 60 s head wait plus a maintenance tick (settled -> commit 65-86 s); projection then waited for the 20 s serving sweep. Only the wake passes a release set (the recovered unit plus up to 32 content siblings); each row must still be the owner's pending, ready job inside the commit transaction, and the commit body, late marking, log chain and projection order are unchanged. false restores the head wait and depth hold for recovered units. Log event=batch_merge outcome=recovered_release (closed labels); results carry recovered_release: true. See backend/docs/batch-pipeline-serving.md."} - {"key":"BATCH_PIPELINE_SWEEP_OWNER_CHAIN_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; when the periodic projection sweep advances an owner's chain head it keeps walking that owner's result log in chain order instead of waiting for its cursor to wrap","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","notes":"False restores one attempt per page row. One walk per owner per sweep, 64 rows / 30 s each and 30 s per sweep across owners. Every row still takes the owner claim and transactional predecessor, privacy and receipt fences; no new per-owner concurrency. See backend/docs/batch-pipeline-serving.md."} - - {"key":"BATCH_PIPELINE_SERVING_PROJECTION_SWEEP","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; only off skips Serving.reconcile projection sweep, preserving owner wakes, held drains, metrics and leader-leased maintenance","fail":"open","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} - - {"key":"BATCH_PIPELINE_PROJECTION_PUMP_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} - - {"key":"BATCH_PIPELINE_PUMP_SWEEP_PAGES","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 1; projection pump sweep pages per tick, clamped to 1-10; malformed keeps 1","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} - - {"key":"BATCH_PIPELINE_PUMP_TICK_SECONDS","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 20 seconds; projection pump tick sleep, clamped to 1-60; malformed keeps 20","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} - - {"key":"BATCH_PIPELINE_MAX_OWNER_DRAINS","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 8; concurrent owner drains per process, clamped to 1-32 at import; malformed keeps 8","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Ramp tonight: pump service N=3-4; owner_drain rows/s at least 70% of pre-cutover, claim_busy ratio no higher than pre-cutover; lead2 expands. Verify pump before disabling serving sweeps. Rollback: pump false, serving sweep on, owner drains 8."} + - {"key":"BATCH_PIPELINE_SERVING_PROJECTION_SWEEP","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; only off skips Serving.reconcile projection sweep, preserving owner wakes, held drains, metrics and leader-leased maintenance","fail":"open","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Safe start: dedicated service only, N=4 instances, pages 1, tick 2 s, MAX_OWNER_DRAINS 8. Never enable the pump on backend-integration; the selector is service-wide. Lead2 owns rollout after the deployment-shape decision. Keep serving sweeps ON until owner_drain advanced >=20 rows/s for 30 min and a dedicated-service missing/stale omi_batch_projection_pump_last_tick_seconds alert is installed and verified. Alarm on Firestore read rate above +50% of pre-cutover; owner_drain rows/s must remain at least 70% of pre-cutover. Watch sweep elapsed time and executor saturation. Evaluate claim_busy share no higher than pre-cutover only after serving sweeps are switched off, not during dual-running. Rollback via env-only new revisions: serving sweeps ON on existing service first, pump false on dedicated service, owner drains 8."} + - {"key":"BATCH_PIPELINE_PROJECTION_PUMP_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Safe start: dedicated service only, N=4 instances, pages 1, tick 2 s, MAX_OWNER_DRAINS 8. Never enable the pump on backend-integration; the selector is service-wide. Lead2 owns rollout after the deployment-shape decision. Keep serving sweeps ON until owner_drain advanced >=20 rows/s for 30 min and a dedicated-service missing/stale omi_batch_projection_pump_last_tick_seconds alert is installed and verified. Alarm on Firestore read rate above +50% of pre-cutover; owner_drain rows/s must remain at least 70% of pre-cutover. Watch sweep elapsed time and executor saturation. Evaluate claim_busy share no higher than pre-cutover only after serving sweeps are switched off, not during dual-running. Rollback via env-only new revisions: serving sweeps ON on existing service first, pump false on dedicated service, owner drains 8."} + - {"key":"BATCH_PIPELINE_PUMP_SWEEP_PAGES","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 1; projection pump sweep pages per tick, clamped to 1-4; malformed keeps 1","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Safe start: dedicated service only, N=4 instances, pages 1, tick 2 s, MAX_OWNER_DRAINS 8. Never enable the pump on backend-integration; the selector is service-wide. Lead2 owns rollout after the deployment-shape decision. Keep serving sweeps ON until owner_drain advanced >=20 rows/s for 30 min and a dedicated-service missing/stale omi_batch_projection_pump_last_tick_seconds alert is installed and verified. Alarm on Firestore read rate above +50% of pre-cutover; owner_drain rows/s must remain at least 70% of pre-cutover. Watch sweep elapsed time and executor saturation. Evaluate claim_busy share no higher than pre-cutover only after serving sweeps are switched off, not during dual-running. Rollback via env-only new revisions: serving sweeps ON on existing service first, pump false on dedicated service, owner drains 8."} + - {"key":"BATCH_PIPELINE_PUMP_TICK_SECONDS","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 20 seconds; projection pump tick sleep, clamped to 2-60; malformed keeps 20","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Safe start: dedicated service only, N=4 instances, pages 1, tick 2 s, MAX_OWNER_DRAINS 8. Never enable the pump on backend-integration; the selector is service-wide. Lead2 owns rollout after the deployment-shape decision. Keep serving sweeps ON until owner_drain advanced >=20 rows/s for 30 min and a dedicated-service missing/stale omi_batch_projection_pump_last_tick_seconds alert is installed and verified. Alarm on Firestore read rate above +50% of pre-cutover; owner_drain rows/s must remain at least 70% of pre-cutover. Watch sweep elapsed time and executor saturation. Evaluate claim_busy share no higher than pre-cutover only after serving sweeps are switched off, not during dual-running. Rollback via env-only new revisions: serving sweeps ON on existing service first, pump false on dedicated service, owner drains 8."} + - {"key":"BATCH_PIPELINE_MAX_OWNER_DRAINS","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default 8; concurrent owner drains per process, clamped to 1-16 at import; malformed keeps 8","fail":"closed","owner":"lead2","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Code only; deployment shape pending David. Safe start: dedicated service only, N=4 instances, pages 1, tick 2 s, MAX_OWNER_DRAINS 8. Never enable the pump on backend-integration; the selector is service-wide. Lead2 owns rollout after the deployment-shape decision. Keep serving sweeps ON until owner_drain advanced >=20 rows/s for 30 min and a dedicated-service missing/stale omi_batch_projection_pump_last_tick_seconds alert is installed and verified. Alarm on Firestore read rate above +50% of pre-cutover; owner_drain rows/s must remain at least 70% of pre-cutover. Watch sweep elapsed time and executor saturation. Evaluate claim_busy share no higher than pre-cutover only after serving sweeps are switched off, not during dual-running. Rollback via env-only new revisions: serving sweeps ON on existing service first, pump false on dedicated service, owner drains 8."} - {"key":"BATCH_PIPELINE_OWNER_DRAIN_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; a per-process owner drain walks each discovered owner's result log continuously (up to 8 owners at once, 120 s turns, FIFO re-queue) instead of the sweep-shared 30 s walk budget","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","notes":"False restores the sweep owner-chain walk with every pre-check; read-ahead (BATCH_PIPELINE_PROJECTION_PREFETCH=0) and the deferred index (BATCH_PIPELINE_PROJECTION_INDEX_DEFERRED_ENABLED=false) have their own switches, and all three off is the pre-drain behaviour. On: the sweep only discovers owners; each walk holds the advisory owner claim for its 15 s step and skips the predecessor/per-part pre-check transactions for rows it reached as the watermark's successor. The assignment transaction's predecessor, privacy and receipt fences stay authoritative; one owner never runs two chains. See backend/docs/batch-pipeline-serving.md."} - {"key":"BATCH_PIPELINE_HELD_TERMINAL_STATUS_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; terminally held batch uploads retain phone custody as processing/batch_unit_held with a five-minute poll hint","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","ramp":{"cohorts":["all batch uploads at deploy (default on)"],"clean_metric":"Zero completed status responses for uploads with unresolved held units; projected siblings remain visible and held polls are paced at five minutes","expander":"dazheng (two-lane lead)","done_by":"2026-10-27"},"notes":"Read-only status projection; no ledger, job or retention write. Phones keep their independent WAL copy until actual completion; server artifact retention alone never permits acknowledgement. Open operator intents or in-flight siblings also remain processing. False restores failed/batch_worker_quarantined and still never acknowledges custody. See backend/docs/batch-pipeline-serving.md."} - {"key":"BATCH_PIPELINE_RECOVERY_POLICY_REFRESH_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; a quarantined held unit whose recorded recovery policy is disabled and which has no recovery intent (no paid call authorized) re-resolves the recovery setting at sweep or dispatch, and adopts it only when it is now enabled","fail":"open","owner":"dazheng","created":"2026-10-20","terminal":"keep","decision":"keep","ramp":{"cohorts":["all held units at deploy (default on)"],"clean_metric":"recovery sweep outcome=disabled count falls to zero and paid recovery starts stay within the per-owner allowance","expander":"dazheng (two-lane lead)","done_by":"2026-10-27"},"notes":"Spends at most the one at-most-once paid recovery the unit would have had if admitted while recovery was on; the existing allowance/cap gates and RecoveryStore.start fence still apply. An enabled snapshot is never re-resolved or downgraded. False restores the frozen snapshot (outcome=disabled forever). See backend/docs/batch-pipeline-serving.md."} From a3189eed1b225af1c27bec0c90fbf947e84bce9e Mon Sep 17 00:00:00 2001 From: David Zhang Date: Fri, 6 Nov 2026 01:19:26 +0700 Subject: [PATCH 5/5] fix(batch): keep pump metrics scrapes out of reconciliation --- backend/routers/batch_pipeline.py | 4 +-- .../tests/unit/test_batch_pipeline_serving.py | 25 +++++++++++++++++++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/backend/routers/batch_pipeline.py b/backend/routers/batch_pipeline.py index 31574359a9c..a7d8b5730c9 100644 --- a/backend/routers/batch_pipeline.py +++ b/backend/routers/batch_pipeline.py @@ -11,7 +11,7 @@ from prometheus_client import CONTENT_TYPE_LATEST, Counter, generate_latest from starlette.responses import Response, JSONResponse -from config.batch_pipeline import read_only_metrics_enabled +from config.batch_pipeline import projection_pump_enabled, read_only_metrics_enabled from config.summary_guarantee import enabled as summary_deadline_enabled from database.batch_pipeline_privacy import BatchPrivacyBlocked from utils.batch_pipeline.contracts import PoolCapacity, WorkResult @@ -263,7 +263,7 @@ async def reconcile(body: WorkerRequest): @router.get('/metrics', include_in_schema=False) async def metrics(): service = serving_for_stage(current_stage()) - if read_only_metrics_enabled() and service.metrics is not None: + if (projection_pump_enabled() or read_only_metrics_enabled()) and service.metrics is not None: await run_blocking(db_executor, service.metrics.refresh, service.store, time.time()) else: # Periodic maintenance uses the leader lease; per-unit policy decides follow-ups. diff --git a/backend/tests/unit/test_batch_pipeline_serving.py b/backend/tests/unit/test_batch_pipeline_serving.py index c36a3511789..f967a879e74 100644 --- a/backend/tests/unit/test_batch_pipeline_serving.py +++ b/backend/tests/unit/test_batch_pipeline_serving.py @@ -645,6 +645,31 @@ def test_metrics_retains_legacy_drain_until_both_controls_are_on(api, monkeypatc service.metrics.refresh.assert_not_called() +@pytest.mark.parametrize( + 'scrape_flag,serving_flag', [(None, 'false'), ('true', 'false'), (None, 'true'), ('true', 'true')] +) +def test_pump_metrics_refreshes_without_reconcile_regardless_of_serving_controls( + api, monkeypatch, scrape_flag, serving_flag +): + from unittest.mock import AsyncMock + + client, store = api + monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED', 'true') + monkeypatch.setenv('BATCH_PIPELINE_SERVING_ENABLED', serving_flag) + monkeypatch.delenv('BATCH_PIPELINE_READ_ONLY_METRICS_ENABLED', raising=False) + if scrape_flag is not None: + monkeypatch.setenv('BATCH_PIPELINE_READ_ONLY_METRICS_ENABLED', scrape_flag) + service = Serving(store, metrics=Mock(), leader=Mock()) + service.reconcile_async = AsyncMock(side_effect=AssertionError('pump metrics must never reconcile')) + monkeypatch.setattr(module, 'serving_for_stage', lambda stage: service) + + assert client.get('/v2/batch-pipeline/metrics').status_code == 200 + service.metrics.refresh.assert_called_once() + assert service.metrics.refresh.call_args.args[0] is store + service.reconcile_async.assert_not_awaited() + service.leader.acquire.assert_not_called() + + @pytest.mark.asyncio async def test_recovery_request_awaits_paid_intent_before_response_and_projection(api, monkeypatch, caplog): from routers import batch_pipeline as routes