From 4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 12:59:20 +0800 Subject: [PATCH] fix(alerting): cluster legacy backup incident replay --- apps/api/src/core/config.py | 5 +- .../backup_restore_legacy_backfill_job.py | 40 +++++++++--- ...test_backup_restore_legacy_backfill_job.py | 65 +++++++++++++++---- 3 files changed, 88 insertions(+), 22 deletions(-) diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index 2aec50780..053cc8646 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -1125,10 +1125,11 @@ class Settings(BaseSettings): description="Timeout for fail-open Agent99 SRE alert relay POSTs.", ) ENABLE_BACKUP_RESTORE_LEGACY_BACKFILL: bool = Field( - default=True, + default=False, description=( "Enable bounded durable BackupCheck replay for legacy backup/restore " - "Telegram receipt cards when a trusted Agent99 relay is configured." + "Telegram receipt cards when a trusted Agent99 relay is configured. " + "Keep disabled until the recurrence-cluster canary is verified." ), ) BACKUP_RESTORE_LEGACY_BACKFILL_STARTUP_SLEEP_SECONDS: float = Field( 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 4d37f9ff5..5372770a2 100644 --- a/apps/api/src/jobs/backup_restore_legacy_backfill_job.py +++ b/apps/api/src/jobs/backup_restore_legacy_backfill_job.py @@ -37,7 +37,7 @@ from src.services.backup_restore_alertmanager_ingress import ( logger = structlog.get_logger(__name__) -BACKFILL_SCHEMA_VERSION = "backup_restore_legacy_backfill_v1" +BACKFILL_SCHEMA_VERSION = "backup_restore_legacy_backfill_v2" BACKFILL_CURSOR_SCHEMA_VERSION = "backup_restore_legacy_backfill_cursor_v1" BACKFILL_RESULT_SCHEMA_VERSION = "backup_restore_legacy_backfill_result_v1" BACKFILL_CURSOR_AGENT_ID = "backup_restore_legacy_backfill_cursor" @@ -59,6 +59,9 @@ MAX_SCOPE_ROWS = 100 REPO_KNOWN_LAN_RELAY_HOST = "192.168.0.99" REPO_KNOWN_LAN_RELAY_PORT = 8787 REPO_KNOWN_LAN_RELAY_PATH = "/agent99/sre-alert" +LEGACY_OCCURRENCE_SCHEMA_VERSION = "backup_restore_legacy_occurrence_v1" +RECURRENCE_OCCURRENCE_SCHEMA_VERSION = "backup_restore_recurrence_occurrence_v2" +RECURRENCE_FINGERPRINT_PREFIX = "legacy-backup-recurrence:" Processor = Callable[..., Awaitable[dict[str, Any]]] @@ -878,21 +881,22 @@ def build_legacy_backup_occurrence( raise ValueError("backup_backfill_source_fingerprint_required") fingerprint_hash = _sha256(fingerprint)[:16] occurrence_id = f"legacy-backup:{source_message}:{fingerprint_hash}" - dispatch_source_fingerprint = "legacy-backup-occurrence:" + _sha256( - f"{project}\x1f{source_message}\x1f{fingerprint}" + recurrence_id = f"{RECURRENCE_FINGERPRINT_PREFIX}{fingerprint_hash}" + dispatch_source_fingerprint = RECURRENCE_FINGERPRINT_PREFIX + _sha256( + f"{project}\x1f{fingerprint}" ) owner = build_backup_restore_ingress_owner( project_id=project, source_fingerprint=dispatch_source_fingerprint, - source_event_fingerprint=occurrence_id, - source_started_at=queued, + source_event_fingerprint=recurrence_id, + source_started_at="", ) item_run_id = uuid5( NAMESPACE_URL, f"{BACKFILL_SCHEMA_VERSION}:{project}:{occurrence_id}", ) return { - "schema_version": "backup_restore_legacy_occurrence_v1", + "schema_version": RECURRENCE_OCCURRENCE_SCHEMA_VERSION, "project_id": project, "source_message_id": source_message, "source_fingerprint": fingerprint, @@ -900,6 +904,7 @@ def build_legacy_backup_occurrence( "dispatch_source_fingerprint": dispatch_source_fingerprint, "source_queued_at": queued, "occurrence_id": occurrence_id, + "recurrence_id": recurrence_id, "backfill_run_id": str(item_run_id), "incident_id": owner["incident_id"], "approval_id": owner["approval_id"], @@ -992,7 +997,7 @@ def _item_payload_from_row( else: missing_hash = _sha256(f"missing:{project_id}:{message_id}")[:16] occurrence = { - "schema_version": "backup_restore_legacy_occurrence_v1", + "schema_version": RECURRENCE_OCCURRENCE_SCHEMA_VERSION, "project_id": project_id, "source_message_id": message_id, "source_fingerprint": "", @@ -1000,6 +1005,7 @@ def _item_payload_from_row( "dispatch_source_fingerprint": "", "source_queued_at": queued_at, "occurrence_id": f"legacy-backup-blocked:{message_id}:{missing_hash}", + "recurrence_id": None, "backfill_run_id": str( uuid5( NAMESPACE_URL, @@ -1760,6 +1766,22 @@ async def _replay_backfill_claim( processor: Processor, relay_auth_mode: str = "injected_processor", ) -> str: + if claim.item.get("schema_version") == LEGACY_OCCURRENCE_SCHEMA_VERSION: + result = _public_result( + status="superseded_unique_incident_replay_no_dispatch", + occurrence=claim.item, + reason="superseded_by_recurrence_cluster_v2", + auth_mode=relay_auth_mode, + ) + finished = await _finish_backfill_claim( + project_id=project_id, + claim=claim, + state="completed", + result=result, + error_code=None, + ) + return "deduplicated" if finished else "lease_lost" + card = await _load_legacy_backup_card(project_id=project_id, claim=claim) if card is None: result = _public_result( @@ -1887,8 +1909,8 @@ async def _replay_backfill_claim( }, annotations={"summary": "Read-only BackupCheck historical evidence replay"}, source_fingerprint=occurrence["dispatch_source_fingerprint"], - source_event_fingerprint=occurrence["occurrence_id"], - source_started_at=card.queued_at, + source_event_fingerprint=occurrence["recurrence_id"], + source_started_at="", alert_category="backup", notification_type="TYPE-1", source_url=None, 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 6d9b59de2..c498ff7be 100644 --- a/apps/api/tests/test_backup_restore_legacy_backfill_job.py +++ b/apps/api/tests/test_backup_restore_legacy_backfill_job.py @@ -10,6 +10,7 @@ from uuid import NAMESPACE_URL, UUID, uuid5 import pytest +from src.core.config import Settings from src.core.context import ( clear_project_context, get_current_project_id, @@ -159,17 +160,57 @@ def test_legacy_occurrence_identity_is_stable_and_source_specific() -> None: ) assert first == replay - for other in (next_message, next_fingerprint): - assert other["occurrence_id"] != first["occurrence_id"] - assert other["incident_id"] != first["incident_id"] - assert other["work_item_id"] != first["work_item_id"] - assert other["backfill_run_id"] != first["backfill_run_id"] + assert next_message["occurrence_id"] != first["occurrence_id"] + assert next_message["backfill_run_id"] != first["backfill_run_id"] + assert next_message["recurrence_id"] == first["recurrence_id"] + assert next_message["incident_id"] == first["incident_id"] + assert next_message["work_item_id"] == first["work_item_id"] + for key in ("recurrence_id", "incident_id", "work_item_id"): + assert next_fingerprint[key] != first[key] + assert next_fingerprint["backfill_run_id"] != first["backfill_run_id"] assert first["source_message_id"] in first["occurrence_id"] - assert first["dispatch_source_fingerprint"].startswith("legacy-backup-occurrence:") + assert first["dispatch_source_fingerprint"].startswith( + job.RECURRENCE_FINGERPRINT_PREFIX + ) assert first["controlled_apply"] is False assert first["telegram_dispatch_performed"] is False +def test_legacy_replay_is_disabled_until_recurrence_canary_is_verified() -> None: + assert ( + Settings.model_fields["ENABLE_BACKUP_RESTORE_LEGACY_BACKFILL"].default + is False + ) + + +@pytest.mark.asyncio +async def test_v1_unique_incident_claim_is_superseded_without_dispatch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + claim = _claim_from_row(1) + claim.item["schema_version"] = job.LEGACY_OCCURRENCE_SCHEMA_VERSION + processor = AsyncMock() + finish = AsyncMock(return_value=True) + monkeypatch.setattr(job, "_finish_backfill_claim", finish) + + result = await job._replay_backfill_claim( + project_id="awoooi", + claim=claim, + processor=processor, + ) + + assert result == "deduplicated" + processor.assert_not_awaited() + finish.assert_awaited_once() + call = finish.await_args.kwargs + assert call["state"] == "completed" + assert call["error_code"] is None + assert call["result"]["status"] == ( + "superseded_unique_incident_replay_no_dispatch" + ) + assert call["result"]["reason"] == "superseded_by_recurrence_cluster_v2" + + def test_cursor_never_reports_completed_when_any_item_failed() -> None: detail = { "source_exhausted": True, @@ -776,7 +817,9 @@ async def test_exact_98_legacy_cards_replay_in_bounded_restart_safe_batches( assert kwargs["project_id"] == "awoooi" assert kwargs["labels"]["event_type"] == job.BACKUP_RESTORE_EVENT_TYPE assert kwargs["labels"]["lane"] == job.BACKUP_RESTORE_LANE - assert kwargs["source_event_fingerprint"].startswith("legacy-backup:") + assert kwargs["source_event_fingerprint"].startswith( + job.RECURRENCE_FINGERPRINT_PREFIX + ) processor_calls.append(kwargs) return _durable_dispatch_result(kwargs) @@ -871,10 +914,10 @@ async def test_exact_98_legacy_cards_replay_in_bounded_restart_safe_batches( len({call["labels"]["legacy_source_message_id"] for call in processor_calls}) == 98 ) - assert len({call["source_fingerprint"] for call in processor_calls}) == 98 - assert len({value["incident_id"] for value in projected.values()}) == 98 - assert len({value["work_item_id"] for value in projected.values()}) == 98 - assert len({value["run_id"] for value in projected.values()}) == 98 + assert len({call["source_fingerprint"] for call in processor_calls}) == 1 + assert len({value["incident_id"] for value in projected.values()}) == 1 + assert len({value["work_item_id"] for value in projected.values()}) == 1 + assert len({value["run_id"] for value in projected.values()}) == 1 assert {value["auth_mode"] for value in projected.values()} == { "injected_processor" }