diff --git a/apps/api/src/jobs/backup_restore_legacy_backfill_job.py b/apps/api/src/jobs/backup_restore_legacy_backfill_job.py index ceeebf62f..453b4cf0d 100644 --- a/apps/api/src/jobs/backup_restore_legacy_backfill_job.py +++ b/apps/api/src/jobs/backup_restore_legacy_backfill_job.py @@ -43,6 +43,7 @@ BACKFILL_RESULT_SCHEMA_VERSION = "backup_restore_legacy_backfill_result_v1" BACKFILL_CURSOR_AGENT_ID = "backup_restore_legacy_backfill_cursor" BACKFILL_ITEM_AGENT_ID = "backup_restore_legacy_backfill" BACKFILL_IDEMPOTENCY_CHANNEL = "backup_restore_backfill" +BACKFILL_DB_CONTRACT_REPAIR_GENERATION = "cost_usd_state_bind_v1" BACKUP_RESTORE_EVENT_TYPE = "backup_restore_escrow_signal" BACKUP_RESTORE_LANE = "backup_restore_escrow_triage" PROJECTION_SCHEMA_VERSION = "telegram_agent99_dispatch_receipt_projection_v1" @@ -582,6 +583,58 @@ _ITEM_STATE_COUNTS_SQL = text( """ ) +_REQUEUE_REPAIRED_DBAPI_FAILURES_SQL = text( + """ + WITH repairable AS ( + SELECT item.run_id + FROM awooop_run_state item + WHERE item.project_id = :project_id + AND item.agent_id = :agent_id + AND item.state = 'failed' + AND item.error_code = 'E-BACKFILL-EXHAUSTED' + AND COALESCE(NULLIF(item.error_detail, ''), '{}')::jsonb #>> + '{last_result,reason}' = 'processor_error:DBAPIError' + AND COALESCE(NULLIF(item.error_detail, ''), '{}')::jsonb #>> + '{item,runtime_execution_authorized}' = 'false' + AND COALESCE( + COALESCE(NULLIF(item.error_detail, ''), '{}')::jsonb #>> + '{item,recovery_generation}', + '' + ) <> :repair_generation + AND item.worker_id IS NULL + AND item.lease_until IS NULL + ORDER BY item.created_at ASC, item.run_id ASC + FOR UPDATE SKIP LOCKED + LIMIT :limit + ) + UPDATE awooop_run_state item + SET state = 'waiting_tool', + attempt_count = 0, + max_attempts = GREATEST(item.max_attempts, 5), + worker_id = NULL, + lease_until = NULL, + heartbeat_at = NOW(), + completed_at = NULL, + error_code = NULL, + error_detail = JSONB_SET( + JSONB_SET( + COALESCE(NULLIF(item.error_detail, ''), '{}')::jsonb, + '{item,recovery_generation}', + TO_JSONB(CAST(:repair_generation AS TEXT)), + TRUE + ), + '{item,recovery_reason}', + TO_JSONB(CAST('verified_database_contract_repair' AS TEXT)), + TRUE + ) + FROM repairable + WHERE item.project_id = :project_id + AND item.run_id = repairable.run_id + AND item.agent_id = :agent_id + RETURNING item.run_id +""" +) + @dataclass(frozen=True) class LegacyBackupClaim: @@ -1187,6 +1240,44 @@ async def _claim_legacy_backup_items( return claims +async def _requeue_repaired_dbapi_failures( + *, + project_id: str, + limit: int, +) -> int: + """Requeue one bounded historical DB-contract failure generation. + + The original attempts predate the explicit ``cost_usd`` insert and typed + state binds. Only those exact, no-runtime-write failures are eligible, + and the recovery generation is persisted inside the stable item envelope + so the same source row cannot be reset repeatedly. + """ + + bounded_limit = max(1, min(int(limit), MAX_SCOPE_ROWS)) + async with get_db_context(project_id) as db: + result = await db.execute( + _REQUEUE_REPAIRED_DBAPI_FAILURES_SQL, + { + "project_id": project_id, + "agent_id": BACKFILL_ITEM_AGENT_ID, + "repair_generation": BACKFILL_DB_CONTRACT_REPAIR_GENERATION, + "limit": bounded_limit, + }, + ) + rows = result.mappings().all() + requeued = len(rows) + if requeued: + logger.info( + "backup_restore_legacy_backfill_db_contract_failures_requeued", + project_id=project_id, + repair_generation=BACKFILL_DB_CONTRACT_REPAIR_GENERATION, + requeued=requeued, + runtime_execution_authorized=False, + telegram_dispatch_performed=False, + ) + return requeued + + async def _load_legacy_backup_card( *, project_id: str, @@ -1876,6 +1967,8 @@ async def run_backup_restore_legacy_backfill_once( "failed": 0, "retried": 0, "lease_lost": 0, + "requeued_repaired_dbapi": 0, + "backfill_state_write_performed": False, "dispatch_total": 0, "source_total_at_start": None, "source_exhausted": False, @@ -1892,6 +1985,13 @@ async def run_backup_restore_legacy_backfill_once( "error": None, } try: + if processor is process_backup_restore_alertmanager_signal: + requeued = await _requeue_repaired_dbapi_failures( + project_id=normalized_project, + limit=max_rows, + ) + stats["requeued_repaired_dbapi"] = requeued + stats["backfill_state_write_performed"] = requeued > 0 reserved = await _reserve_legacy_backup_page( project_id=normalized_project, scan_limit=scan_limit, diff --git a/apps/api/tests/test_backup_restore_legacy_backfill_job.py b/apps/api/tests/test_backup_restore_legacy_backfill_job.py index 911819187..178c1e321 100644 --- a/apps/api/tests/test_backup_restore_legacy_backfill_job.py +++ b/apps/api/tests/test_backup_restore_legacy_backfill_job.py @@ -482,6 +482,51 @@ async def test_claim_sweeps_expired_max_attempt_before_selecting_retryable_work( assert statements == [job._EXHAUST_EXPIRED_CLAIMS_SQL, job._CLAIM_SQL] +@pytest.mark.asyncio +async def test_db_contract_repair_requeues_only_one_bounded_generation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + returned = [ + {"run_id": _claim_from_row(index).run_id} + for index in range(3) + ] + + class FakeDb: + async def execute( + self, + statement: Any, + params: dict[str, Any], + ) -> _Result: + assert statement is job._REQUEUE_REPAIRED_DBAPI_FAILURES_SQL + assert params == { + "project_id": "awoooi", + "agent_id": job.BACKFILL_ITEM_AGENT_ID, + "repair_generation": job.BACKFILL_DB_CONTRACT_REPAIR_GENERATION, + "limit": job.MAX_SCOPE_ROWS, + } + return _Result(rows=returned) + + monkeypatch.setattr( + job, + "get_db_context", + lambda _project_id: _DbContext(FakeDb()), + ) + + requeued = await job._requeue_repaired_dbapi_failures( + project_id="awoooi", + limit=job.MAX_SCOPE_ROWS + 500, + ) + + assert requeued == 3 + sql = str(job._REQUEUE_REPAIRED_DBAPI_FAILURES_SQL) + assert "E-BACKFILL-EXHAUSTED" in sql + assert "processor_error:DBAPIError" in sql + assert "runtime_execution_authorized" in sql + assert "recovery_generation" in sql + assert "FOR UPDATE SKIP LOCKED" in sql + assert "attempt_count = 0" in sql + + @pytest.mark.asyncio @pytest.mark.parametrize( ("item_total", "completed_total", "expected_status", "expected_cursor_state"), @@ -1356,6 +1401,8 @@ async def test_default_processor_accepts_repo_known_lan_without_token( update_cursor = AsyncMock( return_value={"status": "completed", "remaining_item_total": 0} ) + requeue = AsyncMock(return_value=0) + monkeypatch.setattr(job, "_requeue_repaired_dbapi_failures", requeue) monkeypatch.setattr(job, "_reserve_legacy_backup_page", reserve) monkeypatch.setattr(job, "_claim_legacy_backup_items", claim) monkeypatch.setattr(job, "_update_cursor_counters", update_cursor) @@ -1371,6 +1418,9 @@ async def test_default_processor_accepts_repo_known_lan_without_token( assert result["status"] == "completed" assert result["auth_mode"] == "lan_allowlist" assert result["relay_preflight"]["ready"] is True + assert result["requeued_repaired_dbapi"] == 0 + assert result["backfill_state_write_performed"] is False + requeue.assert_awaited_once_with(project_id="awoooi", limit=100) reserve.assert_awaited_once() claim.assert_awaited_once()