diff --git a/backend/charts/batch-pipeline-worker/templates/record-seal-repair.yaml b/backend/charts/batch-pipeline-worker/templates/record-seal-repair.yaml index 3dc7171a283..05608234995 100644 --- a/backend/charts/batch-pipeline-worker/templates/record-seal-repair.yaml +++ b/backend/charts/batch-pipeline-worker/templates/record-seal-repair.yaml @@ -53,6 +53,7 @@ spec: image: "{{ required "image.repository required" .Values.image.repository }}@{{ required "image.digest required" .Values.image.digest }}" command: [python, scripts/record_seal_repair_job.py] env: + - {name: PUSHER_PROJECTION_DEBT_DURABLE_ENABLED, value: {{ .Values.recordSealRepair.projectionDebtDurableEnabled | quote }}} - {name: RECORD_SEAL_REPAIR_JOBS, value: {{ .Values.recordSealRepair.jobsDiscovery | quote }}} - {name: RECORD_SEAL_REPAIR_ENABLED, value: {{ .Values.recordSealRepair.runtimeEnabled | quote }}} - {name: RECORD_AUDIO_PARTIAL_ENABLED, value: {{ .Values.recordSealRules.partialEnabled | quote }}} diff --git a/backend/charts/batch-pipeline-worker/values.yaml b/backend/charts/batch-pipeline-worker/values.yaml index d964bc3ca9e..aa6acd440f9 100644 --- a/backend/charts/batch-pipeline-worker/values.yaml +++ b/backend/charts/batch-pipeline-worker/values.yaml @@ -220,6 +220,7 @@ recordSealRepair: suspend: true runtimeEnabled: true jobsDiscovery: "on" + projectionDebtDurableEnabled: true mode: apply workers: 8 rowSeconds: 270 diff --git a/backend/config/batch_pipeline_projection_rebuild.py b/backend/config/batch_pipeline_projection_rebuild.py new file mode 100644 index 00000000000..1bd8417bff4 --- /dev/null +++ b/backend/config/batch_pipeline_projection_rebuild.py @@ -0,0 +1,7 @@ +"""Runtime kill switch for optional projection-debt handoff and maintenance.""" + +import os + + +def enabled() -> bool: + return os.getenv('PUSHER_PROJECTION_DEBT_DURABLE_ENABLED', '1').strip().lower() not in {'0', 'false', 'off'} diff --git a/backend/config/flag_budget.json b/backend/config/flag_budget.json index b86d632e70b..5d020c2b2e4 100644 --- a/backend/config/flag_budget.json +++ b/backend/config/flag_budget.json @@ -24,8 +24,9 @@ }, "seed_review_by": "2026-11-15", "budget": 169, - "count": 259, + "count": 260, "names": { + "PUSHER_PROJECTION_DEBT_DURABLE_ENABLED": {"owner": "uicleanup / omi-lifecycle", "review_by": "2026-12-06"}, "BATCH_PIPELINE_SERVING_PROJECTION_SWEEP": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"}, "BATCH_PIPELINE_PROJECTION_PUMP_ENABLED": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"}, "BATCH_PIPELINE_PUMP_SWEEP_PAGES": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"}, diff --git a/backend/database/record_projection_rebuild.py b/backend/database/record_projection_rebuild.py new file mode 100644 index 00000000000..ad5724def23 --- /dev/null +++ b/backend/database/record_projection_rebuild.py @@ -0,0 +1,107 @@ +"""Content-free playback repair debt; no seal, producer-token or object authority.""" + +import hashlib +import json +from datetime import datetime, timezone + +from google.cloud import firestore + +from database._client import get_firestore_client, run_transactional +from database.batch_pipeline_privacy import BatchPrivacyBlocked, BatchPrivacyReason, assert_batch_write_allowed + +COLLECTION = 'record_projection_rebuild_jobs' +KIND = 'record_projection_rebuild_due' +MAX_MARKERS_PER_BATCH = 256 + + +def marker_id(uid, cid): + for value in (uid, cid): + if not isinstance(value, str) or not value or '/' in value or len(value.encode()) > 256: + raise ValueError('bounded projection owner required') + return hashlib.sha256(json.dumps([uid, cid], separators=(',', ':')).encode()).hexdigest() + + +def mark_due(uid, cid, *, firestore_client=None): + client = firestore_client if firestore_client is not None else get_firestore_client() + client.collection(COLLECTION).document(marker_id(uid, cid)).set( + {'kind': KIND, 'owner_id': uid, 'conversation_id': cid, 'requested_at': datetime.now(timezone.utc)}, + timeout=5, + retry=None, + ) + + +def mark_many(uid, conversation_ids, *, firestore_client=None): + """One bounded cleanup RPC, rather than one timeout per socket obligation.""" + owners = set(conversation_ids) + if not 0 < len(owners) <= MAX_MARKERS_PER_BATCH: + raise ValueError('bounded projection marker batch required') + client = firestore_client if firestore_client is not None else get_firestore_client() + batch = client.batch() + requested_at = datetime.now(timezone.utc) + for cid in owners: + ref = client.collection(COLLECTION).document(marker_id(uid, cid)) + batch.set(ref, {'kind': KIND, 'owner_id': uid, 'conversation_id': cid, 'requested_at': requested_at}) + batch.commit(timeout=1, retry=None) + + +def retire_empty(uid, cid, *, observed_update_time, firestore_client=None): + """Retire only observed empty debt with a transactionally clear producer map. + + An absent or malformed map is unknown, not proof of quiescence. Reading + both documents makes a producer mint or marker replacement before commit + conflict with deletion. This never clears or interprets producer tokens. + """ + if observed_update_time is None: + raise ValueError('observed projection marker version required') + client = firestore_client if firestore_client is not None else get_firestore_client() + marker = client.collection(COLLECTION).document(marker_id(uid, cid)) + owner = client.collection('users').document(uid).collection('conversations').document(cid) + + @firestore.transactional + def commit(tx): + observed = marker.get(transaction=tx) + if not observed.exists or observed.update_time != observed_update_time: + return 'changed' + snapshot = owner.get(transaction=tx) + sessions = (snapshot.to_dict() or {}).get('record_upload_sessions') + if not snapshot.exists or not isinstance(sessions, dict) or sessions: + return 'producer_fenced' + tx.delete(marker) + return 'retired_no_objects' + + return run_transactional(client, commit, transient=True) + + +def apply_projection(uid, cid, payload, *, expected_audio_files, firestore_client=None): + """Cold repair only: deletion intent and projection CAS share the commit.""" + from database.conversations import AudioFilesChanged, audio_files_revision, retain_aliases + + client = firestore_client if firestore_client is not None else get_firestore_client() + ref = client.collection('users').document(uid).collection('conversations').document(cid) + + @firestore.transactional + def commit(tx): + snapshot = ref.get(transaction=tx) + if not snapshot.exists: + return False + row = snapshot.to_dict() or {} + if row.get('deleted') or row.get('discarded') or row.get('private_cloud_sync_enabled') is False: + return False + try: + assert_batch_write_allowed(client, tx, uid, conversation_ids=(cid,)) + except BatchPrivacyBlocked as blocked: + if blocked.reason in { + BatchPrivacyReason.ACCOUNT_DELETED, + BatchPrivacyReason.ACCOUNT_MISSING, + BatchPrivacyReason.CONVERSATION_DELETED, + }: + return False + return None # Migration, generation holds and unknown refusals retain debt without a write. + if audio_files_revision(row.get('audio_files') or []) != audio_files_revision(expected_audio_files): + raise AudioFilesChanged('audio projection changed') + tx.update(ref, {'audio_files': retain_aliases(payload['audio_files'], row.get('audio_files') or [])}) + # A retained socket worker may still publish a later tail. Its existing + # producer fence is read-only here: repair never clears or seals it. + return None if row.get('record_upload_sessions') else True + + return run_transactional(client, commit, transient=True) diff --git a/backend/docs/feature-flag-registry.md b/backend/docs/feature-flag-registry.md index 3dc476bfc8e..92b01700d4a 100644 --- a/backend/docs/feature-flag-registry.md +++ b/backend/docs/feature-flag-registry.md @@ -714,6 +714,7 @@ and an explicit empty literal renders as `''`. | `PUBLIC_SHARED_CONVERSATION_CHAT_MODE` | Enable chat on public shared conversations | backend | env | closed | declared (gke/backend-listen); gateway (cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill) | gateway (backend-listen (chart), cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | gateway (cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill); off (backend-listen (chart), gke/backend-listen) | — | keep | keep | — | dazheng | | `PUSHER_ABANDON_EVICTED_TOKENS` | On unless off; capacity-evicted pusher drains attest token abandonment only after worker and executor settlement | backend | env | open | — | — | on (pusher (chart)); {value: 'on', category: ops_kill} (gke/pusher) | — | keep | keep | — | two-lane lead | | `PUSHER_AUDIO_FILE_INCREMENTAL_ENABLED` | Default on; refresh the live pusher audio_files projection from exact uploaded-object metadata between full prefix rebuilds; 0 restores per-flush listing | backend | env | open | — | — | — | — | keep | keep | — | planscost | +| `PUSHER_PROJECTION_DEBT_DURABLE_ENABLED` | Default on; persist exhausted or evicted playback projection debt and rebuild it in the record repair CronJob | backend | env | open | — | — | — | — | keep | keep | — | uicleanup / omi-lifecycle | | `PUSHER_PROJECTION_QUEUE_WAIT_BOUND_ENABLED` | Default on; bound only committed pusher projection queue capacity waits to the existing 300-second receive-liveness budget | backend | env | open | — | — | — | — | keep | keep | — | two-lane lead | | `RATE_LIMIT_SHADOW_MODE` | Shadow backend rate limits | backend | env | closed | — | — | — | — | keep | keep | — | unowned | | `RECORDING_SPEAKER_RENAME_ENABLED` | Default on; bounded scoped recording speaker naming. Off reports disabled and clients use explicit single-conversation assignments. | backend, mobile | env | open | — | — | — | — | keep | keep | — | speakerrename | @@ -1379,6 +1380,7 @@ their code default (`fail` tells you which way a missing value resolves). - `PLAYBACK_MANIFEST_RECOVERY_ENABLED` — Default on; next-access recovery for merge_failed playback markers after 15 minutes, deduped by marker generation (fail: closed) - `PROJECTION_DRAIN_CLAIM_AWARE` — Default on; read-only fleet claim admission and jittered backoff for redundant claim-busy owner-drain requests (fail: open) - `PUSHER_AUDIO_FILE_INCREMENTAL_ENABLED` — Default on; refresh the live pusher audio_files projection from exact uploaded-object metadata between full prefix rebuilds; 0 restores per-flush listing (fail: open) +- `PUSHER_PROJECTION_DEBT_DURABLE_ENABLED` — Default on; persist exhausted or evicted playback projection debt and rebuild it in the record repair CronJob (fail: open) - `PUSHER_PROJECTION_QUEUE_WAIT_BOUND_ENABLED` — Default on; bound only committed pusher projection queue capacity waits to the existing 300-second receive-liveness budget (fail: open) - `RATE_LIMIT_SHADOW_MODE` — Shadow backend rate limits (fail: closed) - `RECORDING_SESSION_MODE` — Select recording session migration mode (fail: closed) diff --git a/backend/docs/listen_pusher_pipeline.mdx b/backend/docs/listen_pusher_pipeline.mdx index ce53355dd77..09d80cc04f0 100644 --- a/backend/docs/listen_pusher_pipeline.mdx +++ b/backend/docs/listen_pusher_pipeline.mdx @@ -9351,3 +9351,67 @@ owner-drain admissions use a read-only Redis claim GET and fail open on errors. Claim-busy re-requests through `again` back off 5–15 seconds; productive `more=True` continuations retain immediate FIFO progress without the peek. The advisory claim and transactional ordering fences remain unchanged. + + +### Durable playback projection rebuild debt + +`PUSHER_PROJECTION_DEBT_DURABLE_ENABLED` defaults **on**; `0`, `false`, or `off` +disables pusher handoff and repair consumption. Pusher receives the flag through +its existing `runtimeConfigMapName` `envFrom` binding; no literal env entry +shadows that ConfigMap key. The repair CronJob declares it independently through +`recordSealRepair.projectionDebtDurableEnabled` (default `true`). Rollback sets +the consumer option `false` first, then the Pusher runtime ConfigMap flag off. +Re-enabling resumes retained markers. Consumption also requires the existing +repair CronJob to be enabled, unsuspended and runtime-enabled; this contract +does not activate it. + +Exhausted close retries, empty close discovery, socket cleanup, and terminal +transient/InvalidArgument batch failures hand off one +`record_projection_rebuild_due` marker per user/conversation into +`record_projection_rebuild_jobs/{sha256(JSON([uid,cid]))}`. Fields are `kind`, +`owner_id`, `conversation_id`, and UTC `requested_at`; no content or conversation +status is stored. Successful handoffs coalesce. Terminal handoffs enqueue into +one metadata-only background task and never await Firestore or change recording +completion. A separate bounded retry inventory holds up to 128 overflow owners; +cleanup can add the 128 cached owners (256 total). Each task attempts at most two +batch writes with a one-second SDK timeout and 1.25-second async attempt budget. +Cleanup shares one three-second total budget across every marker and retry; +marker failure never gates socket close or recording completion. Failure emits +`marker_write_failed`; inventory saturation emits `marker_retry_overflow`, with +sanitized logs and `omi_pusher_projection_debt_total` outcomes. Persistent storage +failure, saturation, or process death before commit can still prevent handoff; +the marker is not a WAL. + +The existing repair CronJob runs a separate 60-second metadata-only pass over +at most 128 markers ordered by request time. A bounded collection scan reads at +most 10,001 markers; the maintenance ceiling is 10,000. A confirmed empty +inventory retires as `retired_no_objects` after seven days from Firestore document +creation (request-time rotation and coalesced writes do not reset that age), or +immediately under collection pressure. Retiring deletes only the observed marker, +never audio. Empty retirement rereads the observed marker version and the +conversation's `record_upload_sessions` in the same transaction as marker deletion. +Only an explicit empty producer map permits retirement; nonempty, missing or +malformed maps retain and rotate the marker. A producer mint or marker replacement +before commit conflicts with deletion. No producer token is changed or declared +dead. Nonempty or privacy-held debt remains due even above the ceiling; +capacity is enforced by retiring confirmed empty work, not discarding useful debt. +The aggregate report and Prometheus gauges expose marker count (saturated at +10,001), oldest age in that bounded scan, and capacity state. `capacity_exceeded` +is alertable. `omi_projection_rebuild_pressure_blocked` plus aggregate +`pressure_fenced_count`/`pressure_blocked` reports a pass whose empty candidates +could not be evicted because their producer fences were not clear. These gauges +are bounded observations, not an exact count/oldest +claim when the collection is above the ceiling. Apply rotates discovered markers +before validation/discovery, so refusal, interruption or malformed ordered +markers cannot monopolize the page. Preview never rebuilds or writes. Full +rebuild refuses a tombstone or absent conversation before discovery, then +transactionally rechecks privacy/deletion and the prior `audio_files` revision. +Only `audio_files` may change. Temporary privacy/generation holds retain debt +without metadata writes. Nonempty existing producer fences retain the marker +even after metadata refresh, until their existing owner settles them. No seal +or token authority is added. Raw-index failure retains debt; confirmed empty +discovery uses the bounded retirement policy above. +Successful final rebuild or permanent owner revocation clears only the observed +marker version; a replacement marker always wins. No PCM or audio-object +mutation occurs. The [phase 5C change note](./listen_pusher_pipeline/changes/2026-11-06-durable-projection-debt.md) +records scope and verification limits. diff --git a/backend/docs/listen_pusher_pipeline/changes/2026-11-06-durable-projection-debt.md b/backend/docs/listen_pusher_pipeline/changes/2026-11-06-durable-projection-debt.md new file mode 100644 index 00000000000..8a697839d3a --- /dev/null +++ b/backend/docs/listen_pusher_pipeline/changes/2026-11-06-durable-projection-debt.md @@ -0,0 +1,66 @@ +# Durable playback projection debt (phase 5C) + +After the bounded close rebuild retries fail, pusher hands playback-only debt +into `record_projection_rebuild_jobs/{sha256([uid,cid])}` before discarding its +socket state. Socket cleanup also hands off outstanding debt. Terminal transient/InvalidArgument batch +projection failures persist by exact owner even if the bounded close cache is full. The marker kind +is `record_projection_rebuild_due`; its only other fields are owner ID, +conversation ID and request time. Repeated handoffs coalesce by owner. +Ordinary flushes add no reads or writes. Terminal marker handoffs enqueue into +one bounded background task. A separate metadata-only retry inventory preserves +overflow outside the close cache; marker failure cannot change recording-close +success. Socket cleanup has one three-second total deadline for batched writes +and retries, independent of debt count. `marker_write_failed` and bounded-list +saturation emit sanitized logs and counter outcomes. Temporary privacy holds +retain markers without a metadata update. + +The [owning contract](../../listen_pusher_pipeline.mdx#durable-playback-projection-rebuild-debt) +defines payloads, ordering, limits, gauges, flag bindings and rollback. Process +kill, persistent Firestore failure, or saturation can still prevent handoff; +this marker is not a WAL. + +The existing record-seal-repair CronJob runs a separate 60-second pass before +seal work, selecting at most 128 markers by request time. It uses a fresh, +socket-independent full metadata rebuild even if the incremental optimization +is disabled. Successful rebuild or revoked owner clears only the observed +marker version. Existing nonempty producer fences retain the marker after a +metadata update: a retained worker may still upload a later tail. This reads +the existing map and never clears a token or grants seal authority. Incomplete discovery or failure retains and rotates the +marker; malformed markers also rotate. Concurrent replacement wins over cleanup. Preview mode never rebuilds +or writes. Reports contain aggregate dispositions, never owner IDs or paths. +The existing CronJob suspension/runtime settings still govern execution; no +CronJob is unsuspended by this code change. + +`projection_owner_gone` refuses a tombstone or absent row before discovery. +The cold projection transaction rechecks account/conversation privacy, +tombstones, soft deletion, sync opt-out and the prior audio-files revision +before updating only `audio_files`. A deletion or another projection arriving +during discovery cannot recreate the owner or overwrite a newer projection. +Raw-index failure cannot count as successful durable repair, including forced +socket close. Empty close discovery hands off its obligation too. Empty live +inventory retains the marker until seven days from Firestore creation, then +retires as `retired_no_objects`. A 10,000-marker maintenance ceiling also retires +confirmed empty work under pressure, never useful or privacy-held debt. Per-pass +work stays capped at 128; the collection scan caps at 10,001. Aggregate logs and +Prometheus gauges expose bounded count, observed oldest age and capacity state. +No retirement deletes audio. Empty-marker retirement checks the observed marker +version and the conversation's explicit empty producer map transactionally at +retirement time, not discovery time. Unknown or nonempty producer maps retain +and rotate debt, including under collection pressure. A late tail still has a +marker to repair its playback metadata. The pressure-blocked gauge and aggregate +fenced count make an all-fenced maintenance pass alertable. No producer token +or seal field is written. + +`PUSHER_PROJECTION_DEBT_DURABLE_ENABLED` defaults **on**; `0`, `false`, or `off` +disables both handoff and consumption. Rollback disables the consumer runtime +first (`recordSealRepair.projectionDebtDurableEnabled=false`), then pusher; existing markers remain inert. Re-enabling the flag resumes retained markers. +A normal socket close does not probe or clear a preexisting marker: the cold +consumer clears it after its own full rebuild, avoiding hot-path writes. + +This adds no seal authority, producer token changes, entry clock, heartbeat, +dead-letter, audio-object mutation or PCM reads. Object discovery reads metadata +only, including the existing raw-pack index. No production measurements, IAM +checks, cloud access, live CronJob activation, end-user playback acceptance or +cost/latency claims are included. Hermetic unit tests and lifecycle emulator +scenarios exercise exhaustion, eviction, deletion during discovery and marker +replacement. Phase 5 A/B simulator changes remain outside this lane. diff --git a/backend/routers/pusher.py b/backend/routers/pusher.py index deb77abc785..f38b1427a01 100644 --- a/backend/routers/pusher.py +++ b/backend/routers/pusher.py @@ -21,7 +21,8 @@ from utils.pusher_audio_window import append_window_run, v2_overlap_conflict, AUDIO_TIMELINE_CONTINUITY_TOLERANCE from config.audio_custody import framed_pack from config.audio_outbox import enabled as audio_outbox_enabled -from utils.pusher_audio_projection import PusherAudioProjection, projection_owner_gone +from utils.pusher_audio_projection import PusherAudioProjection +from utils.batch_pipeline.record_projection_rebuild import projection_owner_gone from utils.pusher_record_barrier import PusherRecordBarrier from utils.pusher_receive_metrics import SocketObservation, observe_work from utils.pusher_record_finish import ( @@ -545,6 +546,7 @@ async def flush_refs(conv_id: str) -> None: logger.warning('Durable audio projection conflict; archive receipt retained') except InvalidArgument as exc: audio_projection.inventory.pop(conv_id, None) + await audio_projection.defer_rebuild(conv_id) # Custody and its generation-pinned manifest remain durable. # Retrying a deterministic projection rejection cannot repair # the payload, and a sticky custody failure would strand it. @@ -629,6 +631,9 @@ async def _transient_projection_failure(exc: Exception, conv_id: str, refs: List # counting toward the bounded retry map. projection_retries.pop(conv_id, None) audio_projection.inventory.pop(conv_id, None) + # The close-debt cache may be full. Persist terminal projection debt + # by its exact owner rather than relying on that bounded inventory. + await audio_projection.defer_rebuild(conv_id) await record_barrier.fail_recording(conv_id) record_failure('projection', reason='transient') record_fallback( @@ -1822,6 +1827,7 @@ async def drain_socket_tasks() -> None: all_to_cancel = [t for t in bg_main_tasks if not t.done()] await drain_tasks(all_to_cancel, timeout=5.0, label="pusher_cleanup", cancel=True) + await audio_projection.persist_pending() if application_failed and observation.reason == 'peer': observation.reason = 'application_error' diff --git a/backend/scripts/record_seal_repair_job.py b/backend/scripts/record_seal_repair_job.py index b2570a16697..6aef2caf1ff 100644 --- a/backend/scripts/record_seal_repair_job.py +++ b/backend/scripts/record_seal_repair_job.py @@ -19,7 +19,11 @@ def emit(payload): logger = logging.getLogger('record-seal-repair') - level = logging.ERROR if payload.get('outcome') == 'error' else logging.INFO + level = ( + logging.ERROR + if payload.get('outcome') == 'error' + else logging.WARNING if payload.get('outcome') == 'capacity_exceeded' else logging.INFO + ) record = logger.makeRecord( logger.name, level, '', 0, json.dumps(payload, sort_keys=True, allow_nan=False), (), None ) @@ -43,6 +47,14 @@ def terminate(*_): output.write(json.dumps(repair.row_main(payload)) + '\n') # Private child IPC, closed outcome only. output.flush() elif not sys.argv[1:]: + try: + from utils import record_projection_rebuild + + report = record_projection_rebuild.run_job() + if report['outcome'] != 'disabled': + emit(report) + except Exception: + emit({'event': 'record_projection_rebuild', 'outcome': 'error'}) emit(repair.run(progress=emit)) else: raise ValueError('invalid arguments') diff --git a/backend/testing/lifecycle_sim/test_scenarios.py b/backend/testing/lifecycle_sim/test_scenarios.py index dd3456db983..d0ae4f35d0f 100644 --- a/backend/testing/lifecycle_sim/test_scenarios.py +++ b/backend/testing/lifecycle_sim/test_scenarios.py @@ -11,6 +11,7 @@ import httpx from google.api_core.exceptions import Aborted +from database.conversations import create_audio_files_from_chunks from database import record_audio from database import conversation_finalization_jobs as finalization_jobs from testing.lifecycle_sim.faults import BoundaryFault, InjectedLifecycleFault @@ -333,3 +334,220 @@ async def test_server_batch_completion_ordering(lifecycle_sim, closed): row['status'] != 'completed' or (row.get('record_audio_closed') and not row.get('record_upload_sessions')), 'conversation completed while audio was open and its upload session remained pending', ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('ending', ['exhaustion', 'eviction', 'tombstone', 'missing', 'late_tombstone', 'marker_race']) +async def test_projection_close_debt_survives_socket_and_respects_privacy(lifecycle_sim, monkeypatch, ending): + from database import record_projection_rebuild as jobs + from utils import pusher_audio_projection as projection + from utils import record_projection_rebuild as consumer + from utils.batch_pipeline.record_seal_budget import SweepBudget + + sim = lifecycle_sim + sim.seed(sessions={'live': True}) + # Earlier summary scenarios install an empty playback leaf globally. This + # scenario must exercise the real grouping implementation captured at import. + monkeypatch.setattr(processing.conversations_db, 'create_audio_files_from_chunks', create_audio_files_from_chunks) + monkeypatch.setenv('PUSHER_PROJECTION_DEBT_DURABLE_ENABLED', '1') + monkeypatch.setenv('PUSHER_AUDIO_FILE_INCREMENTAL_ENABLED', '1') + before = sim.row() + + def retained_objects(): + bucket = sim.storage.bucket(os.environ['BUCKET_PRIVATE_CLOUD_SYNC']) + return { + blob.name: (blob.generation, blob.metadata, blob.download_as_bytes()) + for blob in bucket.list_blobs(prefix=f'chunks/{sim.uid}/{sim.cid}/') + } + + retained_before = retained_objects() + assert retained_before + from utils.batch_pipeline import record_projection_rebuild as full_projection + + discovery = full_projection.discover(sim.uid, sim.cid, strict_raw=True) + assert discovery is not None and discovery[2], discovery + failures = [] + original_rebuild = consumer.rebuild + + def checked_rebuild(*args): + try: + return original_rebuild(*args) + except Exception as error: + failures.append(repr(error)) + raise + + monkeypatch.setattr(consumer, 'rebuild', checked_rebuild) + cache = projection.PusherAudioProjection(sim.uid) + cache.has_uploaded_paths = True + cache.close_debt.add(sim.cid) + marker = sim.stack.firestore.collection(jobs.COLLECTION).document(jobs.marker_id(sim.uid, sim.cid)) + try: + if ending == 'exhaustion': + + def fail(*_): + raise OSError('synthetic projection failure') + + monkeypatch.setattr(cache, 'finish', fail) + assert await cache.finish_pending(sim.cid, set(), set()) is False + assert await cache.finish_pending(sim.cid, set(), set()) is False + assert await cache.finish_pending(sim.cid, set(), set()) is True + await cache.persist_pending() + assert marker.get().exists + tombstone = sim.ref.parent.parent.collection('deleted_conversations').document(sim.cid) + if ending == 'tombstone': + tombstone.set({'deleted': True}) + elif ending == 'missing': + sim.ref.delete() + elif ending == 'late_tombstone': + apply = jobs.apply_projection + + def race(*args, **kwargs): + tombstone.set({'deleted': True}) + return apply(*args, **kwargs) + + monkeypatch.setattr(jobs, 'apply_projection', race) + elif ending == 'marker_race': + rebuild = consumer.rebuild + + def race(*args): + result = rebuild(*args) + jobs.mark_due(*args, firestore_client=sim.stack.firestore) + return result + + monkeypatch.setattr(consumer, 'rebuild', race) + report = consumer.run(budget=SweepBudget(30), firestore_client=sim.stack.firestore) + if ending in {'exhaustion', 'eviction', 'marker_race'}: + # Existing live producer fences prevent retirement of debt published + # while a retained worker can still upload. This is a synthetic + # fixture transition, never a consumer token/seal mutation. + assert marker.get().exists + assert sim.row()['record_upload_sessions'] == {'live': True} + sim.ref.update({'record_upload_sessions': {}}) + before['record_upload_sessions'] = {} + report = consumer.run(budget=SweepBudget(30), firestore_client=sim.stack.firestore) + assert not failures + assert marker.get().exists is (ending == 'marker_race'), report + if ending == 'missing': + assert not sim.ref.get().exists + else: + after = sim.row() + if ending in {'tombstone', 'late_tombstone'}: + assert after == before + else: + assert after.get('audio_files') + assert {k: v for k, v in after.items() if k != 'audio_files'} == before + # The retained PCM and its metadata are untouched by the repair pass. + assert live.list_audio_chunks(sim.uid, sim.cid, require_archive=True) + assert retained_objects() == retained_before + finally: + marker.delete() + + +@pytest.mark.parametrize( + 'ending', ['aged_fenced', 'aged_clear', 'pressure_fenced', 'late_fence', 'marker_race', 'tail'] +) +def test_empty_projection_retirement_rechecks_producer_fence(lifecycle_sim, monkeypatch, ending): + from database import record_projection_rebuild as jobs + from database.audio_timeline import span_blob_metadata + from utils import record_projection_rebuild as consumer + from utils.batch_pipeline.record_seal_budget import SweepBudget + + sim = lifecycle_sim + sim.seed() + cid = sim.cid + '-empty' + ref = sim.ref.parent.document(cid) + row = {**sim.row(), 'id': cid, 'audio_files': [], 'record_upload_sessions': {}} + if ending in {'aged_fenced', 'pressure_fenced', 'tail'}: + row['record_upload_sessions'] = {'live': True} + ref.set(row) + # CI shares this collection with earlier stack scenarios. Include a clear + # peer locally too, so the test cannot assume aggregate counts belong only + # to its fenced owner. Only clear candidates may be pressure-evicted. + peer_cid = cid + '-peer' + peer_ref = sim.ref.parent.document(peer_cid) + peer_row = {**row, 'id': peer_cid, 'record_upload_sessions': {}} + peer_ref.set(peer_row) + peer_marker = sim.stack.firestore.collection(jobs.COLLECTION).document(jobs.marker_id(sim.uid, peer_cid)) + jobs.mark_due(sim.uid, peer_cid, firestore_client=sim.stack.firestore) + marker = sim.stack.firestore.collection(jobs.COLLECTION).document(jobs.marker_id(sim.uid, cid)) + jobs.mark_due(sim.uid, cid, firestore_client=sim.stack.firestore) + observed = marker.get() + monkeypatch.setenv('PUSHER_PROJECTION_DEBT_DURABLE_ENABLED', '1') + monkeypatch.setattr(processing.conversations_db, 'create_audio_files_from_chunks', create_audio_files_from_chunks) + # Firestore create_time is immutable. Exercise age eligibility immediately; + # the unit contract separately supplies an actual eight-day-old snapshot. + monkeypatch.setattr(consumer, 'NO_OBJECTS_MAX_AGE_SECONDS', 0) + if ending == 'pressure_fenced': + marker_count = len(list(sim.stack.firestore.collection(jobs.COLLECTION).stream())) + assert 2 <= marker_count <= consumer.MAX_PER_PASS + monkeypatch.setattr(consumer, 'MAX_COLLECTION_MARKERS', marker_count - 1) + rebuild = consumer.rebuild + retire = jobs.retire_empty + decisions = [] + + def checked_retirement(uid, conversation_id, **kwargs): + result = retire(uid, conversation_id, **kwargs) + if (uid, conversation_id) == (sim.uid, cid): + decisions.append(result) + return result + + monkeypatch.setattr(jobs, 'retire_empty', checked_retirement) + + def after_discovery(uid, conversation_id): + result = rebuild(uid, conversation_id) + if (uid, conversation_id) != (sim.uid, cid): + return result + assert result == 'no_objects' + if ending == 'late_fence': + assert record_audio.set_record_upload_pending( + uid, conversation_id, 'late', True, firestore_client=sim.stack.firestore + ) + elif ending == 'marker_race': + jobs.mark_due(uid, conversation_id, firestore_client=sim.stack.firestore) + return result + + try: + # This new conversation genuinely has no published audio prefix yet. + with monkeypatch.context() as first: + first.setattr(consumer, 'rebuild', after_discovery) + report = consumer.run(budget=SweepBudget(30), firestore_client=sim.stack.firestore) + assert not peer_marker.get().exists + assert peer_ref.get().to_dict() == peer_row + if ending == 'aged_clear': + assert not marker.get().exists and ref.get().to_dict() == row + assert decisions == ['retired_no_objects'] + return + assert marker.get().exists + assert marker.get().update_time != observed.update_time + if ending == 'marker_race': + assert decisions == ['changed'] + else: + assert decisions == ['producer_fenced'] + assert ref.get().to_dict()['record_upload_sessions'] + if ending == 'pressure_fenced': + assert report['gauges']['pressure_blocked'] is False # The clear peer was evicted. + assert report['gauges']['pressure_fenced_count'] >= 1 + if ending == 'tail': + # Only the synthetic producer fixture publishes this late object. + # The consumer must still own a marker after its earlier empty pass. + start = row['started_at'].timestamp() + blob = sim.storage.bucket(os.environ['BUCKET_PRIVATE_CLOUD_SYNC']).blob( + f'chunks/{sim.uid}/{cid}/{start:.3f}.batch.bin' + ) + blob.metadata = span_blob_metadata({'start': start, 'samples': 16000, 'sample_rate': 16000}) + blob.upload_from_string(b'\x01\x00' * 16000, if_generation_match=0) + retained = (blob.generation, blob.metadata, blob.download_as_bytes()) + report = consumer.run(budget=SweepBudget(30), firestore_client=sim.stack.firestore) + assert ref.get().to_dict()['audio_files'], report + assert marker.get().exists # The publishing producer still owns its token. + assert record_audio.set_record_upload_pending( + sim.uid, cid, 'live', False, firestore_client=sim.stack.firestore + ) + consumer.run(budget=SweepBudget(30), firestore_client=sim.stack.firestore) + assert not marker.get().exists + assert (blob.generation, blob.metadata, blob.download_as_bytes()) == retained + finally: + marker.delete() + ref.delete() + peer_marker.delete() + peer_ref.delete() diff --git a/backend/tests/support/firestore_batch_pipeline_drivers.py b/backend/tests/support/firestore_batch_pipeline_drivers.py index a13d7a66eeb..d54c7b27e2c 100644 --- a/backend/tests/support/firestore_batch_pipeline_drivers.py +++ b/backend/tests/support/firestore_batch_pipeline_drivers.py @@ -2,13 +2,26 @@ from datetime import datetime, timezone from types import SimpleNamespace +from unittest.mock import patch from tests.support.firestore_query_drivers import DriverEntry, CoveredByEntry, ref_collection +def _projection_debt_on(client): + from utils import record_projection_rebuild + + return patch.object(record_projection_rebuild, 'durable_debt_enabled', lambda: True) + + def entries(): namespaces = ['batch_pipeline_prod', 'batch_pipeline_dev'] return ( + DriverEntry( + 'utils.record_projection_rebuild.run', + domains={'mode': ['apply', 'plan']}, + neutrals={'budget': (SimpleNamespace(check=lambda: None, timeout=lambda: 10), 'RPC deadline only')}, + patchers=(_projection_debt_on,), + ), DriverEntry( 'database.record_seal_worker.pending_page', domains={'namespace': namespaces, 'after': [None, (datetime(2026, 10, 1, tzinfo=timezone.utc), 'pending')]}, diff --git a/backend/tests/unit/test_audio_durable_outbox.py b/backend/tests/unit/test_audio_durable_outbox.py index 3cc47fec379..a9bbcf67a36 100644 --- a/backend/tests/unit/test_audio_durable_outbox.py +++ b/backend/tests/unit/test_audio_durable_outbox.py @@ -1,6 +1,7 @@ """Fault injection for the capture -> immutable receipt custody boundary.""" from utils import pusher_audio_projection +from utils.batch_pipeline import record_projection_rebuild as projection_rebuild import asyncio from concurrent.futures import Future from collections import deque @@ -106,6 +107,7 @@ def list_blobs(self, prefix, **kwargs): @pytest.fixture def remote(monkeypatch): monkeypatch.setenv('PUSHER_AUDIO_FILE_INCREMENTAL_ENABLED', '0') + monkeypatch.setattr(pusher_audio_projection.record_projection_rebuild, 'mark_many', MagicMock()) monkeypatch.setattr(pusher_audio_projection.conversations_db, 'get_conversation_raw_snapshot', lambda *a, **kw: {}) bucket = Bucket() monkeypatch.setattr(record_audio_db, 'seal_record_audio', MagicMock(return_value='not_batch_owner')) @@ -2272,7 +2274,7 @@ def purged_read(uid, receipt): failures = _projection_harness(remote, monkeypatch, MagicMock(return_value=[])) monkeypatch.setattr(pusher, 'read_pcm', purged_read) monkeypatch.setattr( - pusher_audio_projection.conversation_tombstones, 'is_deleted', lambda uid, conv_id: tombstoned['now'] + projection_rebuild.conversation_tombstones, 'is_deleted', lambda uid, conv_id: tombstoned['now'] ) failure_marker = record_audio_db.mark_record_upload_failed # the fixture's MagicMock sock = Socket([struct.pack('= 2 + assert any('c' in call.args[1] for call in projection.record_projection_rebuild.mark_many.call_args_list) + + +@pytest.mark.asyncio +async def test_cleanup_batches_all_obligations_and_retains_them_on_failure(world): + cache = projection.PusherAudioProjection('u') + cache.close_debt.update(str(i) for i in range(128)) + cache.marker_retry['overflow'] = 1 + expected = cache.close_debt | {'overflow'} + write = projection.record_projection_rebuild.mark_many + write.side_effect = OSError('synthetic marker outage') + await cache.persist_pending() + assert cache.close_debt | set(cache.marker_retry) == expected + assert all(call.args == ('u', expected) for call in write.call_args_list) + write.side_effect = None + await cache.persist_pending() + assert not cache.close_debt and not cache.marker_retry + + +@pytest.mark.asyncio +async def test_all_close_markers_share_small_total_deadline(world, monkeypatch): + import asyncio + import time + + cache = projection.PusherAudioProjection('u') + cache.close_debt.update(str(i) for i in range(128)) + monkeypatch.setattr(projection, 'MARKER_ATTEMPT_SECONDS', 0.02) + monkeypatch.setattr(projection, 'MARKER_CLOSE_SECONDS', 0.05) + + async def stuck(*args): + await asyncio.sleep(10) + + monkeypatch.setattr(projection, 'run_blocking', stuck) + started = time.monotonic() + await cache.persist_pending() + assert time.monotonic() - started < 0.2 + assert len(cache.marker_retry) == 128 + # The independently bounded task also terminates, leaving no socket waiter. + await cache.marker_task + + +@pytest.mark.asyncio +async def test_overflow_inventory_is_bounded_and_reports_failed_handoff(world, monkeypatch): + cache = projection.PusherAudioProjection('u') + monkeypatch.setattr(projection, 'MAX_MARKER_RETRIES', projection.MAX_CONVERSATIONS + 2) + projection.record_projection_rebuild.mark_many.side_effect = OSError() + assert await cache.defer_rebuild('one') + assert await cache.defer_rebuild('two') + assert not await cache.defer_rebuild('three') + assert set(cache.marker_retry) == {'one', 'two'} + await cache.persist_pending() + assert set(cache.marker_retry) == {'one', 'two'} + + +def test_confirmed_empty_cold_discovery_is_distinct_from_privacy_deferral(world, monkeypatch): + monkeypatch.setattr(full_projection, 'projection_owner_gone', lambda *_: False) + write = MagicMock(side_effect=AssertionError('empty inventory must not write the conversation')) + monkeypatch.setattr(projection.record_projection_rebuild, 'apply_projection', write) + assert full_projection.rebuild('u', 'c') == 'no_objects' + write.assert_not_called() + assert not world.writes + + +@pytest.mark.asyncio +@pytest.mark.parametrize('empty', [False, True]) +async def test_full_overflow_list_preserves_cached_close_debt_until_cleanup(world, monkeypatch, empty): + cache = projection.PusherAudioProjection('u') + cache.has_uploaded_paths = True + cache.close_debt.add('cached') + cache.marker_retry.update({str(i): i for i in range(128)}) + finish = ( + MagicMock(return_value=None) + if empty + else MagicMock(side_effect=projection.InvalidArgument('synthetic projection rejection')) + ) + monkeypatch.setattr(cache, 'finish', finish) + # Best-effort queue refusal cannot block close or discard the cached owner. + assert await cache.finish_pending('cached', set(), set()) is True + assert cache.close_debt == {'cached'} + await cache.persist_pending() + assert not cache.marker_retry and not cache.close_debt + assert any('cached' in call.args[1] for call in projection.record_projection_rebuild.mark_many.call_args_list) diff --git a/backend/tests/unit/test_record_projection_rebuild.py b/backend/tests/unit/test_record_projection_rebuild.py new file mode 100644 index 00000000000..d7ac69e3fe4 --- /dev/null +++ b/backend/tests/unit/test_record_projection_rebuild.py @@ -0,0 +1,461 @@ +"""Projection debt uses real privacy/CAS transactions; never seal or content writes.""" + +from copy import deepcopy +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from google.api_core.exceptions import FailedPrecondition + +from database import record_projection_rebuild as jobs +from database.conversations import AudioFilesChanged +from tests.unit.fixtures.strict_firestore_transaction import StrictFirestore, StrictFirestoreDocument +from utils import record_projection_rebuild as consumer +from utils.batch_pipeline.record_seal_budget import SweepBudget + +ROW = ('users', 'u', 'conversations', 'c') + + +@pytest.fixture +def client(): + return StrictFirestore( + { + ('users', 'u'): {'private_cloud_sync_enabled': True}, + ROW: { + 'audio_files': [], + 'private_cloud_sync_enabled': True, + 'content': 'untouched', + 'record_upload_sessions': {}, + 'record_audio_closed': False, + }, + } + ) + + +def test_rebuild_only_updates_playback_metadata(client): + before = deepcopy(client.rows[ROW]) + assert jobs.apply_projection( + 'u', 'c', {'audio_files': [{'id': 'new'}]}, expected_audio_files=[], firestore_client=client + ) + assert client.rows[ROW] == {**before, 'audio_files': [{'id': 'new'}]} + assert client.transactions[-1].updates == [(ROW, {'audio_files': [{'id': 'new'}]})] + + +@pytest.mark.parametrize( + 'fence', ['tombstone', 'deleted', 'redirect', 'discarded', 'missing', 'account', 'e2ee', 'sync'] +) +def test_privacy_refusal_never_recreates_or_updates_conversation(client, fence): + if fence == 'tombstone': + client.rows[('users', 'u', 'deleted_conversations', 'c')] = {} + elif fence == 'missing': + client.rows.pop(ROW) + elif fence == 'account': + client.rows[('account_deletions', 'u')] = {'wipe_status': 'completed'} + elif fence == 'e2ee': + client.rows[('users', 'u')]['data_protection_level'] = 'e2ee' + elif fence == 'sync': + client.rows[ROW]['private_cloud_sync_enabled'] = False + elif fence == 'redirect': + client.rows[ROW].update(deleted=True, sync_merged_into='target') + client.rows[('users', 'u', 'conversations', 'target')] = {} + else: + client.rows[ROW][fence] = True + before = deepcopy(client.rows) + assert jobs.apply_projection( + 'u', 'c', {'audio_files': [{'id': 'new'}]}, expected_audio_files=[], firestore_client=client + ) is (None if fence == 'e2ee' else False) + assert client.rows == before + + +def test_peer_projection_race_is_retained_for_retry(client): + client.rows[ROW]['audio_files'] = [{'id': 'peer'}] + with pytest.raises(AudioFilesChanged): + jobs.apply_projection( + 'u', 'c', {'audio_files': [{'id': 'new'}]}, expected_audio_files=[], firestore_client=client + ) + assert client.rows[ROW]['audio_files'] == [{'id': 'peer'}] + + +@pytest.mark.parametrize('result', [True, False, None, OSError('private path')]) +def test_consumer_clears_only_observed_marker_and_rotates_failed_work(monkeypatch, result): + client = MagicMock() + ref = MagicMock() + doc = SimpleNamespace( + id=jobs.marker_id('u', 'c'), + reference=ref, + update_time='observed', + to_dict=lambda: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': 'c'}, + ) + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = [doc] + rebuild = MagicMock(side_effect=result) if isinstance(result, Exception) else MagicMock(return_value=result) + monkeypatch.setattr(consumer, 'rebuild', rebuild) + report = consumer.run(budget=SweepBudget(60), firestore_client=client) + client.collection.assert_called_once_with(jobs.COLLECTION) + client.collection.return_value.order_by.assert_called_once_with('requested_at') + client.collection.return_value.order_by.return_value.limit.assert_called_once_with( + consumer.MAX_COLLECTION_MARKERS + 1 + ) + assert client.write_option.call_args_list[0].kwargs == {'last_update_time': 'observed'} + if result is True or result is False: + ref.delete.assert_called_once() + assert client.write_option.call_args.kwargs == {'last_update_time': ref.update.return_value.update_time} + else: + ref.delete.assert_not_called() + ref.update.assert_called_once() + assert 'private' not in str(report) and 'owner_id' not in str(report) + + +def test_consumer_never_deletes_concurrently_replaced_debt(monkeypatch): + client = MagicMock() + ref = MagicMock() + ref.delete.side_effect = FailedPrecondition('newer marker') + doc = SimpleNamespace( + id=jobs.marker_id('u', 'c'), + reference=ref, + update_time='old', + to_dict=lambda: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': 'c'}, + ) + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = [doc] + monkeypatch.setattr(consumer, 'rebuild', lambda *_: True) + assert consumer.run(budget=SweepBudget(60), firestore_client=client)['dispositions'] == {'changed': 1} + ref.update.assert_called_once() + + +@pytest.mark.parametrize('mode', ['off', 'plan']) +def test_kill_switch_and_preview_never_rebuild_or_write(monkeypatch, mode): + client = MagicMock() + doc = SimpleNamespace( + id=jobs.marker_id('u', 'c'), + reference=MagicMock(), + update_time='old', + to_dict=lambda: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': 'c'}, + ) + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = [doc] + rebuild = MagicMock(side_effect=AssertionError('must not rebuild')) + monkeypatch.setattr(consumer, 'rebuild', rebuild) + if mode == 'off': + monkeypatch.setenv('PUSHER_PROJECTION_DEBT_DURABLE_ENABLED', 'off') + consumer.run(mode='plan' if mode == 'plan' else 'apply', budget=SweepBudget(60), firestore_client=client) + rebuild.assert_not_called() + doc.reference.delete.assert_not_called() + doc.reference.update.assert_not_called() + if mode == 'off': + client.collection.assert_not_called() + + +def test_marker_is_content_free_and_coalesces_by_owner(): + client = MagicMock() + jobs.mark_due('u', 'c', firestore_client=client) + jobs.mark_due('u', 'c', firestore_client=client) + assert [call.args for call in client.collection.return_value.document.call_args_list] == [ + (jobs.marker_id('u', 'c'),) + ] * 2 + assert jobs.marker_id('u', 'c') != jobs.marker_id('u', 'other') + assert client.collection.call_args.args == (jobs.COLLECTION,) + assert set(client.collection.return_value.document.return_value.set.call_args.args[0]) == { + 'kind', + 'owner_id', + 'conversation_id', + 'requested_at', + } + + +def test_job_budget_outcome_does_not_disable_seal_pass(monkeypatch): + from config import record_seal_repair as config + from utils.batch_pipeline.record_seal_budget import BudgetExhausted + + monkeypatch.setattr(config, 'require_runtime', lambda: None) + monkeypatch.setattr(config, 'settings', lambda: (8, 270, 1500, 'apply')) + monkeypatch.setattr(consumer, 'run', MagicMock(side_effect=BudgetExhausted())) + assert consumer.run_job()['outcome'] == 'budget_exhausted' + + +def test_rebuild_updates_metadata_but_retains_marker_while_producer_is_pending(client): + client.rows[ROW]['record_upload_sessions'] = {'live': True} + assert ( + jobs.apply_projection( + 'u', 'c', {'audio_files': [{'id': 'tail'}]}, expected_audio_files=[], firestore_client=client + ) + is None + ) + assert client.rows[ROW]['audio_files'] == [{'id': 'tail'}] + assert client.rows[ROW]['record_upload_sessions'] == {'live': True} + assert client.rows[ROW]['record_audio_closed'] is False + + +def test_invalid_marker_rotates_without_rebuild_or_deletion(monkeypatch): + client = MagicMock() + doc = SimpleNamespace( + id='wrong', + reference=MagicMock(), + update_time='old', + to_dict=lambda: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': 'c'}, + ) + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = [doc] + rebuild = MagicMock() + monkeypatch.setattr(consumer, 'rebuild', rebuild) + assert consumer.run(budget=SweepBudget(60), firestore_client=client)['dispositions'] == {'invalid': 1} + doc.reference.update.assert_called_once() + doc.reference.delete.assert_not_called() + rebuild.assert_not_called() + + +@pytest.mark.parametrize('hold', ['migration', 'generation']) +def test_temporary_privacy_hold_retains_debt_until_it_clears(client, hold): + from database.batch_pipeline_privacy import privacy_owner_key + + if hold == 'migration': + client.rows[('users', 'u')]['migration_status'] = 'migrating' + else: + client.rows[('batch_pipeline_privacy', privacy_owner_key('u'))] = {'purging': True} + before = deepcopy(client.rows) + assert ( + jobs.apply_projection( + 'u', 'c', {'audio_files': [{'id': 'new'}]}, expected_audio_files=[], firestore_client=client + ) + is None + ) + assert client.rows == before + if hold == 'migration': + client.rows[('users', 'u')].pop('migration_status') + else: + client.rows.pop(('batch_pipeline_privacy', privacy_owner_key('u'))) + assert ( + jobs.apply_projection( + 'u', 'c', {'audio_files': [{'id': 'new'}]}, expected_audio_files=[], firestore_client=client + ) + is True + ) + + +def test_deadline_rotates_before_discovery_and_never_clears_marker(monkeypatch): + from utils.batch_pipeline.record_seal_budget import BudgetExhausted + + client = MagicMock() + doc = SimpleNamespace( + id=jobs.marker_id('u', 'c'), + reference=MagicMock(), + update_time='old', + to_dict=lambda: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': 'c'}, + ) + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = [doc] + + def exhaust(*args): + doc.reference.update.assert_called_once() + raise BudgetExhausted() + + monkeypatch.setattr(consumer, 'rebuild', exhaust) + with pytest.raises(BudgetExhausted): + consumer.run(budget=SweepBudget(60), firestore_client=client) + doc.reference.delete.assert_not_called() + + +@pytest.mark.parametrize('age_days,over_capacity,retired', [(0, False, False), (8, False, True), (0, True, True)]) +def test_confirmed_empty_markers_retire_by_creation_age_or_collection_pressure( + monkeypatch, age_days, over_capacity, retired +): + from datetime import datetime, timedelta, timezone + + client = MagicMock() + doc = SimpleNamespace( + id=jobs.marker_id('u', 'c'), + reference=MagicMock(), + update_time='observed', + create_time=datetime.now(timezone.utc) - timedelta(days=age_days), + to_dict=lambda: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': 'c'}, + ) + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = [doc] + monkeypatch.setattr(consumer, 'MAX_COLLECTION_MARKERS', 0 if over_capacity else 10000) + monkeypatch.setattr(consumer, 'rebuild', lambda *_: 'no_objects') + retire = MagicMock(return_value='retired_no_objects') + monkeypatch.setattr(jobs, 'retire_empty', retire) + report = consumer.run(budget=SweepBudget(60), firestore_client=client) + doc.reference.delete.assert_not_called() + assert bool(retire.call_count) is retired + if retired: + assert retire.call_args.kwargs == { + 'observed_update_time': doc.reference.update.return_value.update_time, + 'firestore_client': client, + } + assert report['dispositions'] == {'retired_no_objects' if retired else 'deferred': 1} + assert report['gauges']['marker_count'] == 1 + assert report['gauges']['oldest_age_seconds'] >= age_days * 86400 + assert report['gauges']['at_capacity'] is over_capacity + + +def test_collection_pressure_never_discards_privacy_hold_or_incomplete_discovery(monkeypatch): + from datetime import datetime, timedelta, timezone + + client = MagicMock() + doc = SimpleNamespace( + id=jobs.marker_id('u', 'c'), + reference=MagicMock(), + update_time='observed', + create_time=datetime.now(timezone.utc) - timedelta(days=30), + to_dict=lambda: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': 'c'}, + ) + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = [doc] + monkeypatch.setattr(consumer, 'MAX_COLLECTION_MARKERS', 0) + monkeypatch.setattr(consumer, 'rebuild', lambda *_: None) + assert consumer.run(budget=SweepBudget(60), firestore_client=client)['dispositions'] == {'deferred': 1} + doc.reference.delete.assert_not_called() + + +def test_consumer_caps_work_independently_of_collection_scan(monkeypatch): + client = MagicMock() + docs = [ + SimpleNamespace( + id=jobs.marker_id('u', str(i)), + reference=MagicMock(), + update_time='observed', + to_dict=lambda i=i: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': str(i)}, + ) + for i in range(3) + ] + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = docs + monkeypatch.setattr(consumer, 'MAX_PER_PASS', 2) + rebuild = MagicMock(return_value=True) + monkeypatch.setattr(consumer, 'rebuild', rebuild) + assert consumer.run(budget=SweepBudget(60), firestore_client=client)['dispositions'] == {'rebuilt': 2} + assert rebuild.call_count == 2 + docs[-1].reference.update.assert_not_called() + + +def test_socket_cleanup_is_one_content_free_bounded_batch(): + client = MagicMock() + jobs.mark_many('u', ['one', 'two', 'one'], firestore_client=client) + batch = client.batch.return_value + assert batch.set.call_count == 2 + for call in batch.set.call_args_list: + assert set(call.args[1]) == {'kind', 'owner_id', 'conversation_id', 'requested_at'} + batch.commit.assert_called_once_with(timeout=1, retry=None) + with pytest.raises(ValueError): + jobs.mark_many('u', [str(i) for i in range(jobs.MAX_MARKERS_PER_BATCH + 1)], firestore_client=client) + assert batch.commit.call_count == 1 + + +@pytest.fixture +def retirement_store(client, monkeypatch): + marker = (jobs.COLLECTION, jobs.marker_id('u', 'c')) + client.rows[marker] = {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': 'c'} + get = StrictFirestoreDocument.get + version = ['rotated'] + + def snapshot(ref, *args, **kwargs): + result = get(ref, *args, **kwargs) + if ref.path == marker: + result.update_time = version[0] + return result + + # Only supply immutable snapshot metadata. The fixture still enforces all + # transactional reads-before-delete; real version fencing is emulator-tested. + monkeypatch.setattr(StrictFirestoreDocument, 'get', snapshot) + return client, marker, version + + +@pytest.mark.parametrize( + 'sessions', [{'live': True}, {'abandoned': {'abandoned_at': 0}}, None, True, [], 'missing', {}] +) +def test_empty_retirement_requires_explicit_clear_producer_map(retirement_store, sessions): + client, marker, _ = retirement_store + if sessions == 'missing': + client.rows[ROW].pop('record_upload_sessions') + else: + client.rows[ROW]['record_upload_sessions'] = sessions + before = deepcopy(client.rows) + result = jobs.retire_empty('u', 'c', observed_update_time='rotated', firestore_client=client) + clear = isinstance(sessions, dict) and not sessions + assert result == ('retired_no_objects' if clear else 'producer_fenced') + assert client.rows == ({key: row for key, row in before.items() if key != marker} if clear else before) + assert client.transactions[-1].updates == [] + assert client.transactions[-1].deletes == ([marker] if clear else []) + + +@pytest.mark.parametrize('race', ['replacement', 'missing_marker', 'missing_owner']) +def test_empty_retirement_refuses_changed_marker_or_unknown_owner(retirement_store, race): + client, marker, version = retirement_store + if race == 'replacement': + version[0] = 'newer' + elif race == 'missing_marker': + client.rows.pop(marker) + else: + client.rows.pop(ROW) + before = deepcopy(client.rows) + assert jobs.retire_empty('u', 'c', observed_update_time='rotated', firestore_client=client) == ( + 'producer_fenced' if race == 'missing_owner' else 'changed' + ) + assert client.rows == before + assert not client.transactions[-1].deletes + + +@pytest.mark.parametrize('pressure', [False, True]) +def test_old_empty_marker_keeps_producer_minted_after_discovery(retirement_store, monkeypatch, pressure): + from datetime import datetime, timedelta, timezone + + store, marker, _ = retirement_store + client = MagicMock() + doc = SimpleNamespace( + id=marker[-1], + reference=MagicMock(), + update_time='before_rotation', + create_time=datetime.now(timezone.utc) - timedelta(days=8), + to_dict=lambda: store.rows[marker], + ) + doc.reference.update.return_value.update_time = 'rotated' + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = [doc] + monkeypatch.setattr(consumer, 'MAX_COLLECTION_MARKERS', 0 if pressure else 10000) + + def discover_then_mint(*args): + store.rows[ROW]['record_upload_sessions'] = {'new_tail': True} + return 'no_objects' + + monkeypatch.setattr(consumer, 'rebuild', discover_then_mint) + retire = jobs.retire_empty + monkeypatch.setattr( + jobs, + 'retire_empty', + lambda uid, cid, **kw: retire( + uid, cid, observed_update_time=kw['observed_update_time'], firestore_client=store + ), + ) + report = consumer.run(budget=SweepBudget(60), firestore_client=client) + assert report['dispositions'] == {'producer_fenced': 1} + assert report['gauges']['pressure_blocked'] is pressure + assert report['gauges']['pressure_fenced_count'] == (1 if pressure else 0) + assert marker in store.rows and not store.transactions[-1].deletes + doc.reference.delete.assert_not_called() + doc.reference.update.assert_called_once() # Deferred work is already rotated. + + +@pytest.mark.parametrize('clear_one', [False, True]) +def test_pressure_alerts_when_all_empty_candidates_are_fenced(monkeypatch, clear_one): + from datetime import datetime, timedelta, timezone + + client = MagicMock() + docs = [ + SimpleNamespace( + id=jobs.marker_id('u', str(i)), + reference=MagicMock(), + update_time='old', + create_time=datetime.now(timezone.utc) - timedelta(days=8), + to_dict=lambda i=i: {'kind': jobs.KIND, 'owner_id': 'u', 'conversation_id': str(i)}, + ) + for i in range(3) + ] + client.collection.return_value.order_by.return_value.limit.return_value.stream.return_value = docs + monkeypatch.setattr(consumer, 'MAX_COLLECTION_MARKERS', 2) + monkeypatch.setattr(consumer, 'rebuild', lambda *_: 'no_objects') + monkeypatch.setattr( + jobs, + 'retire_empty', + MagicMock( + side_effect=['producer_fenced', 'producer_fenced', 'retired_no_objects' if clear_one else 'producer_fenced'] + ), + ) + report = consumer.run(budget=SweepBudget(60), firestore_client=client) + assert report['outcome'] == 'capacity_exceeded' + assert report['gauges']['pressure_blocked'] is (not clear_one) + assert report['gauges']['pressure_fenced_count'] == (2 if clear_one else 3) + assert consumer.PRESSURE_BLOCKED.collect()[0].samples[0].value == (0 if clear_one else 1) + for doc in docs: + doc.reference.delete.assert_not_called() # Only the fenced transaction may retire empty debt. diff --git a/backend/tests/unit/test_record_seal_cluster_repair.py b/backend/tests/unit/test_record_seal_cluster_repair.py index 7e8bba6fc51..a270c18c218 100644 --- a/backend/tests/unit/test_record_seal_cluster_repair.py +++ b/backend/tests/unit/test_record_seal_cluster_repair.py @@ -163,7 +163,10 @@ def test_native_and_dependency_diagnostics_never_reach_cli_output(tmp_path): 'raise SystemExit(entry["main"]())\n' ) result = subprocess.run( - [sys.executable, str(script)], capture_output=True, text=True, env={**os.environ, 'OMI_LOG_FORMAT': 'json'} + [sys.executable, str(script)], + capture_output=True, + text=True, + env={**os.environ, 'OMI_LOG_FORMAT': 'json', 'PUSHER_PROJECTION_DEBT_DURABLE_ENABLED': 'off'}, ) assert result.returncode == 0 and not result.stderr assert private not in result.stdout diff --git a/backend/tests/unit/test_record_seal_cluster_repair_deploy.py b/backend/tests/unit/test_record_seal_cluster_repair_deploy.py index c6eb1ca722c..e8cb69a8cad 100644 --- a/backend/tests/unit/test_record_seal_cluster_repair_deploy.py +++ b/backend/tests/unit/test_record_seal_cluster_repair_deploy.py @@ -200,3 +200,39 @@ def test_jobs_discovery_rollback_reaches_only_repair_cron(): ][0] env = {entry['name']: entry.get('value') for entry in container['env']} assert env['RECORD_SEAL_REPAIR_JOBS'] == 'off' + + +def test_projection_debt_kill_switch_reaches_cron_without_disabling_seal_repair(): + for setting in ('true', 'false'): + docs = documents('--set', f'recordSealRepair.projectionDebtDurableEnabled={setting}') + pod = docs[('CronJob', 'record-seal-repair')]['spec']['jobTemplate']['spec']['template']['spec'] + env = {e['name']: e.get('value') for e in pod['containers'][0]['env']} + assert env['PUSHER_PROJECTION_DEBT_DURABLE_ENABLED'] == setting + assert env['RECORD_SEAL_REPAIR_ENABLED'] == 'true' + + +def test_pusher_flag_can_arrive_unshadowed_from_existing_runtime_configmap(): + backend = Path(__file__).resolve().parents[2] + result = subprocess.run( + [ + 'helm', + 'template', + 'projection-debt-test', + str(backend / 'charts/pusher'), + '-f', + str(backend / 'charts/pusher/prod_omi_pusher_values.yaml'), + '--set-string', + 'image.tag=synthetic', + '--set-string', + 'image.digest=', + '--set-string', + 'runtimeConfigMapName=projection-debt-test-config', + ], + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stderr + deployment = next(doc for doc in yaml.safe_load_all(result.stdout) if doc and doc['kind'] == 'Deployment') + container = deployment['spec']['template']['spec']['containers'][0] + assert {'configMapRef': {'name': 'projection-debt-test-config'}} in container['envFrom'] + assert not any(item['name'] == 'PUSHER_PROJECTION_DEBT_DURABLE_ENABLED' for item in container.get('env', [])) diff --git a/backend/utils/batch_pipeline/record_projection_rebuild.py b/backend/utils/batch_pipeline/record_projection_rebuild.py new file mode 100644 index 00000000000..34689a22f15 --- /dev/null +++ b/backend/utils/batch_pipeline/record_projection_rebuild.py @@ -0,0 +1,80 @@ +"""Socket-independent playback metadata discovery; no audio decoding or custody.""" + +from typing import Literal + +from config import audio_archive +from database import conversations as conversations_db, conversation_tombstones, record_projection_rebuild as jobs +from database._client import get_firestore_client +from utils.other import storage, raw_audio_read + + +def projection_owner_gone(uid: str, conversation_id: str) -> bool: + """Deletion intent and hard absence; the cold commit rechecks privacy atomically.""" + if conversation_tombstones.is_deleted(uid, conversation_id): + return True + doc = ( + get_firestore_client() + .collection('users') + .document(uid) + .collection('conversations') + .document(conversation_id) + .get() + ) + return not doc.exists + + +def projected_chunks(uid, cid, chunks): + if not audio_archive.admitted(uid, 'READ'): + return chunks + packs = storage.list_packs(storage.get_private_cloud_sync_bucket(), uid, cid) + return sorted([*chunks, *raw_audio_read.uncovered_raw_packs(packs, chunks)], key=lambda chunk: chunk['timestamp']) + + +def discover(uid, cid, *, strict_raw=False, fallback=None): + """Snapshot before LIST; share the full rebuild with pusher's live cache.""" + owner = conversations_db.get_conversation_raw_snapshot(uid, cid, field_paths=['audio_files', 'deleted']) + if owner is None or owner.get('deleted'): + return None + expected = owner.get('audio_files') or [] + chunks = storage.list_audio_chunks(uid, cid, source_identity=True, include_archive=False) + raw_complete = True + try: + listed = projected_chunks(uid, cid, chunks) + except Exception: + if strict_raw: + raise + raw_complete, listed = False, chunks + if fallback is not None: + fallback( + component='pusher', + from_mode='raw_audio', + to_mode='encoded_audio_projection', + reason='other', + outcome='degraded', + ) + files = conversations_db.create_audio_files_from_chunks(uid, cid, listed_chunks=listed) + return expected, chunks, files, raw_complete + + +def rebuild(uid: str, cid: str) -> bool | Literal['no_objects'] | None: + """Full metadata rebuild with deletion/privacy and prior projection CAS. + + None retains debt for incomplete discovery. No PCM or audio object is read, + created, rewritten or deleted; raw coverage uses the existing pack index. + """ + if projection_owner_gone(uid, cid): + return False + for _ in range(2): + discovery = discover(uid, cid, strict_raw=True) + if discovery is None: + return False + expected, _, files, _ = discovery + if not files: + return 'no_objects' + try: + return jobs.apply_projection( + uid, cid, {'audio_files': [af.model_dump() for af in files]}, expected_audio_files=expected + ) + except conversations_db.AudioFilesChanged: + continue + raise RuntimeError('audio projection refresh exhausted') diff --git a/backend/utils/pusher_audio_projection.py b/backend/utils/pusher_audio_projection.py index 1995da0a144..fae57e3e864 100644 --- a/backend/utils/pusher_audio_projection.py +++ b/backend/utils/pusher_audio_projection.py @@ -1,5 +1,6 @@ """Keep the live playback projection fresh without rediscovering its prefix per write.""" +import asyncio import logging import os import time @@ -10,12 +11,16 @@ from prometheus_client import Counter from config import audio_archive -from database import conversations as conversations_db, conversation_tombstones -from database._client import get_firestore_client +from config.batch_pipeline_projection_rebuild import enabled as durable_debt_enabled +from database import conversations as conversations_db, record_projection_rebuild from database.audio_chunk_metadata import audio_chunk_entry -from utils.other import storage, raw_audio_read +from utils.other import storage +from utils.batch_pipeline.record_projection_rebuild import ( + projected_chunks, + discover, +) from utils.observability.fallback import record_fallback -from utils.executors import run_blocking, storage_executor +from utils.executors import run_blocking, storage_executor, db_executor, start_background_task from utils.other.audio_outbox_metrics import record_failure from utils.speaker_learning_jobs import schedule_person_voice_learning_retry @@ -25,6 +30,10 @@ REFRESH_SECONDS = 300 MAX_CONVERSATIONS = 128 MAX_CHUNKS = 2048 +MAX_MARKER_RETRIES = record_projection_rebuild.MAX_MARKERS_PER_BATCH +MARKER_CLOSE_SECONDS = 3.0 +MARKER_ATTEMPT_SECONDS = 1.25 +MARKER_WRITES = Counter('omi_pusher_projection_debt_total', 'Best-effort projection marker handoffs', ['outcome']) def enabled() -> bool: @@ -38,26 +47,6 @@ class Inventory: listed_at: float -def projection_owner_gone(uid: str, conversation_id: str) -> bool: - """Deletion-intent read for a projection NotFound: owner gone, or live? - - Same tombstone read as the custody write gate (utils/other/storage.py); - a missing conversation doc covers hard-deleted rows without a tombstone. - Runs only on the NotFound arm, so live conversations pay no extra reads. - """ - if conversation_tombstones.is_deleted(uid, conversation_id): - return True - doc = ( - get_firestore_client() - .collection('users') - .document(uid) - .collection('conversations') - .document(conversation_id) - .get() - ) - return not doc.exists - - def get_uploaded_audio_chunks(uid: str, conversation_id: str, paths: List[str]) -> List[Dict[str, Any]]: """Read exact encoded objects; never discover siblings or raw archive packs. @@ -103,18 +92,15 @@ def __init__(self, uid: str): self.uid = uid self.has_uploaded_paths = False self.close_debt: set[str] = set() + # Separate metadata-only overflow inventory; never retain PCM receipts + # or gate recording completion on optional marker persistence. + self.marker_retry: dict[str, int] = {} + self.marker_sequence = 0 + self.marker_task: asyncio.Task | None = None self.finished: set[str] = set() self.close_retries: dict[str, int] = {} self.inventory: dict[str, Inventory | None] = {} - def projected_chunks(self, conversation_id: str, chunks: list[dict[str, Any]]) -> list[dict[str, Any]]: - if not audio_archive.admitted(self.uid, 'READ'): - return chunks - packs = storage.list_packs(storage.get_private_cloud_sync_bucket(), self.uid, conversation_id) - return sorted( - [*chunks, *raw_audio_read.uncovered_raw_packs(packs, chunks)], key=lambda chunk: chunk['timestamp'] - ) - def refresh(self, conversation_id: str, paths: list[str] | None = None, *, force: bool = False) -> bool | None: state = self.inventory.pop(conversation_id, None) self.finished.discard(conversation_id) @@ -154,39 +140,15 @@ def refresh(self, conversation_id: str, paths: list[str] | None = None, *, force raw_complete = True if chunks is None: if enabled(): - # Snapshot before discovery: a full rebuild must not - # overwrite a projection another pod publishes during LIST. - owner = conversations_db.get_conversation_raw_snapshot( - self.uid, conversation_id, field_paths=['audio_files', 'deleted'] + discovery = discover( + self.uid, conversation_id, strict_raw=force and durable_debt_enabled(), fallback=record_fallback ) - if owner is None or owner.get('deleted'): + if discovery is None: self.inventory.pop(conversation_id, None) return False - expected = owner.get('audio_files') or [] + expected, chunks, files, raw_complete = discovery else: expected = None - if enabled(): - chunks = storage.list_audio_chunks( - self.uid, conversation_id, source_identity=True, include_archive=False - ) - try: - listed = self.projected_chunks(conversation_id, chunks) - except Exception: - # Preserve main's full-list fallback on a broken raw - # index, but never cache that incomplete discovery. - raw_complete = False - listed = chunks - record_fallback( - component='pusher', - from_mode='raw_audio', - to_mode='encoded_audio_projection', - reason='other', - outcome='degraded', - ) - files = conversations_db.create_audio_files_from_chunks( - self.uid, conversation_id, listed_chunks=listed - ) - else: files = conversations_db.create_audio_files_from_chunks(self.uid, conversation_id) listed_at = time.monotonic() mode = 'full' @@ -194,7 +156,7 @@ def refresh(self, conversation_id: str, paths: list[str] | None = None, *, force # Raw fragments can shrink or split as encoded coverage grows; # re-derive them, never append yesterday's projected fragments. try: - listed = self.projected_chunks(conversation_id, chunks) + listed = projected_chunks(self.uid, conversation_id, chunks) except Exception: # An incremental encoded view is not a full fallback. # Rediscover encoded siblings before matching main's @@ -257,6 +219,8 @@ def finish(self, conversation_id: str) -> bool | None: return None # Rollback/unknown writer retains per-flush-only behavior. result = self.refresh(conversation_id, force=True) self.inventory.pop(conversation_id, None) + if result is None and durable_debt_enabled(): + return None # Empty discovery cannot discharge the close obligation. self.close_debt.discard(conversation_id) if len(self.finished) < MAX_CONVERSATIONS: self.finished.add(conversation_id) @@ -278,9 +242,11 @@ async def finish_pending(self, conv_id: str, deleted_conversations: set[str], id self.close_retries[conv_id] = attempts + 1 logger.warning('Close audio projection rebuild deferred exception_type=%s', type(exc).__name__) return False + queued = await self.defer_rebuild(conv_id) self.close_retries.pop(conv_id, None) self.inventory.pop(conv_id, None) - self.close_debt.discard(conv_id) + if queued: + self.close_debt.discard(conv_id) if len(self.finished) < MAX_CONVERSATIONS: self.finished.add(conv_id) record_failure('projection', reason='invalid_argument' if isinstance(exc, InvalidArgument) else 'transient') @@ -295,6 +261,11 @@ async def finish_pending(self, conv_id: str, deleted_conversations: set[str], id # The per-flush projection and raw receipt remain authoritative; # optional rediscovery failure cannot turn a clean drain into loss. return True + if applied is None and self.has_uploaded_paths and conv_id in self.close_debt and durable_debt_enabled(): + if await self.defer_rebuild(conv_id): + self.close_debt.discard(conv_id) + if len(self.finished) < MAX_CONVERSATIONS: + self.finished.add(conv_id) self.close_retries.pop(conv_id, None) if applied: schedule_person_voice_learning_retry(self.uid, conv_id) @@ -303,3 +274,66 @@ async def finish_pending(self, conv_id: str, deleted_conversations: set[str], id elif applied is False: deleted_conversations.add(conv_id) return True + + async def defer_rebuild(self, conv_id: str) -> bool: + """Queue optional metadata without awaiting Firestore on the close path.""" + if not durable_debt_enabled(): + return True + if conv_id not in self.marker_retry and len(self.marker_retry) >= MAX_MARKER_RETRIES - MAX_CONVERSATIONS: + MARKER_WRITES.labels(outcome='marker_retry_overflow').inc() + logger.error('Projection debt retry inventory full; optional marker handoff refused') + return False + self.marker_sequence += 1 + self.marker_retry[conv_id] = self.marker_sequence + self._start_marker_task() + return True + + def _start_marker_task(self) -> None: + if self.marker_retry and (self.marker_task is None or self.marker_task.done()): + self.marker_task = start_background_task(self._persist_markers(), name='projection-debt-handoff') + + async def _persist_markers(self) -> None: + # At most two batched RPCs, including executor queue time. A late SDK + # commit is metadata-only, idempotent and has its own one-second timeout. + for _ in range(2): + pending = dict(self.marker_retry) + if not pending: + return + try: + async with asyncio.timeout(MARKER_ATTEMPT_SECONDS): + await run_blocking(db_executor, record_projection_rebuild.mark_many, self.uid, set(pending)) + except Exception as exc: + MARKER_WRITES.labels(outcome='marker_write_failed').inc() + logger.warning('Projection marker_write_failed exception_type=%s', type(exc).__name__) + continue + for cid, sequence in pending.items(): + if self.marker_retry.get(cid) == sequence: + self.marker_retry.pop(cid, None) + self.close_debt.discard(cid) + MARKER_WRITES.labels(outcome='persisted').inc(len(pending)) + + async def persist_pending(self) -> None: + """Total socket-cleanup budget, independent of the number of debts.""" + if not durable_debt_enabled(): + self.close_debt.clear() + self.marker_retry.clear() + return + for cid in self.close_debt: + if cid not in self.marker_retry: + self.marker_sequence += 1 + self.marker_retry[cid] = self.marker_sequence + self._start_marker_task() + if self.marker_task is None: + return + try: + async with asyncio.timeout(MARKER_CLOSE_SECONDS): + for attempt in range(2): + await asyncio.shield(self.marker_task) + if not self.marker_retry or attempt == 1: + break + self._start_marker_task() + except TimeoutError: + MARKER_WRITES.labels(outcome='marker_write_failed').inc() + logger.warning('Projection marker_write_failed cleanup_budget_exhausted') + # Background work is bounded independently; socket close never waits + # for a second round after the total cleanup budget expires. diff --git a/backend/utils/record_projection_rebuild.py b/backend/utils/record_projection_rebuild.py new file mode 100644 index 00000000000..b116cf078a7 --- /dev/null +++ b/backend/utils/record_projection_rebuild.py @@ -0,0 +1,127 @@ +"""Bounded playback-metadata maintenance pass for the existing repair CronJob.""" + +from collections import Counter +from datetime import datetime, timezone + +from prometheus_client import Gauge + +from google.api_core.exceptions import FailedPrecondition, NotFound + +from config.batch_pipeline_projection_rebuild import enabled as durable_debt_enabled +from database import record_projection_rebuild as jobs +from database._client import get_firestore_client +from utils.batch_pipeline.record_projection_rebuild import rebuild + +MAX_PER_PASS = 128 +MAX_COLLECTION_MARKERS = 10000 +NO_OBJECTS_MAX_AGE_SECONDS = 7 * 86400 +BACKLOG = Gauge('omi_projection_rebuild_marker_count', 'Bounded marker count; saturated at collection limit plus one') +OLDEST = Gauge('omi_projection_rebuild_oldest_age_seconds', 'Oldest marker age in the bounded collection scan') +AT_CAPACITY = Gauge('omi_projection_rebuild_at_capacity', 'Marker collection exceeds its maintenance limit') +PRESSURE_BLOCKED = Gauge( + 'omi_projection_rebuild_pressure_blocked', + 'Observed empty-marker retirement blocked by producer fences under pressure', +) + + +def run(*, mode='apply', budget, firestore_client=None): + if mode not in {'apply', 'plan'}: + raise ValueError('invalid projection repair mode') + if not durable_debt_enabled(): + return {'event': 'record_projection_rebuild', 'outcome': 'disabled'} + client = firestore_client if firestore_client is not None else get_firestore_client() + counts = Counter() + # One ordered field uses Firestore's automatic single-field index. Rotate + # refused work to the tail so chronic failures cannot starve later markers. + query = client.collection(jobs.COLLECTION).order_by('requested_at').limit(MAX_COLLECTION_MARKERS + 1) + page = list(query.stream(timeout=budget.timeout(), retry=None)) + now = datetime.now(timezone.utc) + ages = { + doc.id: max(0.0, (now - created).total_seconds()) + for doc in page + if isinstance(created := getattr(doc, 'create_time', None), datetime) + } + at_capacity = len(page) > MAX_COLLECTION_MARKERS + gauges = { + 'marker_count': len(page), + 'oldest_age_seconds': max(ages.values(), default=0), + 'at_capacity': at_capacity, + } + BACKLOG.set(len(page)) + OLDEST.set(gauges['oldest_age_seconds']) + AT_CAPACITY.set(at_capacity) + PRESSURE_BLOCKED.set(0) + for doc in page[:MAX_PER_PASS]: + budget.check() + data = doc.to_dict() or {} + uid, cid = data.get('owner_id'), data.get('conversation_id') + try: + # Rotate before identity validation AND discovery: malformed rows or + # a stuck listing cannot monopolize every bounded page. + attempt = None + if mode != 'plan': + attempt = doc.reference.update( + {'requested_at': datetime.now(timezone.utc)}, + option=client.write_option(last_update_time=doc.update_time), + timeout=budget.timeout(), + retry=None, + ) + if ( + not isinstance(uid, str) + or not isinstance(cid, str) + or data.get('kind') != jobs.KIND + or doc.id != jobs.marker_id(uid, cid) + ): + counts['invalid'] += 1 + continue + if mode == 'plan': + counts['due'] += 1 + continue + if attempt is None: + raise RuntimeError('projection repair attempt unavailable') + result = rebuild(uid, cid) + budget.check() + retire_empty = result == 'no_objects' and (ages.get(doc.id, 0) >= NO_OBJECTS_MAX_AGE_SECONDS or at_capacity) + if retire_empty: + counts[ + jobs.retire_empty(uid, cid, observed_update_time=attempt.update_time, firestore_client=client) + ] += 1 + elif result is True or result is False: + doc.reference.delete( + option=client.write_option(last_update_time=attempt.update_time), + timeout=budget.timeout(), + retry=None, + ) + counts['rebuilt' if result else 'owner_gone'] += 1 + else: + counts['deferred'] += 1 + except (FailedPrecondition, NotFound): + counts['changed'] += 1 # Never clear a newer socket's debt. + except Exception: + budget.check() # Let the hard deadline unwind the enclosing pass. + counts['error'] += 1 + gauges['pressure_fenced_count'] = counts['producer_fenced'] if at_capacity else 0 + gauges['pressure_blocked'] = bool(at_capacity and counts['producer_fenced'] and not counts['retired_no_objects']) + PRESSURE_BLOCKED.set(gauges['pressure_blocked']) + return { + 'event': 'record_projection_rebuild', + 'outcome': 'error' if counts['error'] else 'capacity_exceeded' if at_capacity else 'ok', + 'gauges': gauges, + 'dispositions': dict(counts), + } + + +def run_job(): + from config import record_seal_repair as config + from utils.batch_pipeline.record_seal_budget import SweepBudget, BudgetExhausted + + if not durable_debt_enabled() or not config.enabled(): + return {'event': 'record_projection_rebuild', 'outcome': 'disabled'} + config.require_runtime() + mode = config.settings()[3] + budget = SweepBudget(60) + try: + with budget.interrupt_io(): + return run(mode=mode, budget=budget) + except BudgetExhausted: + return {'event': 'record_projection_rebuild', 'outcome': 'budget_exhausted'} diff --git a/config/feature-flags.yaml b/config/feature-flags.yaml index 5e620f7746f..f214f2a67ae 100644 --- a/config/feature-flags.yaml +++ b/config/feature-flags.yaml @@ -1,4 +1,5 @@ flags: + - {"key":"PUSHER_PROJECTION_DEBT_DURABLE_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; persist exhausted or evicted playback projection debt and rebuild it in the record repair CronJob","fail":"open","owner":"uicleanup / omi-lifecycle","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"0, false or off disables both marker handoff and repair consumption. Disable recordSealRepair.projectionDebtDurableEnabled first, then the Pusher runtime ConfigMap flag. Consumer requires the existing CronJob to be enabled and unsuspended; no activation here. Content-free record_projection_rebuild_jobs markers remain inert. No hot-path writes, seal authority, token, PCM or audio-object changes. See backend/docs/listen_pusher_pipeline/changes/2026-11-06-durable-projection-debt.md."} - {"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."}