diff --git a/docs/DURABLE_RECOVERY_OPERATIONS.md b/docs/DURABLE_RECOVERY_OPERATIONS.md new file mode 100644 index 0000000..42a9ad1 --- /dev/null +++ b/docs/DURABLE_RECOVERY_OPERATIONS.md @@ -0,0 +1,25 @@ +# Durable Recovery Operations + +Modules must use `begin_durable_recovery_operation` for work whose effects can +outlive the caller's SQLAlchemy transaction. The helper commits the canonical +request hash, recovery plan, precondition evidence, running state, and lease +fence before the caller mutates object storage, a queue, a filesystem, or an +external provider. + +Each later checkpoint is written through an independent database session. A +business-transaction rollback therefore cannot erase evidence of an earlier +effect. Successful completion requires concrete verification checks and a valid +hash chain. Compensation likewise records recovery-required, recovering, and +verified-recovered checkpoints rather than reporting an ordinary failure. + +If a runtime disappears, another runtime may claim the operation only after the +lease expires. The takeover records both fences. A stale compensatable operation +becomes recovery-required; a stale forward-only or irreversible external effect +becomes outcome-unknown; a database-only atomic operation is recorded failed +because its transaction rolled back. Takeover never re-executes the original +request automatically. + +Evidence and metadata may contain opaque references, digests, counts, and +provider result codes. They must never contain credentials or resolved secrets. +Ops is the platform surface for unresolved operation status; owning modules must +provide the reconciliation action and business-level explanation. diff --git a/src/govoplan_core/core/recovery_runtime.py b/src/govoplan_core/core/recovery_runtime.py new file mode 100644 index 0000000..f5f5300 --- /dev/null +++ b/src/govoplan_core/core/recovery_runtime.py @@ -0,0 +1,446 @@ +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from govoplan_core.core.recovery import ( + RecoveryGuaranteeError, + RecoveryMode, + RecoveryOperation, + RecoveryPlan, + RecoveryStatus, + plan_recovery_operation, + prepare_recovery_operation, + record_recovery_checkpoint, + start_recovery_operation, + transition_recovery_operation, + verify_recovery_evidence_chain, +) +from govoplan_core.core.runtime_coordination import ( + LeaseClaim, + RuntimeIdentity, + acquire_lease, + release_lease, + renew_lease, +) + + +SessionFactory = Callable[[], Session] + + +class RecoveryOperationBusy(RecoveryGuaranteeError): + pass + + +class RecoveryOperationStateConflict(RecoveryGuaranteeError): + def __init__(self, operation_id: str, status: str) -> None: + self.operation_id = operation_id + self.status = status + super().__init__( + f"Recovery operation {operation_id} is already {status}; " + "reconcile it before starting another effect" + ) + + +@dataclass(frozen=True, slots=True) +class DurableRecoveryStart: + operation_id: str + status: str + replayed: bool + operation: DurableRecoveryOperation | None + + +@dataclass(slots=True) +class DurableRecoveryOperation: + """Append checkpoints in independent, committed transactions. + + The caller's business transaction may roll back without erasing evidence + that an object, queue, filesystem, or provider effect already occurred. + """ + + session_factory: SessionFactory + operation_id: str + lease_claim: LeaseClaim + lease_ttl_seconds: int + closed: bool = False + + def checkpoint( + self, + *, + kind: str, + summary: str, + evidence: dict[str, Any], + ) -> None: + with self.session_factory() as session: + operation, claim = self._locked_and_renewed(session) + record_recovery_checkpoint( + session, + operation, + kind=kind, + summary=summary, + evidence=evidence, + lease_claim=claim, + ) + self._verify_chain(session) + session.commit() + + def succeed(self, *, evidence: dict[str, Any]) -> None: + with self.session_factory() as session: + operation, claim = self._locked_and_renewed(session) + transition_recovery_operation( + session, + operation, + status=RecoveryStatus.SUCCEEDED, + kind="verified-success", + summary="Operation effects and authoritative state were verified", + evidence=evidence, + lease_claim=claim, + ) + self._verify_chain(session) + release_lease(session, claim) + session.commit() + self.closed = True + + def compensate( + self, + *, + failure_summary: str, + failure_evidence: dict[str, Any], + recovery_evidence: dict[str, Any], + ) -> None: + with self.session_factory() as session: + operation, claim = self._locked_and_renewed(session) + if operation.status == RecoveryStatus.RUNNING.value: + operation = transition_recovery_operation( + session, + operation, + status=RecoveryStatus.RECOVERY_REQUIRED, + kind="compensation-required", + summary="The started operation requires explicit compensation", + evidence=failure_evidence, + failure_summary=failure_summary, + lease_claim=claim, + ) + if operation.status == RecoveryStatus.RECOVERY_REQUIRED.value: + operation = transition_recovery_operation( + session, + operation, + status=RecoveryStatus.RECOVERING, + kind="compensation-started", + summary="Compensation started", + evidence={"failure_summary": failure_summary}, + lease_claim=claim, + ) + transition_recovery_operation( + session, + operation, + status=RecoveryStatus.RECOVERED, + kind="compensation-verified", + summary="Compensation restored the declared invariant", + evidence=recovery_evidence, + lease_claim=claim, + ) + self._verify_chain(session) + release_lease(session, claim) + session.commit() + self.closed = True + + def unresolved( + self, + *, + status: RecoveryStatus, + summary: str, + evidence: dict[str, Any], + failure_summary: str, + ) -> None: + if status not in { + RecoveryStatus.OUTCOME_UNKNOWN, + RecoveryStatus.RECOVERY_REQUIRED, + }: + raise ValueError("Unresolved operations require an unresolved status") + with self.session_factory() as session: + operation, claim = self._locked_and_renewed(session) + transition_recovery_operation( + session, + operation, + status=status, + kind="unresolved-effect", + summary=summary, + evidence=evidence, + failure_summary=failure_summary, + lease_claim=claim, + ) + self._verify_chain(session) + release_lease(session, claim) + session.commit() + self.closed = True + + def release_unresolved(self) -> None: + """Release authority after a process-local exception. + + This does not alter the operation state. A later recovery claim treats a + stale `running` operation according to its declared recovery mode. + """ + + if self.closed: + return + with self.session_factory() as session: + operation, claim = self._locked_and_renewed(session) + record_recovery_checkpoint( + session, + operation, + kind="authority-released", + summary="Execution authority was released without a terminal claim", + evidence={"status": operation.status}, + lease_claim=claim, + ) + self._verify_chain(session) + release_lease(session, claim) + session.commit() + self.closed = True + + def _locked_and_renewed( + self, + session: Session, + ) -> tuple[RecoveryOperation, LeaseClaim]: + if self.closed: + raise RecoveryGuaranteeError("Recovery operation handle is closed") + claim = renew_lease( + session, + self.lease_claim, + ttl_seconds=self.lease_ttl_seconds, + ) + operation = session.execute( + select(RecoveryOperation) + .where(RecoveryOperation.id == self.operation_id) + .with_for_update() + ).scalar_one() + self.lease_claim = claim + return operation, claim + + def _verify_chain(self, session: Session) -> None: + if not verify_recovery_evidence_chain(session, self.operation_id): + raise RecoveryGuaranteeError( + "Recovery checkpoint chain verification failed" + ) + + +def begin_durable_recovery_operation( + session_factory: SessionFactory, + *, + identity: RuntimeIdentity, + module_id: str, + operation_type: str, + idempotency_key: str, + request: dict[str, Any], + recovery_plan: RecoveryPlan, + precondition_evidence: dict[str, Any], + lease_resource_key: str, + lease_ttl_seconds: int = 300, + resource_type: str | None = None, + resource_id: str | None = None, + metadata: dict[str, Any] | None = None, +) -> DurableRecoveryStart: + if lease_ttl_seconds < 1: + raise ValueError("Recovery lease TTL must be at least one second") + with session_factory() as session: + claim = acquire_lease( + session, + installation_id=identity.installation_id, + resource_key=lease_resource_key, + holder_node_id=identity.node_id, + holder_incarnation=identity.incarnation, + ttl_seconds=lease_ttl_seconds, + metadata={ + "module_id": module_id, + "operation_type": operation_type, + }, + ) + if claim is None: + raise RecoveryOperationBusy( + f"Another runtime owns the recovery fence for {lease_resource_key}" + ) + existing = session.execute( + select(RecoveryOperation).where( + RecoveryOperation.installation_id == identity.installation_id, + RecoveryOperation.module_id == module_id, + RecoveryOperation.idempotency_key == idempotency_key, + ) + ).scalar_one_or_none() + operation = plan_recovery_operation( + session, + installation_id=identity.installation_id, + module_id=module_id, + operation_type=operation_type, + idempotency_key=idempotency_key, + request=request, + recovery_plan=recovery_plan, + resource_type=resource_type, + resource_id=resource_id, + lease_claim=claim, + metadata=metadata, + ) + if existing is not None: + if operation.status == RecoveryStatus.SUCCEEDED.value: + release_lease(session, claim) + session.commit() + return DurableRecoveryStart( + operation_id=operation.id, + status=operation.status, + replayed=True, + operation=None, + ) + session.rollback() + raise RecoveryOperationStateConflict(operation.id, operation.status) + prepare_recovery_operation( + session, + operation, + evidence=precondition_evidence, + lease_claim=claim, + ) + start_recovery_operation( + session, + operation, + evidence={"lease_resource_key": lease_resource_key}, + lease_claim=claim, + ) + if not verify_recovery_evidence_chain(session, operation.id): + raise RecoveryGuaranteeError( + "Recovery checkpoint chain verification failed before side effects" + ) + session.commit() + return DurableRecoveryStart( + operation_id=operation.id, + status=operation.status, + replayed=False, + operation=DurableRecoveryOperation( + session_factory=session_factory, + operation_id=operation.id, + lease_claim=claim, + lease_ttl_seconds=lease_ttl_seconds, + ), + ) + + +def claim_durable_recovery_operation( + session_factory: SessionFactory, + *, + identity: RuntimeIdentity, + operation_id: str, + lease_ttl_seconds: int = 300, +) -> DurableRecoveryOperation: + with session_factory() as session: + candidate = session.get(RecoveryOperation, operation_id) + if candidate is None: + raise RecoveryGuaranteeError("Recovery operation was not found") + if candidate.status in { + RecoveryStatus.SUCCEEDED.value, + RecoveryStatus.FAILED.value, + RecoveryStatus.RECOVERED.value, + RecoveryStatus.MANUAL_INTERVENTION.value, + }: + raise RecoveryOperationStateConflict(candidate.id, candidate.status) + if not candidate.lease_resource_key: + raise RecoveryGuaranteeError( + "Recovery takeover requires an operation-bound lease resource" + ) + claim = acquire_lease( + session, + installation_id=identity.installation_id, + resource_key=candidate.lease_resource_key, + holder_node_id=identity.node_id, + holder_incarnation=identity.incarnation, + ttl_seconds=lease_ttl_seconds, + metadata={"recovery_operation_id": candidate.id}, + ) + if claim is None: + raise RecoveryOperationBusy( + f"Another runtime owns recovery operation {candidate.id}" + ) + operation = session.execute( + select(RecoveryOperation) + .where(RecoveryOperation.id == operation_id) + .with_for_update() + ).scalar_one() + previous_fence = { + "holder_node_id": operation.holder_node_id, + "holder_incarnation": operation.holder_incarnation, + "fence_number": operation.fencing_token, + } + operation.holder_node_id = claim.holder_node_id + operation.holder_incarnation = claim.holder_incarnation + operation.fencing_token = claim.fencing_token + session.add(operation) + record_recovery_checkpoint( + session, + operation, + kind="fence-takeover", + summary="A new runtime claimed explicit recovery authority", + evidence={ + "previous_fence": previous_fence, + "new_fence_number": claim.fencing_token, + }, + lease_claim=claim, + ) + if operation.status == RecoveryStatus.RUNNING.value: + mode = RecoveryMode(operation.mode) + if mode == RecoveryMode.ATOMIC: + transition_recovery_operation( + session, + operation, + status=RecoveryStatus.FAILED, + kind="stale-atomic-operation", + summary="The stale database-only transaction rolled back", + evidence={"previous_fence": previous_fence}, + failure_summary="Execution authority expired before commit", + lease_claim=claim, + ) + elif mode in {RecoveryMode.COMPENSATION, RecoveryMode.SNAPSHOT_RESTORE}: + transition_recovery_operation( + session, + operation, + status=RecoveryStatus.RECOVERY_REQUIRED, + kind="stale-effect-requires-recovery", + summary="Execution authority expired after effects may have started", + evidence={"previous_fence": previous_fence}, + failure_summary="Execution authority expired during a non-atomic operation", + lease_claim=claim, + ) + else: + transition_recovery_operation( + session, + operation, + status=RecoveryStatus.OUTCOME_UNKNOWN, + kind="stale-effect-outcome-unknown", + summary="Execution authority expired after an external effect may have started", + evidence={"previous_fence": previous_fence}, + failure_summary="External effect outcome requires reconciliation", + lease_claim=claim, + ) + if not verify_recovery_evidence_chain(session, operation.id): + raise RecoveryGuaranteeError("Recovery checkpoint chain verification failed") + if operation.status == RecoveryStatus.FAILED.value: + release_lease(session, claim) + session.commit() + raise RecoveryOperationStateConflict(operation.id, operation.status) + session.commit() + return DurableRecoveryOperation( + session_factory=session_factory, + operation_id=operation.id, + lease_claim=claim, + lease_ttl_seconds=lease_ttl_seconds, + ) + + +__all__ = [ + "DurableRecoveryOperation", + "DurableRecoveryStart", + "RecoveryOperationBusy", + "RecoveryOperationStateConflict", + "begin_durable_recovery_operation", + "claim_durable_recovery_operation", +] diff --git a/src/govoplan_core/server/default_config.py b/src/govoplan_core/server/default_config.py index e7a4477..7c476f8 100644 --- a/src/govoplan_core/server/default_config.py +++ b/src/govoplan_core/server/default_config.py @@ -12,6 +12,7 @@ from govoplan_core.db.bootstrap import bootstrap_dev_data, create_all_tables from govoplan_core.db.session import get_database from govoplan_core.server.config import GovoplanServerConfig from govoplan_core.server.runtime_agent import RuntimeNodeAgent +from govoplan_core.core.runtime_coordination import RuntimeIdentity, runtime_identity from govoplan_core.settings import Settings, settings @@ -49,11 +50,19 @@ async def lifespan(app: FastAPI): if registry is not None else () ) + configured_identity = getattr( + app.state, + "govoplan_runtime_identity", + None, + ) runtime_agent = RuntimeNodeAgent( settings=settings, software_version=app.version, module_ids=module_ids, metadata={"process": "api"}, + identity=configured_identity + if isinstance(configured_identity, RuntimeIdentity) + else None, ) await runtime_agent.start() app.state.govoplan_runtime_agent = runtime_agent @@ -73,6 +82,16 @@ def register_health_details( active_settings = ( config_settings if isinstance(config_settings, Settings) else settings ) + module_ids = tuple(manifest.id for manifest in registry.manifests()) + if not isinstance( + getattr(app.state, "govoplan_runtime_identity", None), + RuntimeIdentity, + ): + app.state.govoplan_runtime_identity = runtime_identity( + active_settings, + software_version=app.version, + module_ids=module_ids, + ) @app.get("/health/details") def health_details( diff --git a/src/govoplan_core/server/runtime_agent.py b/src/govoplan_core/server/runtime_agent.py index 61c862c..4d7bb57 100644 --- a/src/govoplan_core/server/runtime_agent.py +++ b/src/govoplan_core/server/runtime_agent.py @@ -17,6 +17,21 @@ from govoplan_core.db.session import get_database logger = logging.getLogger("govoplan.runtime") +def application_runtime_identity(app: object) -> RuntimeIdentity: + """Return the registered identity used to fence request-owned effects.""" + + state = getattr(app, "state", None) + agent = getattr(state, "govoplan_runtime_agent", None) + identity = getattr(agent, "identity", None) or getattr( + state, + "govoplan_runtime_identity", + None, + ) + if not isinstance(identity, RuntimeIdentity): + raise RuntimeError("The application runtime identity is not available") + return identity + + class RuntimeNodeAgent: """Register one process in the shared runtime directory and heartbeat it.""" @@ -30,9 +45,10 @@ class RuntimeNodeAgent: node_id: str | None = None, queues: tuple[str, ...] | None = None, metadata: dict[str, Any] | None = None, + identity: RuntimeIdentity | None = None, ) -> None: self.settings = settings - self.identity: RuntimeIdentity = runtime_identity( + self.identity: RuntimeIdentity = identity or runtime_identity( settings, software_version=software_version, module_ids=module_ids, @@ -126,4 +142,4 @@ class RuntimeNodeAgent: return True -__all__ = ["RuntimeNodeAgent"] +__all__ = ["RuntimeNodeAgent", "application_runtime_identity"] diff --git a/tests/test_api_smoke.py b/tests/test_api_smoke.py index 532b95b..1166316 100644 --- a/tests/test_api_smoke.py +++ b/tests/test_api_smoke.py @@ -44,6 +44,11 @@ from govoplan_core.db.migrations import alembic_config from govoplan_core.db.session import configure_database, set_database from govoplan_core.core.change_sequence import decode_sequence_watermark, prune_sequence_entries from govoplan_core.core.pagination import encode_keyset_cursor, keyset_query_fingerprint +from govoplan_core.core.recovery import ( + RecoveryOperation, + RecoveryStatus, + verify_recovery_evidence_chain, +) from govoplan_core.tenancy.scope import create_scope_tables, scope_registry from govoplan_access.backend.permissions.catalog import permission_catalog as access_permission_catalog @@ -101,6 +106,14 @@ class ApiSmokeTests(unittest.TestCase): payload = response.json() return {"Authorization": f"Bearer {payload['access_token']}"}, payload + def _stored_campaign_eml(self, job: object) -> bytes: + from govoplan_files.backend.storage.backends import get_storage_backend + + self.assertIsNone(getattr(job, "eml_local_path", None)) + storage_key = getattr(job, "eml_storage_key", None) + self.assertTrue(storage_key) + return get_storage_backend().get_bytes(str(storage_key)) + def _create_test_mail_profile( self, headers: dict[str, str], @@ -3304,6 +3317,29 @@ class ApiSmokeTests(unittest.TestCase): json={"write_eml": False}, ) self.assertEqual(built.status_code, 200, built.text) + replayed_build = self.client.post( + f"/api/v1/campaigns/versions/{version_id}/build", + headers=headers, + json={"write_eml": False}, + ) + self.assertEqual(replayed_build.status_code, 200, replayed_build.text) + with SessionLocal() as session: + operations = ( + session.query(RecoveryOperation) + .filter( + RecoveryOperation.module_id == "campaigns", + RecoveryOperation.resource_id == version_id, + ) + .all() + ) + self.assertEqual(1, len(operations)) + self.assertEqual( + RecoveryStatus.SUCCEEDED.value, + operations[0].status, + ) + self.assertTrue( + verify_recovery_evidence_chain(session, operations[0].id) + ) self.assertEqual(built.json()["built_count"], 1) mocked = self.client.post( @@ -3742,7 +3778,8 @@ class ApiSmokeTests(unittest.TestCase): with SessionLocal() as session: job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one() - self.assertTrue(job.eml_local_path) + self.assertIsNone(job.eml_local_path) + self.assertTrue(job.eml_storage_key) built_use = ( session.query(CampaignAttachmentUse) .filter( @@ -3979,7 +4016,9 @@ class ApiSmokeTests(unittest.TestCase): job = session.query(CampaignJob).filter(CampaignJob.campaign_id == campaign_id).one() self.assertEqual([item["email"] for item in job.resolved_recipients["from_all"]], ["local-from@example.org"]) self.assertEqual([item["email"] for item in job.resolved_recipients["to"]], ["global-to@example.org", "local-to@example.org"]) - message = BytesParser(policy=policy.default).parsebytes(Path(job.eml_local_path).read_bytes()) + message = BytesParser(policy=policy.default).parsebytes( + self._stored_campaign_eml(job) + ) self.assertIsNone(message["Sender"]) self.assertEqual([address.addr_spec for address in message["From"].addresses], ["local-from@example.org"]) self.assertEqual([address.addr_spec for address in message["To"].addresses], ["global-to@example.org", "local-to@example.org"]) @@ -4229,8 +4268,9 @@ class ApiSmokeTests(unittest.TestCase): with SessionLocal() as session: job = session.query(CampaignJob).filter(CampaignJob.campaign_id == campaign_id).one() - eml_path = Path(job.eml_local_path) - message = BytesParser(policy=policy.default).parsebytes(eml_path.read_bytes()) + message = BytesParser(policy=policy.default).parsebytes( + self._stored_campaign_eml(job) + ) uses = ( session.query(CampaignAttachmentUse) .filter( @@ -4337,8 +4377,7 @@ class ApiSmokeTests(unittest.TestCase): with SessionLocal() as session: job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one() - self.assertIsNotNone(job.eml_local_path) - generated_eml = Path(job.eml_local_path).read_bytes() + generated_eml = self._stored_campaign_eml(job) sent = self.client.post( f"/api/v1/campaigns/{campaign_id}/send-now", @@ -4368,13 +4407,19 @@ class ApiSmokeTests(unittest.TestCase): ) from govoplan_campaign.backend.db.models import CampaignJob + from govoplan_files.backend.storage.backends import get_storage_backend from govoplan_mail.backend.dev.mock_mailbox import list_records with SessionLocal() as session: job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one() - self.assertIsNotNone(job.eml_local_path) - eml_path = Path(job.eml_local_path) - eml_path.write_bytes(eml_path.read_bytes() + b"\r\nX-Tampered: true\r\n") + storage_key = job.eml_storage_key + generated_eml = self._stored_campaign_eml(job) + assert storage_key is not None + get_storage_backend().put_bytes( + storage_key, + generated_eml + b"\r\nX-Tampered: true\r\n", + content_type="message/rfc822", + ) sent = self.client.post( f"/api/v1/campaigns/{campaign_id}/send-now", diff --git a/tests/test_recovery_runtime.py b/tests/test_recovery_runtime.py new file mode 100644 index 0000000..b9c70c9 --- /dev/null +++ b/tests/test_recovery_runtime.py @@ -0,0 +1,182 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import create_engine, select +from sqlalchemy.orm import sessionmaker +import pytest + +from govoplan_core.core.recovery import ( + RecoveryCheckpoint, + RecoveryGuaranteeError, + RecoveryMode, + RecoveryOperation, + RecoveryPlan, + RecoveryStatus, + verify_recovery_evidence_chain, +) +from govoplan_core.core.recovery_runtime import ( + RecoveryOperationBusy, + RecoveryOperationStateConflict, + begin_durable_recovery_operation, + claim_durable_recovery_operation, +) +from govoplan_core.core.runtime_coordination import DistributedLease, RuntimeIdentity +from govoplan_core.db.base import Base + + +def _fixture(): + engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all( + engine, + tables=[ + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, + ], + ) + return engine, sessionmaker(bind=engine, expire_on_commit=False) + + +def _identity(node: str, incarnation: str) -> RuntimeIdentity: + return RuntimeIdentity( + installation_id="installation-1", + node_id=node, + incarnation=incarnation, + role="worker", + software_version="test", + composition_hash="a" * 64, + ) + + +def _start(factory, identity, *, key: str = "build-1"): + return begin_durable_recovery_operation( + factory, + identity=identity, + module_id="campaigns", + operation_type="build-artifacts", + idempotency_key=key, + request={"version_id": "version-1", "write_eml": True}, + recovery_plan=RecoveryPlan( + mode=RecoveryMode.COMPENSATION, + preconditions=("validated version is locked",), + compensation_steps=("delete build object prefix",), + verification_steps=("compare database and object manifests",), + ), + precondition_evidence={"validation_sha256": "b" * 64}, + lease_resource_key="campaign:build:version-1", + resource_type="campaign_version", + resource_id="version-1", + ) + + +def test_durable_operation_commits_before_caller_effect_and_replays_success() -> None: + engine, factory = _fixture() + try: + started = _start(factory, _identity("worker-1", "incarnation-1")) + assert started.operation is not None + with factory() as session: + persisted = session.get(RecoveryOperation, started.operation_id) + assert persisted is not None + assert persisted.status == RecoveryStatus.RUNNING.value + assert persisted.checkpoint_count == 3 + + started.operation.checkpoint( + kind="object-prefix-reserved", + summary="Build object prefix reserved", + evidence={"prefix": "campaign-artifacts/build-1/"}, + ) + started.operation.succeed( + evidence={ + "verified": True, + "checks": {"database_manifest": "matched", "object_manifest": "matched"}, + } + ) + + replay = _start(factory, _identity("worker-2", "incarnation-2")) + assert replay.replayed is True + assert replay.operation is None + with factory() as session: + assert verify_recovery_evidence_chain(session, started.operation_id) + finally: + engine.dispose() + + +def test_same_fence_cannot_start_duplicate_running_operation() -> None: + engine, factory = _fixture() + try: + identity = _identity("worker-1", "incarnation-1") + started = _start(factory, identity) + with pytest.raises(RecoveryOperationStateConflict, match="already running"): + _start(factory, identity) + assert started.operation is not None + started.operation.release_unresolved() + finally: + engine.dispose() + + +def test_other_runtime_cannot_use_an_active_fence() -> None: + engine, factory = _fixture() + try: + started = _start(factory, _identity("worker-1", "incarnation-1")) + with pytest.raises(RecoveryOperationBusy): + _start(factory, _identity("worker-2", "incarnation-2"), key="build-2") + assert started.operation is not None + started.operation.release_unresolved() + finally: + engine.dispose() + + +def test_expired_crash_fence_is_taken_over_as_recovery_required() -> None: + engine, factory = _fixture() + try: + started = _start(factory, _identity("worker-1", "incarnation-1")) + assert started.operation is not None + with factory() as session: + lease = session.execute(select(DistributedLease)).scalar_one() + lease.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1) + session.add(lease) + session.commit() + + recovery = claim_durable_recovery_operation( + factory, + identity=_identity("worker-2", "incarnation-2"), + operation_id=started.operation_id, + ) + with factory() as session: + operation = session.get(RecoveryOperation, started.operation_id) + assert operation is not None + assert operation.status == RecoveryStatus.RECOVERY_REQUIRED.value + assert operation.fencing_token == 2 + assert verify_recovery_evidence_chain(session, operation.id) + recovery.compensate( + failure_summary="worker stopped during object publication", + failure_evidence={"object_prefix": "campaign-artifacts/build-1/"}, + recovery_evidence={ + "verified": True, + "checks": {"object_prefix_empty": True}, + }, + ) + finally: + engine.dispose() + + +def test_tampered_checkpoint_blocks_verified_success() -> None: + engine, factory = _fixture() + try: + started = _start(factory, _identity("worker-1", "incarnation-1")) + assert started.operation is not None + with factory() as session: + checkpoint = session.execute( + select(RecoveryCheckpoint).order_by(RecoveryCheckpoint.sequence) + ).scalars().first() + assert checkpoint is not None + checkpoint.summary = "tampered" + session.add(checkpoint) + session.commit() + with pytest.raises(RecoveryGuaranteeError, match="chain verification failed"): + started.operation.succeed( + evidence={"verified": True, "checks": {"objects": "matched"}} + ) + finally: + engine.dispose()