diff --git a/docs/DURABLE_RECOVERY_OPERATIONS.md b/docs/DURABLE_RECOVERY_OPERATIONS.md index 82f19da..1ebae91 100644 --- a/docs/DURABLE_RECOVERY_OPERATIONS.md +++ b/docs/DURABLE_RECOVERY_OPERATIONS.md @@ -27,6 +27,14 @@ 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. +Database-only operations must use the durable handle's atomic terminal methods +when their module rows and final recovery checkpoint belong to one invariant. +Those methods stage the terminal checkpoint and lease release in the caller's +SQLAlchemy transaction, then commit the domain rows and recovery evidence +together. A failed commit rolls both back and leaves the previously durable +`running` record available for stale-fence handling; modules must not commit +their domain state first and close an `atomic` recovery record afterwards. + An owning module may reconcile an `outcome_unknown` provider effect through the claimed durable handle's `resolve_unknown` method. External evidence that the effect occurred records verified success. Evidence that it did not occur moves diff --git a/src/govoplan_core/core/recovery_runtime.py b/src/govoplan_core/core/recovery_runtime.py index f000b42..945d6ca 100644 --- a/src/govoplan_core/core/recovery_runtime.py +++ b/src/govoplan_core/core/recovery_runtime.py @@ -105,6 +105,56 @@ class DurableRecoveryOperation: session.commit() self.closed = True + def commit_atomic_success( + self, + session: Session, + *, + evidence: dict[str, Any], + ) -> None: + """Commit domain writes and verified success in one DB transaction.""" + + self._commit_atomic_terminal( + session, + status=RecoveryStatus.SUCCEEDED, + summary="Operation effects and authoritative state were verified", + kind="verified-success", + evidence=evidence, + ) + + def commit_atomic_failure( + self, + session: Session, + *, + summary: str, + evidence: dict[str, Any], + ) -> None: + """Commit domain failure evidence and the terminal state atomically.""" + + self._commit_atomic_terminal( + session, + status=RecoveryStatus.FAILED, + summary=summary, + kind="verified-failure", + evidence=evidence, + ) + + def commit_atomic_rejection( + self, + session: Session, + *, + summary: str, + evidence: dict[str, Any], + ) -> None: + """Commit a definitive rejection and its domain evidence atomically.""" + + self._commit_atomic_terminal( + session, + status=RecoveryStatus.REJECTED, + summary=summary, + kind="verified-rejection", + evidence=evidence, + ) + def fail(self, *, summary: str, evidence: dict[str, Any]) -> None: """Finish a verified, ordinary failure that needs no recovery.""" @@ -325,6 +375,49 @@ class DurableRecoveryOperation: self.lease_claim = claim return operation, claim + def _commit_atomic_terminal( + self, + session: Session, + *, + status: RecoveryStatus, + summary: str, + kind: str, + evidence: dict[str, Any], + ) -> None: + if status not in { + RecoveryStatus.SUCCEEDED, + RecoveryStatus.FAILED, + RecoveryStatus.REJECTED, + }: + raise ValueError("Unsupported atomic terminal recovery status") + try: + operation, claim = self._locked_and_renewed(session) + if operation.mode != RecoveryMode.ATOMIC.value: + raise RecoveryGuaranteeError( + "Atomic terminal commits require an atomic recovery plan" + ) + transition_recovery_operation( + session, + operation, + status=status, + kind=kind, + summary=summary, + evidence=evidence, + failure_summary=( + summary + if status in {RecoveryStatus.FAILED, RecoveryStatus.REJECTED} + else None + ), + lease_claim=claim, + ) + self._verify_chain(session) + release_lease(session, claim) + session.commit() + except Exception: + session.rollback() + raise + self.closed = True + def _verify_chain(self, session: Session) -> None: if not verify_recovery_evidence_chain(session, self.operation_id): raise RecoveryGuaranteeError( diff --git a/tests/test_recovery_runtime.py b/tests/test_recovery_runtime.py index 21aa962..7cddb97 100644 --- a/tests/test_recovery_runtime.py +++ b/tests/test_recovery_runtime.py @@ -1,8 +1,9 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone +from unittest.mock import patch -from sqlalchemy import create_engine, select +from sqlalchemy import Column, MetaData, String, Table, create_engine, select from sqlalchemy.orm import sessionmaker import pytest @@ -70,6 +71,24 @@ def _start(factory, identity, *, key: str = "build-1"): ) +def _start_atomic(factory, identity, *, key: str = "sync-1"): + return begin_durable_recovery_operation( + factory, + identity=identity, + module_id="connectors", + operation_type="read-snapshot", + idempotency_key=key, + request={"provider_id": "provider-1", "cursor": "revision-1"}, + recovery_plan=RecoveryPlan( + mode=RecoveryMode.ATOMIC, + preconditions=("the provider read is non-mutating",), + verification_steps=("compare the committed projection",), + ), + precondition_evidence={"provider_mutation": False}, + lease_resource_key="connectors:provider-1", + ) + + def test_durable_operation_commits_before_caller_effect_and_replays_success() -> None: engine, factory = _fixture() try: @@ -102,6 +121,74 @@ def test_durable_operation_commits_before_caller_effect_and_replays_success() -> engine.dispose() +def test_atomic_terminal_commits_domain_rows_and_recovery_evidence_together() -> None: + engine, factory = _fixture() + metadata = MetaData() + projection = Table( + "test_recovery_projection", + metadata, + Column("id", String(36), primary_key=True), + ) + metadata.create_all(engine) + try: + started = _start_atomic(factory, _identity("worker-1", "incarnation-1")) + assert started.operation is not None + with factory() as session: + session.execute(projection.insert().values(id="projection-1")) + started.operation.commit_atomic_success( + session, + evidence={ + "verified": True, + "checks": {"projection_id": "projection-1"}, + }, + ) + + with factory() as session: + assert session.scalar(select(projection.c.id)) == "projection-1" + operation = session.get(RecoveryOperation, started.operation_id) + assert operation is not None + assert operation.status == RecoveryStatus.SUCCEEDED.value + assert verify_recovery_evidence_chain(session, operation.id) + finally: + engine.dispose() + + +def test_failed_atomic_commit_rolls_back_domain_and_terminal_checkpoint() -> None: + engine, factory = _fixture() + metadata = MetaData() + projection = Table( + "test_recovery_projection_rollback", + metadata, + Column("id", String(36), primary_key=True), + ) + metadata.create_all(engine) + try: + started = _start_atomic(factory, _identity("worker-1", "incarnation-1")) + assert started.operation is not None + with factory() as session: + session.execute(projection.insert().values(id="rolled-back")) + with ( + patch.object(session, "commit", side_effect=RuntimeError("commit failed")), + pytest.raises(RuntimeError, match="commit failed"), + ): + started.operation.commit_atomic_success( + session, + evidence={ + "verified": True, + "checks": {"projection_id": "rolled-back"}, + }, + ) + + with factory() as session: + assert session.execute(select(projection.c.id)).all() == [] + operation = session.get(RecoveryOperation, started.operation_id) + assert operation is not None + assert operation.status == RecoveryStatus.RUNNING.value + assert verify_recovery_evidence_chain(session, operation.id) + finally: + engine.dispose() + + def test_same_fence_cannot_start_duplicate_running_operation() -> None: engine, factory = _fixture() try: