diff --git a/backend/config/flag_budget.json b/backend/config/flag_budget.json index d5d25761441..21241602eef 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": 251, + "count": 254, "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"}, "PROJECTION_DRAIN_CLAIM_AWARE": {"owner": "two-lane lead", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"}, "BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED": {"owner": "omi-pipeline", "review_by": "2026-12-06", "registry": "config/feature-flags.yaml"}, "K_REVISION": {"owner": "omi-pipeline", "review_by": "2026-12-06"}, 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..40ef590d549 100644 --- a/backend/database/outbox_provider_stt.py +++ b/backend/database/outbox_provider_stt.py @@ -18,7 +18,9 @@ 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 Row = dict[str, Any] _ROW_ID = re.compile('[0-9a-f]{64}') @@ -103,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: @@ -307,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']): @@ -320,6 +333,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/docs/feature-flag-registry.md b/backend/docs/feature-flag-registry.md index 387ed11b01f..a52ccbbee6d 100644 --- a/backend/docs/feature-flag-registry.md +++ b/backend/docs/feature-flag-registry.md @@ -554,6 +554,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 | — | 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 | @@ -788,11 +789,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 | — | 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 | — | 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 | @@ -1077,6 +1080,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) @@ -1106,6 +1110,7 @@ their code default (`fail` tells you which way a missing value resolves). - `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_PROJECTION_STRIPING_ENABLED` — Default on unless off; seed projection discovery at an instance-specific committed_at stripe, then wrap through every row (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) @@ -1124,6 +1129,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..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] @@ -220,3 +232,72 @@ 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 + + +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 + + +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 new file mode 100644 index 00000000000..f20b1705b01 --- /dev/null +++ b/backend/tests/unit/test_sweep_age_gate.py @@ -0,0 +1,193 @@ +"""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('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( + '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_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') + 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} + 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): + 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 + + +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 bd8c0435c83..eb0563a464b 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 @@ -447,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( @@ -848,6 +853,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 +938,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 +1003,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,11 +1027,19 @@ 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 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 @@ -1040,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 new file mode 100644 index 00000000000..21ee16da855 --- /dev/null +++ b/backend/utils/observability/sweep_age.py @@ -0,0 +1,40 @@ +"""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 + 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) or value <= 0: + return False + timestamps.append(value) + 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 diff --git a/config/feature-flags.yaml b/config/feature-flags.yaml index 00b6e1dcd9d..e921cabe257 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","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":"BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on unless off; seed projection discovery at an instance-specific committed_at stripe, then wrap through every row","fail":"open","owner":"omi-pipeline","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"4096 stripes over oldest matching committed_at through startup time. SHA-256(revision:random process seed) selects the lower bound once per process without network I/O. One head read per scanner on startup; no permanent partition, no page-size, tick or claim changes. Sparse timestamps can share a start row. Off restores the exact unseeded cursor; pagination off also bypasses striping. Startup logs record stripe and stripe count."} - {"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."}