From 01f91154e0327ca97e8dec72181c3d66cc676b9d Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 3 Aug 2026 03:47:41 +0200 Subject: [PATCH] Add durable external-effect runtime identity --- docs/DURABLE_RECOVERY_OPERATIONS.md | 3 + src/govoplan_core/celery_app.py | 12 ++++ src/govoplan_core/core/recovery.py | 17 +++++- src/govoplan_core/core/recovery_runtime.py | 41 +++++++++++++ .../core/runtime_coordination.py | 20 +++++++ src/govoplan_core/server/default_config.py | 7 ++- tests/test_api_smoke.py | 13 +++++ tests/test_recovery_runtime.py | 23 ++++++++ tests/test_runtime_agents.py | 57 ++++++++++++++++++- 9 files changed, 186 insertions(+), 7 deletions(-) diff --git a/docs/DURABLE_RECOVERY_OPERATIONS.md b/docs/DURABLE_RECOVERY_OPERATIONS.md index 42a9ad1..bf6a0a9 100644 --- a/docs/DURABLE_RECOVERY_OPERATIONS.md +++ b/docs/DURABLE_RECOVERY_OPERATIONS.md @@ -11,6 +11,9 @@ 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. +A definitive pre-effect or provider rejection records terminal `rejected` +evidence instead of being mislabeled as success, atomic rollback, or recovery +work. If a runtime disappears, another runtime may claim the operation only after the lease expires. The takeover records both fences. A stale compensatable operation diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index 70463ac..af291f0 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -9,6 +9,7 @@ import time from celery import Celery from celery.signals import ( heartbeat_sent, + task_prerun, worker_process_init, worker_ready, worker_shutdown, @@ -68,6 +69,7 @@ from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.runtime import configure_runtime from govoplan_core.core.runtime_coordination import ( RuntimeIdentity, + bind_process_runtime_identity, heartbeat_runtime_node, register_runtime_node, runtime_identity, @@ -208,6 +210,7 @@ def _worker_runtime_identity(sender: object | None = None) -> RuntimeIdentity: role="worker", node_id=hostname, ) + bind_process_runtime_identity(_worker_identity) return _worker_identity @@ -222,10 +225,19 @@ def _worker_metadata() -> dict[str, object]: @worker_process_init.connect def _reset_worker_process_database(**_kwargs) -> None: + global _worker_identity + # SQLAlchemy pools must not be shared across prefork child processes. + _worker_identity = None + bind_process_runtime_identity(None) configure_database(settings.database_url, dispose_previous=True) +@task_prerun.connect +def _bind_worker_effect_identity(task=None, **_kwargs) -> None: + _worker_runtime_identity(task) + + @worker_ready.connect def _register_worker_runtime(sender=None, **_kwargs) -> None: global _worker_consumer diff --git a/src/govoplan_core/core/recovery.py b/src/govoplan_core/core/recovery.py index 710d4bf..96ba4fb 100644 --- a/src/govoplan_core/core/recovery.py +++ b/src/govoplan_core/core/recovery.py @@ -41,6 +41,7 @@ class RecoveryStatus(StrEnum): PREPARED = "prepared" RUNNING = "running" SUCCEEDED = "succeeded" + REJECTED = "rejected" FAILED = "failed" OUTCOME_UNKNOWN = "outcome_unknown" RECOVERY_REQUIRED = "recovery_required" @@ -52,6 +53,7 @@ class RecoveryStatus(StrEnum): TERMINAL_RECOVERY_STATUSES = frozenset( { RecoveryStatus.SUCCEEDED.value, + RecoveryStatus.REJECTED.value, RecoveryStatus.FAILED.value, RecoveryStatus.RECOVERED.value, RecoveryStatus.MANUAL_INTERVENTION.value, @@ -69,6 +71,7 @@ _TRANSITIONS: dict[str, frozenset[str]] = { RecoveryStatus.RUNNING.value: frozenset( { RecoveryStatus.SUCCEEDED.value, + RecoveryStatus.REJECTED.value, RecoveryStatus.FAILED.value, RecoveryStatus.OUTCOME_UNKNOWN.value, RecoveryStatus.RECOVERY_REQUIRED.value, @@ -431,7 +434,11 @@ def transition_recovery_operation( elif status == RecoveryStatus.RECOVERED: locked.recovered_at = observed_at locked.completed_at = observed_at - elif status in {RecoveryStatus.FAILED, RecoveryStatus.MANUAL_INTERVENTION}: + elif status in { + RecoveryStatus.REJECTED, + RecoveryStatus.FAILED, + RecoveryStatus.MANUAL_INTERVENTION, + }: locked.completed_at = observed_at session.add(locked) record_recovery_checkpoint( @@ -591,7 +598,11 @@ def _validate_transition_evidence( evidence: dict[str, Any], failure_summary: str | None, ) -> None: - if status in {RecoveryStatus.SUCCEEDED, RecoveryStatus.RECOVERED}: + if status in { + RecoveryStatus.SUCCEEDED, + RecoveryStatus.REJECTED, + RecoveryStatus.RECOVERED, + }: checks = evidence.get("checks") if ( evidence.get("verified") is not True @@ -602,7 +613,7 @@ def _validate_transition_evidence( or not checks ): raise RecoveryGuaranteeError( - "Successful recovery transitions require verified evidence and check results" + "Verified terminal transitions require verified evidence and check results" ) if status == RecoveryStatus.MANUAL_INTERVENTION and not failure_summary: raise RecoveryGuaranteeError( diff --git a/src/govoplan_core/core/recovery_runtime.py b/src/govoplan_core/core/recovery_runtime.py index f5f5300..a8e4e8d 100644 --- a/src/govoplan_core/core/recovery_runtime.py +++ b/src/govoplan_core/core/recovery_runtime.py @@ -105,6 +105,46 @@ class DurableRecoveryOperation: session.commit() self.closed = True + def fail(self, *, summary: str, evidence: dict[str, Any]) -> None: + """Finish a verified, ordinary failure that needs no recovery.""" + + with self.session_factory() as session: + operation, claim = self._locked_and_renewed(session) + transition_recovery_operation( + session, + operation, + status=RecoveryStatus.FAILED, + kind="verified-failure", + summary=summary, + evidence=evidence, + failure_summary=summary, + lease_claim=claim, + ) + self._verify_chain(session) + release_lease(session, claim) + session.commit() + self.closed = True + + def reject(self, *, summary: str, evidence: dict[str, Any]) -> None: + """Finish an operation with a verified definitive rejection.""" + + with self.session_factory() as session: + operation, claim = self._locked_and_renewed(session) + transition_recovery_operation( + session, + operation, + status=RecoveryStatus.REJECTED, + kind="verified-rejection", + summary=summary, + evidence=evidence, + failure_summary=summary, + lease_claim=claim, + ) + self._verify_chain(session) + release_lease(session, claim) + session.commit() + self.closed = True + def compensate( self, *, @@ -339,6 +379,7 @@ def claim_durable_recovery_operation( raise RecoveryGuaranteeError("Recovery operation was not found") if candidate.status in { RecoveryStatus.SUCCEEDED.value, + RecoveryStatus.REJECTED.value, RecoveryStatus.FAILED.value, RecoveryStatus.RECOVERED.value, RecoveryStatus.MANUAL_INTERVENTION.value, diff --git a/src/govoplan_core/core/runtime_coordination.py b/src/govoplan_core/core/runtime_coordination.py index 350de5f..d389ce3 100644 --- a/src/govoplan_core/core/runtime_coordination.py +++ b/src/govoplan_core/core/runtime_coordination.py @@ -122,6 +122,26 @@ class RuntimeIdentity: queues: tuple[str, ...] = () +_process_runtime_identity: RuntimeIdentity | None = None + + +def bind_process_runtime_identity(identity: RuntimeIdentity | None) -> None: + """Bind the authority identity used by effects in this OS process.""" + + global _process_runtime_identity + _process_runtime_identity = identity + + +def process_runtime_identity() -> RuntimeIdentity: + """Return the process authority or fail before a consequential effect.""" + + if _process_runtime_identity is None: + raise RuntimeCoordinationError( + "No runtime identity is bound to the current process" + ) + return _process_runtime_identity + + @dataclass(frozen=True, slots=True) class LeaseClaim: installation_id: str diff --git a/src/govoplan_core/server/default_config.py b/src/govoplan_core/server/default_config.py index 7c476f8..a80d3e5 100644 --- a/src/govoplan_core/server/default_config.py +++ b/src/govoplan_core/server/default_config.py @@ -12,7 +12,11 @@ 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.core.runtime_coordination import ( + RuntimeIdentity, + bind_process_runtime_identity, + runtime_identity, +) from govoplan_core.settings import Settings, settings @@ -92,6 +96,7 @@ def register_health_details( software_version=app.version, module_ids=module_ids, ) + bind_process_runtime_identity(app.state.govoplan_runtime_identity) @app.get("/health/details") def health_details( diff --git a/tests/test_api_smoke.py b/tests/test_api_smoke.py index 1166316..766f96a 100644 --- a/tests/test_api_smoke.py +++ b/tests/test_api_smoke.py @@ -4377,6 +4377,7 @@ class ApiSmokeTests(unittest.TestCase): with SessionLocal() as session: job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one() + job_id = job.id generated_eml = self._stored_campaign_eml(job) sent = self.client.post( @@ -4398,6 +4399,18 @@ class ApiSmokeTests(unittest.TestCase): self.assertTrue(raw_filename) captured_eml = (_TEST_ROOT / "mock-mailbox" / "messages" / str(raw_filename)).read_bytes() self.assertEqual(captured_eml, generated_eml) + with SessionLocal() as session: + operation = ( + session.query(RecoveryOperation) + .filter( + RecoveryOperation.operation_type + == "external-channel-delivery", + RecoveryOperation.resource_id == job_id, + ) + .one() + ) + self.assertEqual(operation.status, RecoveryStatus.SUCCEEDED.value) + self.assertTrue(verify_recovery_evidence_chain(session, operation.id)) def test_send_now_rejects_modified_generated_eml_before_delivery(self) -> None: headers, _ = self._login() diff --git a/tests/test_recovery_runtime.py b/tests/test_recovery_runtime.py index b9c70c9..5d184f8 100644 --- a/tests/test_recovery_runtime.py +++ b/tests/test_recovery_runtime.py @@ -115,6 +115,29 @@ def test_same_fence_cannot_start_duplicate_running_operation() -> None: engine.dispose() +def test_verified_provider_rejection_is_terminal_without_recovery() -> None: + engine, factory = _fixture() + try: + started = _start(factory, _identity("worker-1", "incarnation-1")) + assert started.operation is not None + started.operation.reject( + summary="Provider definitively rejected the request", + evidence={ + "verified": True, + "provider_outcome": "rejected", + "checks": {"provider_response": "definitive-rejection"}, + }, + ) + with factory() as session: + operation = session.get(RecoveryOperation, started.operation_id) + assert operation is not None + assert operation.status == RecoveryStatus.REJECTED.value + assert operation.completed_at is not None + assert verify_recovery_evidence_chain(session, operation.id) + finally: + engine.dispose() + + def test_other_runtime_cannot_use_an_active_fence() -> None: engine, factory = _fixture() try: diff --git a/tests/test_runtime_agents.py b/tests/test_runtime_agents.py index 80e1082..13fe561 100644 --- a/tests/test_runtime_agents.py +++ b/tests/test_runtime_agents.py @@ -6,6 +6,8 @@ from types import SimpleNamespace from govoplan_core.core.runtime_coordination import ( RuntimeCoordinationError, RuntimeIdentity, + bind_process_runtime_identity, + process_runtime_identity, ) from govoplan_core.server.runtime_agent import RuntimeNodeAgent @@ -64,6 +66,32 @@ def test_api_runtime_agent_fails_readiness_on_heartbeat_error() -> None: assert agent.coordination_healthy is True +def test_process_runtime_identity_is_explicit_and_replaceable() -> None: + from govoplan_core.core import runtime_coordination + + previous = runtime_coordination._process_runtime_identity + identity = RuntimeIdentity( + installation_id="installation-1", + node_id="api-1", + incarnation="incarnation-1", + role="api", + software_version="0.1.14", + composition_hash="a" * 64, + ) + try: + bind_process_runtime_identity(None) + try: + process_runtime_identity() + except RuntimeCoordinationError: + pass + else: # pragma: no cover - assertion branch + raise AssertionError("An unbound process identity must fail closed") + bind_process_runtime_identity(identity) + assert process_runtime_identity() is identity + finally: + bind_process_runtime_identity(previous) + + def test_worker_disables_consumers_without_reclaiming_stale_identity( monkeypatch, ) -> None: @@ -127,8 +155,19 @@ def test_worker_disables_consumers_without_reclaiming_stale_identity( def test_worker_child_replaces_inherited_database_pool(monkeypatch) -> None: from govoplan_core import celery_app + from govoplan_core.core import runtime_coordination calls: list[tuple[str, bool]] = [] + previous_process_identity = runtime_coordination._process_runtime_identity + previous_worker_identity = celery_app._worker_identity + inherited = RuntimeIdentity( + installation_id="installation-1", + node_id="parent-worker", + incarnation="parent-incarnation", + role="worker", + software_version="0.1.14", + composition_hash="a" * 64, + ) monkeypatch.setattr( celery_app, "configure_database", @@ -136,7 +175,19 @@ def test_worker_child_replaces_inherited_database_pool(monkeypatch) -> None: (url, dispose_previous) ), ) + try: + celery_app._worker_identity = inherited + bind_process_runtime_identity(inherited) + celery_app._reset_worker_process_database() - celery_app._reset_worker_process_database() - - assert calls == [(celery_app.settings.database_url, True)] + assert calls == [(celery_app.settings.database_url, True)] + assert celery_app._worker_identity is None + try: + process_runtime_identity() + except RuntimeCoordinationError: + pass + else: # pragma: no cover - assertion branch + raise AssertionError("A worker child must discard inherited authority") + finally: + celery_app._worker_identity = previous_worker_identity + bind_process_runtime_identity(previous_process_identity)