From 8ae0cc93eb0f5d69810c762584de0c2c2ef7a8d1 Mon Sep 17 00:00:00 2001 From: David Zhang Date: Fri, 6 Nov 2026 00:10:08 +0700 Subject: [PATCH 1/5] fix(pipeline): age-gate ancient sweep re-wakes --- backend/config/flag_budget.json | 5 +- backend/config/policy/sweep_age.py | 29 ++++ backend/database/outbox_provider_stt.py | 3 + backend/docs/feature-flag-registry.md | 6 + .../tests/unit/test_provider_stt_outbox.py | 25 ++++ backend/tests/unit/test_sweep_age_gate.py | 139 ++++++++++++++++++ backend/utils/enrichment/outbox.py | 20 +++ backend/utils/observability/sweep_age.py | 35 +++++ config/feature-flags.yaml | 3 + 9 files changed, 264 insertions(+), 1 deletion(-) create mode 100644 backend/config/policy/sweep_age.py create mode 100644 backend/tests/unit/test_sweep_age_gate.py create mode 100644 backend/utils/observability/sweep_age.py diff --git a/backend/config/flag_budget.json b/backend/config/flag_budget.json index 5a7b0686a0b..05d4462dc7a 100644 --- a/backend/config/flag_budget.json +++ b/backend/config/flag_budget.json @@ -24,8 +24,11 @@ }, "seed_review_by": "2026-11-15", "budget": 169, - "count": 248, + "count": 251, "names": { + "BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS": {"owner": "omi-pipeline", "review_by": "2026-11-30"}, + "BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS": {"owner": "omi-pipeline", "review_by": "2026-11-30"}, + "BATCH_PIPELINE_SWEEP_AGE_GATE": {"owner": "omi-pipeline", "review_by": "2026-11-30"}, "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/sweep_age.py b/backend/config/policy/sweep_age.py new file mode 100644 index 00000000000..58968f8e3f5 --- /dev/null +++ b/backend/config/policy/sweep_age.py @@ -0,0 +1,29 @@ +"""Live sweep admission limits; never alter durable obligations or initial enqueue.""" + +import math +import os + +DEFAULT_MAX_AGE_DAYS = 14.0 + + +def enabled() -> bool: + """Default on; only explicit off restores the previous sweep behavior.""" + return os.getenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on').strip().lower() != 'off' + + +def _days(raw: str) -> float: + try: + days = float(raw) + except ValueError: + return DEFAULT_MAX_AGE_DAYS + return days if math.isfinite(days) and 0 < days <= 3650 else DEFAULT_MAX_AGE_DAYS + + +def max_age_seconds(queue: str) -> float: + if queue == 'provider-stt': + days = _days(os.getenv('BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS', '14')) + elif queue == 'enrichment-fresh': + days = _days(os.getenv('BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS', '14')) + else: + raise ValueError('unknown sweep age queue') + return days * 86400 diff --git a/backend/database/outbox_provider_stt.py b/backend/database/outbox_provider_stt.py index 561b351e173..6173d91da50 100644 --- a/backend/database/outbox_provider_stt.py +++ b/backend/database/outbox_provider_stt.py @@ -19,6 +19,7 @@ from typing import Any, Callable, Literal, Mapping from database.outbox import Claim, EnqueueResult, OutboxKind, Outcome, SweepCounts +from utils.observability.sweep_age import skip_ancient Row = dict[str, Any] _ROW_ID = re.compile('[0-9a-f]{64}') @@ -310,6 +311,8 @@ def sweep(self, *, now: float, limit: int = 100) -> SweepCounts: for snapshot in query.stream(): row = snapshot.to_dict() or {} try: + if skip_ancient(row, 'provider-stt', now=now): + continue if self._blocked(snapshot.id, row['owner_id']): self._mutate(snapshot.id, lambda old: (self._close(old, 'privacy_blocked') if old else old, None)) discarded += 1 diff --git a/backend/docs/feature-flag-registry.md b/backend/docs/feature-flag-registry.md index 92f2b2e8f40..db99ce45201 100644 --- a/backend/docs/feature-flag-registry.md +++ b/backend/docs/feature-flag-registry.md @@ -553,6 +553,7 @@ and an explicit empty literal renders as `''`. | `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_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_AGE_GATE` | Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes | backend | env | closed | — | — | — | — | keep | keep | 2026-11-30 | omi-pipeline | | `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 | | `BATCH_PIPELINE_WORKER_ENABLED` | Default-off dedicated GPU process switch; drains admitted work without enabling upload admission | backend | env | closed | false | false (backend-listen (chart), cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | false (backend-listen (chart), cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | — | keep | keep | — | dazheng | @@ -785,11 +786,13 @@ and an explicit empty literal renders as `''`. | `BATCH_PIPELINE_ASR_LEFT_CONTEXT_UNITS` | Process-local predecessor tails kept for ASR left context (default 64, range [0, 1024]; each <=8 s int16, 10 min TTL, keyed by owner, privacy generation and capture stream: source, device and lock from the staged intake context, or the record conversation) | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_CONTEXT_SECONDS` | GPU chain internal ASR boundary context seconds; default 2; requires AMI qualification | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_DIARIZER` | GPU chain diarizer registry selection; default pyannote; unavailable candidates fail closed | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | +| `BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS` | Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 | backend | env | closed | — | — | — | — | keep | keep | 2026-11-30 | omi-pipeline | | `BATCH_PIPELINE_FLIGHT_BATCH_SECONDS` | Audio seconds packed per pyannote flight group; default 1800, range (0, 7200]; longer units take the serial path | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_INDEPENDENT_RECOVERY_MAX_SECONDS` | Default 60 seconds; lower the independent-language recovery per-unit duration cap, never exceed the hard 60-second ceiling | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_MODEL_BATCH_SIZE` | GPU chain maximum model window batch size; default 8; hard ceiling 32 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_PADDED_SECONDS` | GPU chain padded waveform seconds per model batch; default 272 (8 windows with context); shared-live hard cap 120 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_POOL_KIND` | Dedicated batch GPU by default; shared_live requires bounded units and the existing live GPU arbiter | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | +| `BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS` | Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 | backend | env | closed | — | — | — | — | keep | keep | 2026-11-30 | omi-pipeline | | `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 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_REUSE_LIVE_ENABLED` | Default-off GPU-chain pre-STT reuse; requires existing lineage gates and full-span words-of-record quality proof | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `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 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | @@ -1074,6 +1077,7 @@ their code default (`fail` tells you which way a missing value resolves). - `BATCH_PIPELINE_DIARIZER` — GPU chain diarizer registry selection; default pyannote; unavailable candidates fail closed (fail: closed) - `BATCH_PIPELINE_DIARIZE_SPEECH_ONLY_ENABLED` — Default on unless false; batch GPU worker diarizes only padded VAD and word evidence and maps turns back to capture clocks (fail: open) - `BATCH_PIPELINE_EMPTY_ASR_RECOVERY_ENABLED` — Default off; hand speech-bearing empty ASR after padded retry to fenced per-segment recovery (fail: closed) +- `BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS` — Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 (fail: closed) - `BATCH_PIPELINE_FLIGHT_BATCHING_ENABLED` — Default off; pack pyannote segmentation (shared by VAD and diarization) and per-chunk embedding trunks across one GPU flight (fail: closed) - `BATCH_PIPELINE_FLIGHT_BATCH_SECONDS` — Audio seconds packed per pyannote flight group; default 1800, range (0, 7200]; longer units take the serial path (fail: closed) - `BATCH_PIPELINE_FRESH_GPU_PRIORITY_ENABLED` — Default on unless false; batch uploads in the 6 h fresh sync lane whose oldest capture is over 10 min old lease GPU at RECENT priority, after fresh uploads (fail: open) @@ -1102,6 +1106,7 @@ their code default (`fail` tells you which way a missing value resolves). - `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_SOURCE_DELETED_ENABLED` — Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication (fail: open) +- `BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS` — Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 (fail: closed) - `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_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) @@ -1120,6 +1125,7 @@ their code default (`fail` tells you which way a missing value resolves). - `BATCH_PIPELINE_SHORT_SPEECH_SINGLE_ENABLED` — Default off; a batch unit whose speech and words form one region of at most 2 s (pauses up to 0.5 s inclusive) takes the existing skipped_single result without diarization (fail: closed) - `BATCH_PIPELINE_STAGE_PROFILING_ENABLED` — Dedicated worker omi_batch_pipeline_stage_seconds wall-clock histograms; default on only for OMI_ENV_STAGE=dev, prod/unset off, explicit true/false overrides (local/offline opt in) (fail: open) - `BATCH_PIPELINE_STRICT_VAD_SHADOW_PERCENT` — Share of units that run the CPU strict-VAD shadow, sampled deterministically by segment hash; default 10, 0 disables the shadow, 100 restores full sampling; invalid values keep 10 (fail: closed) +- `BATCH_PIPELINE_SWEEP_AGE_GATE` — Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes (fail: closed) - `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 (fail: open) - `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 (fail: open) - `BATCH_PIPELINE_WINDOW_SECONDS` — GPU chain internal ASR core window seconds; default 30; queue units remain conversations (fail: closed) diff --git a/backend/tests/unit/test_provider_stt_outbox.py b/backend/tests/unit/test_provider_stt_outbox.py index 359f198ed15..89e9a576079 100644 --- a/backend/tests/unit/test_provider_stt_outbox.py +++ b/backend/tests/unit/test_provider_stt_outbox.py @@ -220,3 +220,28 @@ def test_wake_task_shape(monkeypatch): monkeypatch.delenv('PROVIDER_STT_TASKS_QUEUE') with pytest.raises(RuntimeError): provider_stt_tasks.create_wake(ROW, 'a' * 32, due_at=0) + + +@pytest.mark.parametrize( + 'age,setting,skipped', [(15, 'on', True), (1, 'on', False), (14, 'on', False), (15, 'off', False)] +) +def test_sweep_age_gate_preserves_old_rows_and_counts(box, monkeypatch, age, setting, skipped): + from copy import deepcopy + from utils.observability.sweep_age import SKIPPED + + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', setting) + monkeypatch.delenv('BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS', raising=False) + enqueue(box, now=100) + box.woken.clear() + before = deepcopy(row(box)) + versions = dict(box.client.versions) + counter = SKIPPED.labels(queue='provider-stt', reason='max_age') + count = counter._value.get() + result = box.outbox.sweep(now=100 + age * 86400) + assert result.rewoken == (0 if skipped else 1) + assert result.errors == 0 + assert len(box.woken) == (0 if skipped else 1) + assert counter._value.get() - count == int(skipped) + if skipped: + assert row(box) == before + assert box.client.versions == versions diff --git a/backend/tests/unit/test_sweep_age_gate.py b/backend/tests/unit/test_sweep_age_gate.py new file mode 100644 index 00000000000..7d01a6aa8ee --- /dev/null +++ b/backend/tests/unit/test_sweep_age_gate.py @@ -0,0 +1,139 @@ +"""Non-mutating re-wake admission: queue scope, age boundary and rollback.""" + +from copy import deepcopy +from types import SimpleNamespace + +import pytest + +from config.policy import sweep_age +from tests.unit.test_sync_reprocess_coalesce import state, sweep_query # noqa: F401 +from utils.enrichment import outbox +from utils.observability.sweep_age import SKIPPED, skip_ancient + + +@pytest.mark.parametrize('queue', ['provider-stt', 'enrichment-fresh']) +@pytest.mark.parametrize('raw', ['', 'invalid', 'nan', 'inf', '-1', '0', '3651']) +def test_malformed_age_config_uses_default(monkeypatch, queue, raw): + name = 'BATCH_PIPELINE_' + queue.upper().replace('-', '_') + '_SWEEP_MAX_AGE_DAYS' + monkeypatch.setenv(name, raw) + assert sweep_age.max_age_seconds(queue) == 14 * 86400 + + +@pytest.mark.parametrize('raw', ['', 'invalid', 'false', '0', 'on']) +def test_only_off_disables_gate(monkeypatch, raw): + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', raw) + assert sweep_age.enabled() + + +@pytest.mark.parametrize('value', [None, 'bad', True, float('nan'), float('inf')]) +def test_unknown_age_keeps_wake(value): + assert not skip_ancient({'created_at': value}, 'provider-stt', now=2_000_000) + + +@pytest.mark.parametrize( + 'age,setting,skipped', [(15, 'on', True), (1, 'on', False), (14, 'on', False), (15, 'off', False)] +) +@pytest.mark.parametrize('injected', [False, True]) +def test_enrichment_sweep_skips_without_mutation(state, sweep_query, monkeypatch, age, setting, skipped, injected): + rows, tasks = state + now = 2_000_000.0 + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', setting) + monkeypatch.delenv('BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS', raising=False) + monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') + outbox.mark('u', 'c', 'en', now=now - age * 86400) + row_id = outbox.key('u', 'c') + rows[row_id]['dispatch']['queue'] = 'enrichment-fresh' + original = deepcopy(rows) + tasks.clear() + sweep_query.stream.side_effect = lambda: [ + SimpleNamespace(id=key, to_dict=lambda key=key: deepcopy(rows[key])) for key in rows + ] + counter = SKIPPED.labels(queue='enrichment-fresh', reason='max_age') + before = counter._value.get() + kwargs = {'wake': lambda row: tasks.append(deepcopy(row))} if injected else {} + result = outbox.sweep(now=now, **kwargs) + assert result == {'repaired': 0 if skipped else 1, 'error': 0} + assert len(tasks) == (0 if skipped else 1) + assert counter._value.get() - before == int(skipped) + if skipped: + assert rows == original + + +def test_ancient_fresh_sweep_precedes_existing_parking(state, sweep_query, monkeypatch): + rows, tasks = state + monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') + outbox.mark('u', 'c', 'en', now=100) + key = outbox.key('u', 'c') + rows[key]['dispatch']['queue'] = 'enrichment-fresh' + original = deepcopy(rows) + tasks.clear() + monkeypatch.setenv('ENRICHMENT_MAX_AGE_HOURS', '72') + sweep_query.stream.side_effect = lambda: [SimpleNamespace(id=key, to_dict=lambda: deepcopy(rows[key]))] + assert outbox.sweep(now=2_000_000) == {'repaired': 0, 'error': 0} + assert rows == original and not tasks + + +def test_other_enrichment_queues_keep_old_behavior(state, sweep_query, monkeypatch): + rows, tasks = state + monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') + outbox.mark('u', 'c', 'en', now=100) + tasks.clear() + assert outbox.sweep(now=2_000_000) == {'repaired': 1, 'error': 0} + assert len(tasks) == 1 + + +@pytest.mark.parametrize('config', ['cloud_run_gmp_sidecar.yaml', 'cloud_run_gmp_sync.yaml']) +def test_skip_counter_passes_existing_gmp_keep_lists(config): + import re + from pathlib import Path + import yaml + + path = Path(__file__).resolve().parents[2] / 'deploy' / config + monitoring = yaml.safe_load(path.read_text()) + rules = monitoring['spec']['endpoints'][0]['metricRelabeling'] + metric = 'omi_batch_pipeline_sweep_skipped_total' + for rule in rules: + if rule.get('sourceLabels') == ['__name__']: + matched = re.fullmatch(rule['regex'], metric) is not None + if rule['action'] == 'keep': + assert matched + elif rule['action'] == 'drop': + assert not matched + + +def test_per_queue_limits_are_independent(monkeypatch): + monkeypatch.setenv('BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS', '20') + monkeypatch.setenv('BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS', '7') + assert sweep_age.max_age_seconds('provider-stt') == 20 * 86400 + assert sweep_age.max_age_seconds('enrichment-fresh') == 7 * 86400 + + +def test_enrichment_cas_rechecks_row_age(state, sweep_query, monkeypatch): + rows, tasks = state + monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') + outbox.mark('u', 'c', 'en', now=100) + key = outbox.key('u', 'c') + rows[key]['dispatch']['queue'] = 'enrichment-fresh' + original = deepcopy(rows) + # The query snapshot is not authoritative: repair must check the CAS row too. + sweep_query.stream.side_effect = lambda: [SimpleNamespace(id=key, to_dict=lambda: {})] + tasks.clear() + assert outbox.sweep(now=2_000_000, wake=lambda row: tasks.append(row)) == {'repaired': 0, 'error': 0} + assert rows == original and not tasks + + +def test_off_restores_existing_parking_policy(state, sweep_query, monkeypatch): + rows, tasks = state + monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'off') + outbox.mark('u', 'c', 'en', now=100) + key = outbox.key('u', 'c') + rows[key]['dispatch']['queue'] = 'enrichment-fresh' + tasks.clear() + monkeypatch.setenv('ENRICHMENT_MAX_AGE_HOURS', '72') + sweep_query.stream.side_effect = lambda: [SimpleNamespace(id=key, to_dict=lambda: deepcopy(rows[key]))] + assert outbox.sweep(now=2_000_000) == {'repaired': 0, 'error': 0} + assert rows[key]['age_gate_parked'] and not tasks diff --git a/backend/utils/enrichment/outbox.py b/backend/utils/enrichment/outbox.py index bd8c0435c83..c035763199b 100644 --- a/backend/utils/enrichment/outbox.py +++ b/backend/utils/enrichment/outbox.py @@ -7,6 +7,7 @@ from __future__ import annotations from utils.observability.delivery import record_delivery +from utils.observability.sweep_age import skip_ancient from utils.observability.fallback import record_fallback import hashlib @@ -848,6 +849,17 @@ def finish(old: Row | None) -> tuple[Row | None, Row | None]: return 'superseded' if superseded else 'processed' +def _skip_ancient_fresh(row: Row, *, now: float) -> bool: + queue = enrichment_config.fresh_queue() + dispatch = row.get('dispatch') + return bool( + queue + and isinstance(dispatch, dict) + and dispatch.get('queue') == queue + and skip_ancient(row, 'enrichment-fresh', now=now) + ) + + def sweep( *, now: float | None = None, @@ -922,6 +934,8 @@ def repair(old: Row | None) -> tuple[Row | None, tuple[str, Row | None]]: nonlocal recent if old is None or old.get('dormant') or old.get('lease_until', 0) > current: return old, ('skipped', None) + if _skip_ancient_fresh(old, now=current): + return old, ('skipped', None) row = dict(old) promoted = False # A valid old wake can wait hours behind historical enrichment. @@ -985,6 +999,8 @@ def repair(old: Row | None) -> tuple[Row | None, tuple[str, Row | None]]: # Claim and fence the repaired binding in the same CAS. Any old # task delivery becomes stale; valid backfill bindings stay put. row.update(dispatch=binding, task_token=uuid.uuid4().hex, task_due_at=current) + if _skip_ancient_fresh(row, now=current): + return old, ('skipped', None) if expired: # Fence the old holder (invariant 1) and retire its delivery: its task already ran. row.update(owner=None, lease_until=0, task_token=uuid.uuid4().hex, task_due_at=current) @@ -1007,6 +1023,10 @@ def update(old: Row | None) -> tuple[Row | None, None]: return update try: + # Skip before the independent historical-parking policy can mutate this row. + candidate = getattr(snapshot, 'to_dict', lambda: {})() or {} + if _skip_ancient_fresh(candidate, now=current): + continue if not injected: from utils.enrichment.age_gate import park_if_expired diff --git a/backend/utils/observability/sweep_age.py b/backend/utils/observability/sweep_age.py new file mode 100644 index 00000000000..c0475d232b1 --- /dev/null +++ b/backend/utils/observability/sweep_age.py @@ -0,0 +1,35 @@ +"""Count ancient sweep candidates without changing their durable rows.""" + +from datetime import datetime, timezone +import math + +from prometheus_client import Counter + +from config.policy import sweep_age + +SKIPPED = Counter( + 'omi_batch_pipeline_sweep_skipped_total', + 'Durable rows left unchanged instead of re-enqueued by maintenance sweeps', + ['queue', 'reason'], +) + + +def skip_ancient(row, queue: str, *, now: float) -> bool: + if not sweep_age.enabled(): + return False + field = 'created_at' if queue == 'provider-stt' else 'first_pending_at' + value = row.get(field) + try: + if isinstance(value, str): + value = datetime.fromisoformat(value.replace('Z', '+00:00')) + if isinstance(value, datetime): + value = (value if value.tzinfo else value.replace(tzinfo=timezone.utc)).timestamp() + # Unknown metadata cannot prove that an obligation is ancient. + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + return False + expired = now - value > sweep_age.max_age_seconds(queue) + except (ValueError, OverflowError): + return False + if expired: + SKIPPED.labels(queue=queue, reason='max_age').inc() + return expired diff --git a/config/feature-flags.yaml b/config/feature-flags.yaml index 1705fb624d2..9e9a327a4c1 100644 --- a/config/feature-flags.yaml +++ b/config/feature-flags.yaml @@ -1,4 +1,7 @@ flags: + - {"key":"BATCH_PIPELINE_SWEEP_AGE_GATE","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","review_by":"2026-11-30","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} + - {"key":"BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","review_by":"2026-11-30","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} + - {"key":"BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","review_by":"2026-11-30","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} - {"key":"PUSHER_MAX_ABANDON_RECORD_DRAINS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Per-pod abandon coordinators; default and prod4, clamped0..16","fail":"closed","owner":"two-lane lead","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Zero refuses all overflow coordination. Full admission logs abandon_overflow and retains the old producer fence for process/pod-death repair. Coordinators wait for actual shared storage_executor futures to settle before attesting stoppage."} - {"key":"PUSHER_ABANDON_EVICTED_TOKENS","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"On unless off; capacity-evicted pusher drains attest token abandonment only after worker and executor settlement","fail":"open","owner":"two-lane lead","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"lead2 88ba4d. Exact-token abandoned_at never certifies a projected tail; repair requires archive and finish/tail proof. Metadata admission retries with bounded backoff; reconnect tokens stay live. Pusher train: one canary pod at 64, watch memory p99, capacity, abandoned and upload_pending before expansion. Rollback off plus cap32. See backend/docs/listen_pusher_pipeline/changes/2026-11-05-pusher-abandon-evicted-tokens.md."} - {"key":"PUSHER_MAX_CLOSED_RECORD_DRAINS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Per-process retained closed-record drain count; default32, clamped8..128, prod64","fail":"closed","owner":"two-lane lead","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"lead2 88ba4d rejects prod128 until a retained-PCM byte budget exists. Invalid values use32. Startup selection logged once at first socket. Pusher train canary and per-pod memory p99 review required; rollback32."} From 880b8c200a62948f433dc7231ddc69f40ecdea8f Mon Sep 17 00:00:00 2001 From: David Zhang Date: Fri, 6 Nov 2026 00:12:44 +0700 Subject: [PATCH 2/5] fix(pipeline): preserve privacy fence before sweep age check --- backend/database/outbox_provider_stt.py | 4 ++-- backend/tests/unit/test_provider_stt_outbox.py | 15 +++++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/backend/database/outbox_provider_stt.py b/backend/database/outbox_provider_stt.py index 6173d91da50..3a298351491 100644 --- a/backend/database/outbox_provider_stt.py +++ b/backend/database/outbox_provider_stt.py @@ -311,8 +311,6 @@ def sweep(self, *, now: float, limit: int = 100) -> SweepCounts: for snapshot in query.stream(): row = snapshot.to_dict() or {} try: - if skip_ancient(row, 'provider-stt', now=now): - continue if self._blocked(snapshot.id, row['owner_id']): self._mutate(snapshot.id, lambda old: (self._close(old, 'privacy_blocked') if old else old, None)) discarded += 1 @@ -323,6 +321,8 @@ def sweep(self, *, now: float, limit: int = 100) -> SweepCounts: or now - float(row.get('woken_at') or 0) < REWAKE_AFTER_SECONDS ): continue + if skip_ancient(row, 'provider-stt', now=now): + continue if row.get('lease_token'): expired += 1 diff --git a/backend/tests/unit/test_provider_stt_outbox.py b/backend/tests/unit/test_provider_stt_outbox.py index 89e9a576079..cb9cb2dc006 100644 --- a/backend/tests/unit/test_provider_stt_outbox.py +++ b/backend/tests/unit/test_provider_stt_outbox.py @@ -245,3 +245,18 @@ def test_sweep_age_gate_preserves_old_rows_and_counts(box, monkeypatch, age, set if skipped: assert row(box) == before assert box.client.versions == versions + + +def test_ancient_sweep_keeps_existing_privacy_discard(box, monkeypatch): + from utils.observability.sweep_age import SKIPPED + + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') + enqueue(box, now=100) + box.blocked.add(ROW) + box.woken.clear() + counter = SKIPPED.labels(queue='provider-stt', reason='max_age') + before = counter._value.get() + counts = box.outbox.sweep(now=100 + 15 * 86400) + assert counts.discarded == 1 and counts.rewoken == 0 + assert row(box)['closed_reason'] == 'privacy_blocked' + assert not box.woken and counter._value.get() == before From ece6d2cb9dff2d154cc895d702c50da1aa0df2d1 Mon Sep 17 00:00:00 2001 From: David Zhang Date: Fri, 6 Nov 2026 00:17:09 +0700 Subject: [PATCH 3/5] fix(pipeline): align age gate registry lifecycle metadata --- backend/docs/feature-flag-registry.md | 6 +++--- config/feature-flags.yaml | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/backend/docs/feature-flag-registry.md b/backend/docs/feature-flag-registry.md index db99ce45201..83a272237c1 100644 --- a/backend/docs/feature-flag-registry.md +++ b/backend/docs/feature-flag-registry.md @@ -553,7 +553,7 @@ and an explicit empty literal renders as `''`. | `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_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_AGE_GATE` | Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes | backend | env | closed | — | — | — | — | keep | keep | 2026-11-30 | omi-pipeline | +| `BATCH_PIPELINE_SWEEP_AGE_GATE` | Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes | backend | env | closed | — | — | — | — | keep | keep | — | omi-pipeline | | `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 | | `BATCH_PIPELINE_WORKER_ENABLED` | Default-off dedicated GPU process switch; drains admitted work without enabling upload admission | backend | env | closed | false | false (backend-listen (chart), cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | false (backend-listen (chart), cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | — | keep | keep | — | dazheng | @@ -786,13 +786,13 @@ and an explicit empty literal renders as `''`. | `BATCH_PIPELINE_ASR_LEFT_CONTEXT_UNITS` | Process-local predecessor tails kept for ASR left context (default 64, range [0, 1024]; each <=8 s int16, 10 min TTL, keyed by owner, privacy generation and capture stream: source, device and lock from the staged intake context, or the record conversation) | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_CONTEXT_SECONDS` | GPU chain internal ASR boundary context seconds; default 2; requires AMI qualification | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_DIARIZER` | GPU chain diarizer registry selection; default pyannote; unavailable candidates fail closed | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | -| `BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS` | Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 | backend | env | closed | — | — | — | — | keep | keep | 2026-11-30 | omi-pipeline | +| `BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS` | Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 | backend | env | closed | — | — | — | — | keep | keep | — | omi-pipeline | | `BATCH_PIPELINE_FLIGHT_BATCH_SECONDS` | Audio seconds packed per pyannote flight group; default 1800, range (0, 7200]; longer units take the serial path | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_INDEPENDENT_RECOVERY_MAX_SECONDS` | Default 60 seconds; lower the independent-language recovery per-unit duration cap, never exceed the hard 60-second ceiling | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_MODEL_BATCH_SIZE` | GPU chain maximum model window batch size; default 8; hard ceiling 32 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_PADDED_SECONDS` | GPU chain padded waveform seconds per model batch; default 272 (8 windows with context); shared-live hard cap 120 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_POOL_KIND` | Dedicated batch GPU by default; shared_live requires bounded units and the existing live GPU arbiter | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | -| `BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS` | Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 | backend | env | closed | — | — | — | — | keep | keep | 2026-11-30 | omi-pipeline | +| `BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS` | Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 | backend | env | closed | — | — | — | — | keep | keep | — | omi-pipeline | | `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 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_REUSE_LIVE_ENABLED` | Default-off GPU-chain pre-STT reuse; requires existing lineage gates and full-span words-of-record quality proof | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | | `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 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | diff --git a/config/feature-flags.yaml b/config/feature-flags.yaml index 9e9a327a4c1..08a870318f1 100644 --- a/config/feature-flags.yaml +++ b/config/feature-flags.yaml @@ -1,7 +1,7 @@ flags: - - {"key":"BATCH_PIPELINE_SWEEP_AGE_GATE","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","review_by":"2026-11-30","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} - - {"key":"BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","review_by":"2026-11-30","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} - - {"key":"BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","review_by":"2026-11-30","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} + - {"key":"BATCH_PIPELINE_SWEEP_AGE_GATE","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} + - {"key":"BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} + - {"key":"BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} - {"key":"PUSHER_MAX_ABANDON_RECORD_DRAINS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Per-pod abandon coordinators; default and prod4, clamped0..16","fail":"closed","owner":"two-lane lead","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"Zero refuses all overflow coordination. Full admission logs abandon_overflow and retains the old producer fence for process/pod-death repair. Coordinators wait for actual shared storage_executor futures to settle before attesting stoppage."} - {"key":"PUSHER_ABANDON_EVICTED_TOKENS","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"On unless off; capacity-evicted pusher drains attest token abandonment only after worker and executor settlement","fail":"open","owner":"two-lane lead","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"lead2 88ba4d. Exact-token abandoned_at never certifies a projected tail; repair requires archive and finish/tail proof. Metadata admission retries with bounded backoff; reconnect tokens stay live. Pusher train: one canary pod at 64, watch memory p99, capacity, abandoned and upload_pending before expansion. Rollback off plus cap32. See backend/docs/listen_pusher_pipeline/changes/2026-11-05-pusher-abandon-evicted-tokens.md."} - {"key":"PUSHER_MAX_CLOSED_RECORD_DRAINS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Per-process retained closed-record drain count; default32, clamped8..128, prod64","fail":"closed","owner":"two-lane lead","created":"2026-11-05","terminal":"keep","decision":"keep","notes":"lead2 88ba4d rejects prod128 until a retained-PCM byte budget exists. Invalid values use32. Startup selection logged once at first socket. Pusher train canary and per-pod memory p99 review required; rollback32."} From 4235431350a469cb9da5f9f31c7b60c17dbb7e63 Mon Sep 17 00:00:00 2001 From: David Zhang Date: Fri, 6 Nov 2026 00:43:59 +0700 Subject: [PATCH 4/5] fix(pipeline): preserve sweep progress and fresh coalesced work --- backend/database/outbox_provider_stt.py | 14 +++++- .../tests/unit/test_provider_stt_outbox.py | 49 +++++++++++++++++-- backend/tests/unit/test_sweep_age_gate.py | 48 ++++++++++++++++++ backend/utils/enrichment/outbox.py | 11 +++-- backend/utils/observability/sweep_age.py | 29 ++++++----- 5 files changed, 131 insertions(+), 20 deletions(-) diff --git a/backend/database/outbox_provider_stt.py b/backend/database/outbox_provider_stt.py index 3a298351491..40ef590d549 100644 --- a/backend/database/outbox_provider_stt.py +++ b/backend/database/outbox_provider_stt.py @@ -18,6 +18,7 @@ import uuid from typing import Any, Callable, Literal, Mapping +from config.policy import sweep_age from database.outbox import Claim, EnqueueResult, OutboxKind, Outcome, SweepCounts from utils.observability.sweep_age import skip_ancient @@ -104,6 +105,7 @@ def __init__( self._privacy_blocked = privacy_blocked self._create_wake = create_wake or _default_wake self._collection = collection + self._sweep_cursor: Any = None # The production factory caches this instance across maintenance ticks. # --- plumbing ------------------------------------------------------------------------------- def _client(self) -> Any: @@ -308,7 +310,17 @@ def sweep(self, *, now: float, limit: int = 100) -> SweepCounts: .where(filter=FieldFilter('state', '==', OPEN)) .limit(max(1, min(limit, 500))) ) - for snapshot in query.stream(): + # The state single-field index already includes document-ID ordering. Advance even over + # age-skipped rows, without writing them; a short/empty page wraps the next bounded pass. + paged = sweep_age.enabled() + if paged: + query = query.order_by('__name__') + if self._sweep_cursor is not None: + query = query.start_after(self._sweep_cursor) + page = list(query.stream()) + if paged: + self._sweep_cursor = page[-1] if len(page) == max(1, min(limit, 500)) else None + for snapshot in page: row = snapshot.to_dict() or {} try: if self._blocked(snapshot.id, row['owner_id']): diff --git a/backend/tests/unit/test_provider_stt_outbox.py b/backend/tests/unit/test_provider_stt_outbox.py index cb9cb2dc006..8fe2b937ecc 100644 --- a/backend/tests/unit/test_provider_stt_outbox.py +++ b/backend/tests/unit/test_provider_stt_outbox.py @@ -42,19 +42,31 @@ def update(self, data, option=None): class Query: - def __init__(self, store, name, filters=(), limit=None): + def __init__(self, store, name, filters=(), limit=None, after=None): self.store, self.name, self.filters, self._limit = store, name, filters, limit + self.after = after def where(self, *, filter): - return Query(self.store, self.name, (*self.filters, filter), self._limit) + return Query(self.store, self.name, (*self.filters, filter), self._limit, self.after) def limit(self, n): - return Query(self.store, self.name, self.filters, n) + return Query(self.store, self.name, self.filters, n, self.after) + + def order_by(self, field): + assert field == '__name__' + return self + + def start_after(self, snapshot): + return Query(self.store, self.name, self.filters, self._limit, snapshot.id) def stream(self): out = [] for (name, key), row in sorted(self.store.rows.items()): - if name == self.name and all(row.get(f.field_path) == f.value for f in self.filters): + if ( + name == self.name + and (self.after is None or key > self.after) + and all(row.get(f.field_path) == f.value for f in self.filters) + ): out.append(SimpleNamespace(id=key, to_dict=lambda row=row: dict(row))) return out[: self._limit] @@ -260,3 +272,32 @@ def test_ancient_sweep_keeps_existing_privacy_discard(box, monkeypatch): assert counts.discarded == 1 and counts.rewoken == 0 assert row(box)['closed_reason'] == 'privacy_blocked' assert not box.woken and counter._value.get() == before + + +def test_sweep_advances_past_ancient_page_and_wraps(box, monkeypatch): + from copy import deepcopy + + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') + now = 2_000_000.0 + collection = 'batch_pipeline_test_provider_stt_outbox' + ids = [f'{i:064x}' for i in range(4)] + for key in ids: + box.outbox.enqueue(key, OWNER, {'segment_id': key}, 100, now=100 if key != ids[-1] else now - 3600) + ancient = {key: deepcopy(box.client.rows[(collection, key)]) for key in ids[:-1]} + box.woken.clear() # Simulate every original wake being lost. + assert box.outbox.sweep(now=now, limit=3).rewoken == 0 + assert box.outbox.sweep(now=now, limit=3).rewoken == 1 + assert box.woken == [box.client.rows[(collection, ids[-1])]['task_token']] + assert {key: box.client.rows[(collection, key)] for key in ids[:-1]} == ancient + # A row inserted behind the cursor is reached after the short page wraps. + box.client.rows[(collection, ids[0])]['created_at'] = now - 3600 + assert box.outbox.sweep(now=now, limit=3).rewoken == 1 + + +def test_sweep_cursor_wraps_an_exactly_full_last_page(box, monkeypatch): + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') + enqueue(box, now=100) + assert box.outbox.sweep(now=2_000_000, limit=1).rewoken == 0 + assert box.outbox._sweep_cursor is not None + assert box.outbox.sweep(now=2_000_000, limit=1).rewoken == 0 + assert box.outbox._sweep_cursor is None diff --git a/backend/tests/unit/test_sweep_age_gate.py b/backend/tests/unit/test_sweep_age_gate.py index 7d01a6aa8ee..0db761acb49 100644 --- a/backend/tests/unit/test_sweep_age_gate.py +++ b/backend/tests/unit/test_sweep_age_gate.py @@ -137,3 +137,51 @@ def test_off_restores_existing_parking_policy(state, sweep_query, monkeypatch): sweep_query.stream.side_effect = lambda: [SimpleNamespace(id=key, to_dict=lambda: deepcopy(rows[key]))] assert outbox.sweep(now=2_000_000) == {'repaired': 0, 'error': 0} assert rows[key]['age_gate_parked'] and not tasks + + +def test_new_mark_on_ancient_open_fresh_row_is_rewoken(state, sweep_query, monkeypatch): + rows, tasks = state + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') + monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') + now = 2_000_000.0 + outbox.mark('u', 'c', 'en', now=100) + key = outbox.key('u', 'c') + rows[key]['dispatch']['queue'] = 'enrichment-fresh' + first_pending = rows[key]['first_pending_at'] + outbox.mark('u', 'c', 'en', now=now - 3600) + outbox.mark('u', 'c', 'en', now=now - 7200) # Delayed older append cannot regress the age. + assert rows[key]['first_pending_at'] == first_pending + assert rows[key]['latest_pending_at'] == now - 3600 + tasks.clear() # The coalesced mark's previous named task was lost. + sweep_query.stream.side_effect = lambda: [SimpleNamespace(id=key, to_dict=lambda: deepcopy(rows[key]))] + counter = SKIPPED.labels(queue='enrichment-fresh', reason='max_age') + before = counter._value.get() + assert outbox.sweep(now=now) == {'repaired': 1, 'error': 0} + assert len(tasks) == 1 and counter._value.get() == before + + +def test_ancient_fresh_sweep_runs_privacy_retirement_before_age_skip(state, sweep_query, monkeypatch): + rows, tasks = state + monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') + monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') + outbox.mark('u', 'c', 'en', now=100) + key = outbox.key('u', 'c') + rows[key]['dispatch']['queue'] = 'enrichment-fresh' + tasks.clear() + sweep_query.stream.side_effect = lambda: [SimpleNamespace(id=key, to_dict=lambda: deepcopy(rows[key]))] + counter = SKIPPED.labels(queue='enrichment-fresh', reason='max_age') + before = counter._value.get() + + from database.outbox_firestore import FirestoreEnrichmentOutbox + from unittest.mock import Mock + + probe = Mock(return_value=True) + adapter = FirestoreEnrichmentOutbox(privacy_blocked=probe) + assert outbox.sweep(now=2_000_000, discard=adapter._blocked) == { + 'repaired': 0, + 'error': 0, + 'discarded': 1, + 'expired': 0, + } + assert not rows and not tasks and counter._value.get() == before + probe.assert_called_once_with('u', {'conversation_id': 'c'}) diff --git a/backend/utils/enrichment/outbox.py b/backend/utils/enrichment/outbox.py index c035763199b..0d288808139 100644 --- a/backend/utils/enrichment/outbox.py +++ b/backend/utils/enrichment/outbox.py @@ -448,6 +448,10 @@ def update(old: Row | None) -> tuple[Row, tuple[Row, bool]]: # Timestamp every durable obligation, even before rollout. Neither an # append nor a retry owns the original acceptance time. row['first_pending_at'] = current if old and old.get('dormant') else row.get('first_pending_at', current) + # New coalesced work must not inherit an ancient obligation's sweep admission age. + row['latest_pending_at'] = ( + current if old and old.get('dormant') else max(row.get('latest_pending_at', current), current) + ) if snapshot_summary and (not old or old.get('dormant') or row.get('first_summary_policy') is None): if candidate is None: candidate = resolve_stage_policy( @@ -1023,6 +1027,10 @@ def update(old: Row | None) -> tuple[Row | None, None]: return update try: + # Privacy/deletion owns retirement, even when an ancient row would otherwise be held. + if discard is not None and discard(snapshot.id): + counts['discarded'] += 1 + continue # Skip before the independent historical-parking policy can mutate this row. candidate = getattr(snapshot, 'to_dict', lambda: {})() or {} if _skip_ancient_fresh(candidate, now=current): @@ -1060,9 +1068,6 @@ def replace(old: Row | None) -> tuple[Row | None, Row | None]: row = mutate(snapshot.id, replace) if status == 'unbound': UNBOUND.labels(action='repair').inc() - if row is not None and discard is not None and discard(snapshot.id): - counts['discarded'] += 1 - row = None if row is not None: try: enqueue(row) diff --git a/backend/utils/observability/sweep_age.py b/backend/utils/observability/sweep_age.py index c0475d232b1..1d928428df6 100644 --- a/backend/utils/observability/sweep_age.py +++ b/backend/utils/observability/sweep_age.py @@ -17,19 +17,24 @@ def skip_ancient(row, queue: str, *, now: float) -> bool: if not sweep_age.enabled(): return False - field = 'created_at' if queue == 'provider-stt' else 'first_pending_at' - value = row.get(field) - try: - if isinstance(value, str): - value = datetime.fromisoformat(value.replace('Z', '+00:00')) - if isinstance(value, datetime): - value = (value if value.tzinfo else value.replace(tzinfo=timezone.utc)).timestamp() - # Unknown metadata cannot prove that an obligation is ancient. - if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + fields = ('created_at',) if queue == 'provider-stt' else ('first_pending_at', 'latest_pending_at') + timestamps = [] + for field in fields: + if field == 'latest_pending_at' and field not in row: + continue # Existing rows retain their original acceptance clock until a new mark. + value = row.get(field) + try: + if isinstance(value, str): + value = datetime.fromisoformat(value.replace('Z', '+00:00')) + if isinstance(value, datetime): + value = (value if value.tzinfo else value.replace(tzinfo=timezone.utc)).timestamp() + # Unknown metadata cannot prove that an obligation is ancient. + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + return False + timestamps.append(value) + except (ValueError, OverflowError): return False - expired = now - value > sweep_age.max_age_seconds(queue) - except (ValueError, OverflowError): - return False + expired = now - max(timestamps) > sweep_age.max_age_seconds(queue) if expired: SKIPPED.labels(queue=queue, reason='max_age').inc() return expired From eee5427b9a8b9f4ab659ac125d6dddd207e6b442 Mon Sep 17 00:00:00 2001 From: David Zhang Date: Fri, 6 Nov 2026 01:05:09 +0700 Subject: [PATCH 5/5] fix(pipeline): preserve historical parking before sweep age skip --- backend/tests/unit/test_sweep_age_gate.py | 16 +++++++++++----- backend/utils/enrichment/outbox.py | 8 ++++---- backend/utils/observability/sweep_age.py | 2 +- 3 files changed, 16 insertions(+), 10 deletions(-) diff --git a/backend/tests/unit/test_sweep_age_gate.py b/backend/tests/unit/test_sweep_age_gate.py index 0db761acb49..f20b1705b01 100644 --- a/backend/tests/unit/test_sweep_age_gate.py +++ b/backend/tests/unit/test_sweep_age_gate.py @@ -25,9 +25,10 @@ def test_only_off_disables_gate(monkeypatch, raw): assert sweep_age.enabled() -@pytest.mark.parametrize('value', [None, 'bad', True, float('nan'), float('inf')]) -def test_unknown_age_keeps_wake(value): - assert not skip_ancient({'created_at': value}, 'provider-stt', now=2_000_000) +@pytest.mark.parametrize('queue,field', [('provider-stt', 'created_at'), ('enrichment-fresh', 'first_pending_at')]) +@pytest.mark.parametrize('value', [None, 'bad', True, 0, -1, float('nan'), float('inf')]) +def test_unknown_age_keeps_wake(queue, field, value): + assert not skip_ancient({field: value}, queue, now=2_000_000) @pytest.mark.parametrize( @@ -59,7 +60,7 @@ def test_enrichment_sweep_skips_without_mutation(state, sweep_query, monkeypatch assert rows == original -def test_ancient_fresh_sweep_precedes_existing_parking(state, sweep_query, monkeypatch): +def test_existing_parking_precedes_ancient_fresh_sweep(state, sweep_query, monkeypatch): rows, tasks = state monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') @@ -71,7 +72,12 @@ def test_ancient_fresh_sweep_precedes_existing_parking(state, sweep_query, monke monkeypatch.setenv('ENRICHMENT_MAX_AGE_HOURS', '72') sweep_query.stream.side_effect = lambda: [SimpleNamespace(id=key, to_dict=lambda: deepcopy(rows[key]))] assert outbox.sweep(now=2_000_000) == {'repaired': 0, 'error': 0} - assert rows == original and not tasks + parked = rows[key] + assert parked['age_gate_parked'] and parked['dormant'] and parked['due_at'] is None + assert parked['deferred_reason'] == 'age_gate_72h' and not tasks + assert parked['first_pending_at'] == original[key]['first_pending_at'] + assert parked['version'] == original[key]['version'] + assert parked['uid'] == original[key]['uid'] def test_other_enrichment_queues_keep_old_behavior(state, sweep_query, monkeypatch): diff --git a/backend/utils/enrichment/outbox.py b/backend/utils/enrichment/outbox.py index 0d288808139..eb0563a464b 100644 --- a/backend/utils/enrichment/outbox.py +++ b/backend/utils/enrichment/outbox.py @@ -1031,15 +1031,15 @@ def update(old: Row | None) -> tuple[Row | None, None]: if discard is not None and discard(snapshot.id): counts['discarded'] += 1 continue - # Skip before the independent historical-parking policy can mutate this row. - candidate = getattr(snapshot, 'to_dict', lambda: {})() or {} - if _skip_ancient_fresh(candidate, now=current): - continue if not injected: from utils.enrichment.age_gate import park_if_expired if park_if_expired(snapshot.id, None, now=current, client=bound): continue + # The existing privacy/parking policies take precedence; gate only remaining open debt. + candidate = getattr(snapshot, 'to_dict', lambda: {})() or {} + if _skip_ancient_fresh(candidate, now=current): + continue status, row = mutate(snapshot.id, repair) if status == 'expired' and 'expired' in counts: counts['expired'] += 1 diff --git a/backend/utils/observability/sweep_age.py b/backend/utils/observability/sweep_age.py index 1d928428df6..21ee16da855 100644 --- a/backend/utils/observability/sweep_age.py +++ b/backend/utils/observability/sweep_age.py @@ -29,7 +29,7 @@ def skip_ancient(row, queue: str, *, now: float) -> bool: if isinstance(value, datetime): value = (value if value.tzinfo else value.replace(tzinfo=timezone.utc)).timestamp() # Unknown metadata cannot prove that an obligation is ancient. - if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0: return False timestamps.append(value) except (ValueError, OverflowError):