From dfa717b9ba8f9fc1fe284859f23f6a178c91879b Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 3 Aug 2026 05:43:54 +0200 Subject: [PATCH] Fence governed connector acquisitions --- README.md | 10 + docs/CONNECTOR_SOURCE_LIFECYCLE.md | 20 + src/govoplan_connectors/backend/manifest.py | 24 +- src/govoplan_connectors/backend/recovery.py | 381 ++++++++++++++++++ src/govoplan_connectors/backend/router.py | 87 +++- .../backend/sanctions_sources.py | 205 +++++++++- .../backend/tabular_sources.py | 4 + tests/test_recovery.py | 263 ++++++++++++ tests/test_sanctions_sources.py | 67 +++ 9 files changed, 1042 insertions(+), 19 deletions(-) create mode 100644 src/govoplan_connectors/backend/recovery.py create mode 100644 tests/test_recovery.py diff --git a/README.md b/README.md index 462f0f8..856b21c 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,16 @@ connector implementations or stores connector credentials. Database, REST/HTTP, directory, managed-file, and warehouse providers can implement the same origin contract without changing Datasources or Dataflow. +Governed sanctions and feed snapshot acquisitions use Core recovery operations. +The source revision/cursor, redacted dry-run decision, canonical request digest, +and distributed lease are durable before network I/O. Immutable snapshot rows +and the terminal recovery checkpoint commit atomically, and an +`Idempotency-Key` replays the committed result without contacting the provider. +Current acquisition transports are read-only. The exported external-mutation +recovery contract requires stable idempotency, provider verification, and +operator reconciliation, but no connector currently claims a production write +or delete path. + Development: ```bash diff --git a/docs/CONNECTOR_SOURCE_LIFECYCLE.md b/docs/CONNECTOR_SOURCE_LIFECYCLE.md index 5261df1..509ad02 100644 --- a/docs/CONNECTOR_SOURCE_LIFECYCLE.md +++ b/docs/CONNECTOR_SOURCE_LIFECYCLE.md @@ -138,6 +138,26 @@ Every connector that writes to an external system needs a reconciliation story: - explicit requested, approved, dispatched, possibly-executed, confirmed, and reconciled/corrected effect states +## Durable recovery operations + +Connectors declares two recovery classes. A read-only acquisition into an +immutable snapshot is `atomic`: the source revision or conditional cursor, +redacted dry-run decision, canonical request digest, and distributed +tenant/provider lease are durable before the fetch. The acquired domain rows +and terminal Core recovery checkpoint commit in one PostgreSQL transaction. A +caller-supplied `Idempotency-Key` replays that committed result without a second +provider request. A failed or stale transaction has no remote mutation and may +be repeated only as a new deliberate acquisition. + +An external create, update, publish, or delete is `forward_recovery`. It must +start through the connector mutation recovery contract with a stable +idempotency key, SHA-256 request digest, source revision/cursor, and dry-run +evidence. Definitive rejection is terminal. A timeout or lost acknowledgement +after dispatch is `outcome_unknown` and blocks replay until the owning connector +verifies provider state. The contract and conformance tests exist; no current +connector advertises a production external mutation, so write/delete adoption +remains explicitly planned rather than implied. + ## Provider Declaration An executable connector type should publish machine-readable metadata for: diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py index 319e8c4..cf5a8af 100644 --- a/src/govoplan_connectors/backend/manifest.py +++ b/src/govoplan_connectors/backend/manifest.py @@ -88,6 +88,11 @@ ARCHITECTURE = ModuleArchitectureDeclaration( reference="tests/test_sanctions_sources.py", summary="Exercises source acquisition health, checksums, retries, and immutable evidence.", ), + ModuleMaturityEvidence( + kind="recovery", + reference="tests/test_recovery.py", + summary="Proves atomic snapshot commits, idempotent replay, distributed fences, tamper rejection, and unknown external-effect handling.", + ), ModuleMaturityEvidence( kind="documentation", reference="docs/CONNECTOR_SOURCE_LIFECYCLE.md", @@ -149,8 +154,13 @@ EXTERNAL_PROVIDERS = ( freshness="Snapshot acquisition time and source timestamp are exposed.", health="Import validation and source-read failures are explicit.", max_read_items=1000, + idempotency="Feed imports accept a caller request key and replay the same committed immutable source without refetching.", + retry="Read-only acquisition may be retried only as a new deliberate request after a failed atomic operation.", + outcome_unknown="Provider reads do not mutate remote state; an uncertain database commit is resolved by the atomic recovery transaction.", + outcome_unknown_supported=False, evidence="Rows, schema, fingerprint, source metadata, and acquisition provenance remain linked.", correction="Import a replacement snapshot; retain the prior snapshot as evidence.", + rollback="Snapshot rows and the terminal recovery checkpoint commit or roll back together.", reconciliation="Compare source and snapshot fingerprints before selecting a new current state.", outage="Existing snapshots remain available and visibly stale; no live-source claim is made.", classifications=("internal", "confidential", "restricted"), @@ -187,12 +197,18 @@ EXTERNAL_PROVIDERS = ( ), behavior=ProviderBehaviorDeclaration( revision_tokens="Provider source version, ETag, Last-Modified, and SHA-256 digest are retained when available.", - concurrency="Refreshes use conditional source requests and create immutable snapshots.", + concurrency="Refreshes use conditional source requests, a distributed per-tenant/provider fence, and immutable snapshots.", freshness="Latest successful acquisition, source timestamp, and stale health are reported.", health="Transport, parsing, source-change, and malformed-source states are explicit.", max_read_items=5000, + idempotency="A caller request key identifies one acquisition run and replays its committed result without contacting the source again.", + retry="Bounded HTTP retries are safe because acquisition is read-only; failed runs require a new deliberate request key.", + timeout_seconds=30, + outcome_unknown="The external operation is read-only; snapshot rows and recovery evidence commit atomically.", + outcome_unknown_supported=False, evidence="Raw source bytes, checksum, acquisition run, parser result, and normalized entry count are linked.", correction="A corrected source creates a new immutable snapshot and acquisition run.", + rollback="A failed database transaction leaves no snapshot and the stale atomic fence resolves as failed.", reconciliation="Compare source version and digest, then preserve both prior and corrected evidence.", outage="The latest accepted snapshot stays usable with stale/unavailable source health.", classifications=("public", "internal"), @@ -494,7 +510,11 @@ manifest = ModuleManifest( "the official United Nations Security Council consolidated " "XML source. Each fetch records conditional transport " "evidence, bounded retries, health state, source metadata, " - "raw evidence, and a SHA-256 checksum. Risk Compliance owns " + "raw evidence, and a SHA-256 checksum. Refreshes acquire a " + "distributed recovery fence before provider I/O; the immutable " + "snapshot and terminal recovery checkpoint then commit in one " + "transaction. A repeated request key returns the same result. " + "Risk Compliance owns " "normalization, matching, legal review, and dispositions." ), layer="available", diff --git a/src/govoplan_connectors/backend/recovery.py b/src/govoplan_connectors/backend/recovery.py new file mode 100644 index 0000000..70f2a6a --- /dev/null +++ b/src/govoplan_connectors/backend/recovery.py @@ -0,0 +1,381 @@ +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +from typing import Any +from uuid import NAMESPACE_URL, uuid4, uuid5 + +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_core.core.recovery import ( + RecoveryGuaranteeError, + RecoveryMode, + RecoveryPlan, + RecoveryStatus, +) +from govoplan_core.core.recovery_runtime import ( + DurableRecoveryOperation, + RecoveryOperationBusy, + RecoveryOperationStateConflict, + begin_durable_recovery_operation, +) +from govoplan_core.core.runtime_coordination import process_runtime_identity + + +class ConnectorRecoveryError(RuntimeError): + pass + + +@dataclass(frozen=True, slots=True) +class ConnectorRecoveryDeclaration: + operation_type: str + mode: RecoveryMode + provider_mutation: bool + idempotency: str + verification: tuple[str, ...] + recovery: tuple[str, ...] + implemented: bool + + +CONNECTOR_RECOVERY_OPERATIONS = ( + ConnectorRecoveryDeclaration( + operation_type="read-snapshot", + mode=RecoveryMode.ATOMIC, + provider_mutation=False, + idempotency=( + "Caller-supplied request keys replay a committed immutable snapshot; " + "otherwise each deliberate acquisition receives a generated key." + ), + verification=( + "provider revision or conditional cursor is recorded before fetch", + "domain snapshot and terminal recovery checkpoint commit together", + "stored bytes and provider evidence are checksum verified", + ), + recovery=( + "a stale running transaction is failed after its database transaction rolls back", + "a new deliberate acquisition may then use a new request key", + ), + implemented=True, + ), + ConnectorRecoveryDeclaration( + operation_type="external-mutation", + mode=RecoveryMode.FORWARD_RECOVERY, + provider_mutation=True, + idempotency="A stable caller key and canonical request digest are mandatory.", + verification=( + "record the remote revision and bounded provider result", + "verify the provider state before reporting success", + ), + recovery=( + "unknown outcomes remain unresolved until provider-backed reconciliation", + "never retry the same remote effect solely to reconstruct local state", + ), + implemented=False, + ), +) + + +def connector_session_factory(session: Session) -> sessionmaker[Session]: + bind = session.get_bind() + if bind is None: + raise ConnectorRecoveryError("Connector recovery requires a bound database session") + return sessionmaker(bind=bind, expire_on_commit=False) + + +def _digest(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _clean_key(value: str | None) -> str: + clean = str(value or "").strip() + if clean and len(clean) > 500: + raise ConnectorRecoveryError("Connector idempotency keys are limited to 500 characters") + return clean or str(uuid4()) + + +def _stable_resource_id( + *, + tenant_id: str, + provider_id: str, + operation_type: str, + request_key: str, +) -> str: + return str( + uuid5( + NAMESPACE_URL, + f"govoplan:{tenant_id}:{provider_id}:{operation_type}:{request_key}", + ) + ) + + +@dataclass(slots=True) +class ConnectorReadSnapshotRecovery: + operation: DurableRecoveryOperation | None + operation_id: str + request_key: str + resource_id: str + replayed: bool + + def commit_success(self, session: Session, *, evidence: dict[str, Any]) -> None: + if self.operation is None: + raise ConnectorRecoveryError("A replayed connector read cannot be committed again") + try: + self.operation.commit_atomic_success(session, evidence=evidence) + except Exception as exc: + raise ConnectorRecoveryError( + "The connector snapshot and recovery evidence did not commit atomically" + ) from exc + + def commit_failure( + self, + session: Session, + *, + summary: str, + evidence: dict[str, Any], + ) -> None: + if self.operation is None: + raise ConnectorRecoveryError("A replayed connector read cannot be failed again") + try: + self.operation.commit_atomic_failure( + session, + summary=summary, + evidence=evidence, + ) + except Exception as exc: + raise ConnectorRecoveryError( + "The connector failure evidence did not commit atomically" + ) from exc + + def fail_without_projection( + self, + *, + summary: str, + code: str, + ) -> None: + if self.operation is None: + return + self.operation.fail( + summary=summary, + evidence={ + "verified": True, + "checks": { + "provider_mutation": False, + "projection_committed": False, + "failure_code": code, + }, + }, + ) + + +def begin_connector_read_snapshot( + session: Session, + *, + tenant_id: str, + provider_id: str, + idempotency_key: str | None, + source_revision: str | None, + cursor: str | None, + dry_run_evidence: dict[str, Any], + request_metadata: dict[str, Any] | None = None, + resource_type: str = "connector_sync_run", +) -> ConnectorReadSnapshotRecovery: + request_key = _clean_key(idempotency_key) + resource_id = _stable_resource_id( + tenant_id=tenant_id, + provider_id=provider_id, + operation_type="read-snapshot", + request_key=request_key, + ) + request = { + "tenant_id": tenant_id, + "provider_id": provider_id, + "dry_run": dry_run_evidence, + "request_key_sha256": _digest(request_key), + **dict(request_metadata or {}), + } + try: + started = begin_durable_recovery_operation( + connector_session_factory(session), + identity=process_runtime_identity(), + module_id="connectors", + operation_type="read-snapshot", + idempotency_key=f"connector-read:{_digest(f'{tenant_id}:{provider_id}:{request_key}')}", + request=request, + recovery_plan=RecoveryPlan( + mode=RecoveryMode.ATOMIC, + preconditions=( + "the actor is authorized for the connector source", + "the provider request is read-only", + "the source revision, cursor, and dry-run decision are durable", + ), + verification_steps=( + "validate the bounded provider response and source revision", + "commit the immutable snapshot and terminal checkpoint atomically", + "compare the stored content digest with the acquired bytes", + ), + ), + precondition_evidence={ + "provider_id": provider_id, + "source_revision": source_revision, + "cursor_sha256": _digest(cursor) if cursor else None, + "dry_run": dry_run_evidence, + "provider_mutation": False, + }, + lease_resource_key=f"connectors:read:{tenant_id}:{_digest(provider_id)[:40]}", + lease_ttl_seconds=15 * 60, + resource_type=resource_type, + resource_id=resource_id, + metadata={ + "resources": ["postgresql", "external-provider"], + "provider_mutation": False, + "recovery_declaration": "read-snapshot", + }, + ) + except RecoveryOperationBusy as exc: + raise ConnectorRecoveryError( + "Another runtime is already acquiring this connector source" + ) from exc + except RecoveryOperationStateConflict as exc: + raise ConnectorRecoveryError( + "This connector request is active or unresolved; reconcile it before retrying" + ) from exc + except (RecoveryGuaranteeError, RuntimeError) as exc: + raise ConnectorRecoveryError( + "The connector recovery ledger is unavailable; the provider was not contacted" + ) from exc + return ConnectorReadSnapshotRecovery( + operation=started.operation, + operation_id=started.operation_id, + request_key=request_key, + resource_id=resource_id, + replayed=started.replayed, + ) + + +@dataclass(slots=True) +class ConnectorExternalMutationRecovery: + operation: DurableRecoveryOperation | None + operation_id: str + replayed: bool + + def succeed(self, *, provider_evidence: dict[str, Any]) -> None: + if self.operation is not None: + self.operation.succeed(evidence=provider_evidence) + + def reject(self, *, summary: str, provider_code: str) -> None: + if self.operation is not None: + self.operation.reject( + summary=summary, + evidence={ + "verified": True, + "checks": {"provider_rejection": provider_code}, + }, + ) + + def outcome_unknown(self, *, summary: str, provider_code: str) -> None: + if self.operation is not None: + self.operation.unresolved( + status=RecoveryStatus.OUTCOME_UNKNOWN, + summary=summary, + evidence={"effect_started": True, "provider_code": provider_code}, + failure_summary="Inspect provider state before any retry", + ) + + +def begin_connector_external_mutation( + session: Session, + *, + tenant_id: str, + provider_id: str, + idempotency_key: str, + request_sha256: str, + source_revision: str | None, + cursor: str | None, + dry_run_evidence: dict[str, Any], + resource_type: str, + resource_id: str, +) -> ConnectorExternalMutationRecovery: + if not str(idempotency_key or "").strip(): + raise ConnectorRecoveryError("External connector mutations require an idempotency key") + request_key = _clean_key(idempotency_key) + if len(request_sha256) != 64 or any( + character not in "0123456789abcdefABCDEF" for character in request_sha256 + ): + raise ConnectorRecoveryError("External connector mutations require a SHA-256 request digest") + try: + started = begin_durable_recovery_operation( + connector_session_factory(session), + identity=process_runtime_identity(), + module_id="connectors", + operation_type="external-mutation", + idempotency_key=f"connector-write:{_digest(f'{tenant_id}:{provider_id}:{request_key}')}", + request={ + "tenant_id": tenant_id, + "provider_id": provider_id, + "request_sha256": request_sha256, + "source_revision": source_revision, + "cursor_sha256": _digest(cursor) if cursor else None, + "dry_run": dry_run_evidence, + }, + recovery_plan=RecoveryPlan( + mode=RecoveryMode.FORWARD_RECOVERY, + preconditions=( + "the actor and effective connector policy authorize the mutation", + "a stable idempotency key and canonical request digest are present", + "the dry-run and source revision evidence are durable", + ), + forward_recovery_steps=( + "inspect provider state without repeating the mutation", + "record whether the provider accepted the requested revision", + "retry only under a new deliberate key when absence is proven", + ), + verification_steps=( + "compare provider identity and revision with the canonical request", + "verify the consuming domain state independently", + ), + ), + precondition_evidence={ + "request_sha256": request_sha256, + "source_revision": source_revision, + "cursor_sha256": _digest(cursor) if cursor else None, + "dry_run": dry_run_evidence, + "provider_mutation": True, + }, + lease_resource_key=( + f"connectors:write:{tenant_id}:{_digest(provider_id)[:24]}:" + f"{_digest(resource_id)[:24]}" + ), + lease_ttl_seconds=15 * 60, + resource_type=resource_type, + resource_id=resource_id, + metadata={ + "resources": ["postgresql", "queue", "external-provider"], + "provider_mutation": True, + "recovery_declaration": "external-mutation", + }, + ) + except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: + raise ConnectorRecoveryError( + "This external connector effect is active or unresolved" + ) from exc + except (RecoveryGuaranteeError, RuntimeError) as exc: + raise ConnectorRecoveryError( + "The connector recovery ledger is unavailable; no external mutation started" + ) from exc + return ConnectorExternalMutationRecovery( + operation=started.operation, + operation_id=started.operation_id, + replayed=started.replayed, + ) + + +__all__ = [ + "CONNECTOR_RECOVERY_OPERATIONS", + "ConnectorExternalMutationRecovery", + "ConnectorReadSnapshotRecovery", + "ConnectorRecoveryDeclaration", + "ConnectorRecoveryError", + "begin_connector_external_mutation", + "begin_connector_read_snapshot", + "connector_session_factory", +] diff --git a/src/govoplan_connectors/backend/router.py b/src/govoplan_connectors/backend/router.py index 4ab0bd0..c0d1af8 100644 --- a/src/govoplan_connectors/backend/router.py +++ b/src/govoplan_connectors/backend/router.py @@ -1,8 +1,10 @@ from __future__ import annotations from dataclasses import asdict +import hashlib +from typing import Annotated -from fastapi import APIRouter, Depends, HTTPException, Query, Response, status +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, status from sqlalchemy.orm import Session from govoplan_core.audit.logging import audit_event @@ -42,6 +44,10 @@ from govoplan_connectors.backend.schemas import ( TabularSourceResponse, ) from govoplan_connectors.backend.feeds import ConnectorFeedProvider, feed_rows +from govoplan_connectors.backend.recovery import ( + ConnectorRecoveryError, + begin_connector_read_snapshot, +) from govoplan_connectors.backend.sanctions_sources import ( SANCTIONS_READ_SCOPE, SANCTIONS_REFRESH_SCOPE, @@ -108,6 +114,18 @@ def _feed_http_error(exc: FeedCapabilityError) -> HTTPException: ) +def _recovery_http_error(exc: ConnectorRecoveryError) -> HTTPException: + detail = str(exc) + return HTTPException( + status_code=( + status.HTTP_409_CONFLICT + if "already" in detail.casefold() or "active" in detail.casefold() + else status.HTTP_503_SERVICE_UNAVAILABLE + ), + detail=detail, + ) + + @router.post("/feeds/preview", response_model=FeedDocumentResponse) def api_preview_feed( payload: FeedAcquireRequest, @@ -133,8 +151,45 @@ def api_import_feed_snapshot( payload: FeedImportRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), + idempotency_key: Annotated[ + str | None, + Header(alias="Idempotency-Key", max_length=500), + ] = None, ) -> TabularSourceResponse: _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + try: + recovery = begin_connector_read_snapshot( + session, + tenant_id=principal.tenant_id, + provider_id="connectors.feed_snapshot", + idempotency_key=idempotency_key, + source_revision=None, + cursor=None, + dry_run_evidence={ + "performed": False, + "reason": "read-only acquisition into an immutable snapshot", + }, + request_metadata={ + "source_url_sha256": hashlib.sha256( + payload.url.encode("utf-8") + ).hexdigest(), + "source_name": payload.source_name, + "max_entries": payload.max_entries, + }, + resource_type="connector_tabular_source", + ) + except ConnectorRecoveryError as exc: + raise _recovery_http_error(exc) from exc + if recovery.replayed: + try: + source = provider.get_source( + session, + principal, + source_ref=f"snapshot:{recovery.resource_id}", + ) + except TabularSourceError as exc: + raise _http_error(exc) from exc + return _source_response(source) try: document = feed_transport.fetch( payload.url, @@ -170,8 +225,14 @@ def api_import_feed_snapshot( }, }, ), + source_id=recovery.resource_id, ) except (FeedCapabilityError, TabularSourceError) as exc: + session.rollback() + recovery.fail_without_projection( + summary="The read-only feed import failed before a snapshot committed", + code=exc.__class__.__name__, + ) if isinstance(exc, FeedCapabilityError): raise _feed_http_error(exc) from exc raise _http_error(exc) from exc @@ -190,7 +251,22 @@ def api_import_feed_snapshot( "row_count": source.row_count, }, ) - session.commit() + try: + recovery.commit_success( + session, + evidence={ + "verified": True, + "checks": { + "snapshot_ref": source.ref, + "snapshot_fingerprint": source.fingerprint, + "feed_sha256": document.sha256, + "row_count": source.row_count, + "provider_mutation": False, + }, + }, + ) + except ConnectorRecoveryError as exc: + raise _recovery_http_error(exc) from exc return _source_response(source) @@ -390,6 +466,10 @@ def api_refresh_sanctions_source( provider_id: str, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), + idempotency_key: Annotated[ + str | None, + Header(alias="Idempotency-Key", max_length=500), + ] = None, ) -> SanctionsRefreshResponse: _require_any_scope( principal, @@ -401,7 +481,10 @@ def api_refresh_sanctions_source( session, principal, provider_id=provider_id, + idempotency_key=idempotency_key, ) + except ConnectorRecoveryError as exc: + raise _recovery_http_error(exc) from exc except SanctionsSourceError as exc: raise _sanctions_http_error(exc) from exc audit_event( diff --git a/src/govoplan_connectors/backend/sanctions_sources.py b/src/govoplan_connectors/backend/sanctions_sources.py index ab81001..181ec93 100644 --- a/src/govoplan_connectors/backend/sanctions_sources.py +++ b/src/govoplan_connectors/backend/sanctions_sources.py @@ -26,6 +26,11 @@ from govoplan_connectors.backend.db.models import ( ConnectorSanctionsAcquisitionRun, ConnectorSanctionsSnapshot, ) +from govoplan_connectors.backend.recovery import ( + ConnectorRecoveryError, + begin_connector_read_snapshot, + connector_session_factory, +) from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.core.sanctions import ( SanctionsSnapshotPayload, @@ -320,8 +325,9 @@ class SqlSanctionsSnapshotProvider: principal: object, *, provider_id: str, + idempotency_key: str | None = None, ) -> SanctionsAcquisitionResult: - db, api_principal = _context( + caller_db, api_principal = _context( session, principal, SANCTIONS_REFRESH_SCOPE, @@ -331,23 +337,101 @@ class SqlSanctionsSnapshotProvider: raise SanctionsSourceNotFoundError( "Sanctions source provider is not available." ) - now = utcnow() - latest = db.scalar( - select(ConnectorSanctionsSnapshot) - .where( - ConnectorSanctionsSnapshot.tenant_id - == api_principal.tenant_id, - ConnectorSanctionsSnapshot.provider_id - == provider_id, + factory = connector_session_factory(caller_db) + with factory() as preflight_db: + latest = _latest_snapshot( + preflight_db, + tenant_id=api_principal.tenant_id, + provider_id=provider_id, ) - .order_by( - ConnectorSanctionsSnapshot.acquired_at.desc(), - ConnectorSanctionsSnapshot.id.desc(), - ) - .limit(1) - ) conditional_headers = _conditional_headers(latest) + source_revision = latest.source_version if latest is not None else None + cursor = "\n".join( + f"{key.casefold()}:{value}" + for key, value in sorted(conditional_headers.items()) + ) or None + recovery = begin_connector_read_snapshot( + caller_db, + tenant_id=api_principal.tenant_id, + provider_id=provider_id, + idempotency_key=idempotency_key, + source_revision=source_revision, + cursor=cursor, + dry_run_evidence={ + "performed": False, + "reason": "read-only immutable source acquisition", + }, + request_metadata={ + "source_id": definition.source_id, + "subject_data_transmitted": False, + }, + ) + if recovery.replayed: + with factory() as replay_db: + replayed = _acquisition_result( + replay_db, + run_id=recovery.resource_id, + tenant_id=api_principal.tenant_id, + ) + if replayed is None: + raise ConnectorRecoveryError( + "The completed connector acquisition has no matching durable result" + ) + return replayed + + with factory() as work_db: + result = self._refresh_source_transaction( + work_db, + api_principal, + definition=definition, + provider_id=provider_id, + run_id=recovery.resource_id, + latest=latest, + conditional_headers=conditional_headers, + recovery_operation_id=recovery.operation_id, + request_key_sha256=hashlib.sha256( + recovery.request_key.encode("utf-8") + ).hexdigest(), + ) + evidence = _acquisition_recovery_evidence( + work_db, + result=result, + tenant_id=api_principal.tenant_id, + ) + try: + if result.status in {"succeeded", "not_modified"}: + recovery.commit_success(work_db, evidence=evidence) + else: + recovery.commit_failure( + work_db, + summary=( + "The read-only connector acquisition failed without " + "mutating the provider" + ), + evidence=evidence, + ) + except Exception as exc: + raise ConnectorRecoveryError( + "Connector acquisition state could not be committed with its recovery evidence" + ) from exc + return result + + def _refresh_source_transaction( + self, + db: Session, + api_principal: ApiPrincipal, + *, + definition: SanctionsSourceDefinition, + provider_id: str, + run_id: str, + latest: ConnectorSanctionsSnapshot | None, + conditional_headers: Mapping[str, str], + recovery_operation_id: str, + request_key_sha256: str, + ) -> SanctionsAcquisitionResult: + now = utcnow() run = ConnectorSanctionsAcquisitionRun( + id=run_id, tenant_id=api_principal.tenant_id, provider_id=provider_id, source_id=definition.source_id, @@ -363,6 +447,8 @@ class SqlSanctionsSnapshotProvider: conditional_headers ), "subject_data_transmitted": False, + "recovery_operation_id": recovery_operation_id, + "request_key_sha256": request_key_sha256, }, response_evidence={}, started_at=now, @@ -727,6 +813,95 @@ def _conditional_headers( return result +def _latest_snapshot( + session: Session, + *, + tenant_id: str, + provider_id: str, +) -> ConnectorSanctionsSnapshot | None: + return session.scalar( + select(ConnectorSanctionsSnapshot) + .where( + ConnectorSanctionsSnapshot.tenant_id == tenant_id, + ConnectorSanctionsSnapshot.provider_id == provider_id, + ) + .order_by( + ConnectorSanctionsSnapshot.acquired_at.desc(), + ConnectorSanctionsSnapshot.id.desc(), + ) + .limit(1) + ) + + +def _acquisition_result( + session: Session, + *, + run_id: str, + tenant_id: str, +) -> SanctionsAcquisitionResult | None: + run = session.scalar( + select(ConnectorSanctionsAcquisitionRun).where( + ConnectorSanctionsAcquisitionRun.id == run_id, + ConnectorSanctionsAcquisitionRun.tenant_id == tenant_id, + ) + ) + if run is None: + return None + snapshot = ( + session.get(ConnectorSanctionsSnapshot, run.snapshot_id) + if run.snapshot_id + else None + ) + return SanctionsAcquisitionResult( + run_id=run.id, + provider_id=run.provider_id, + status=run.status, + snapshot=_snapshot_dto(snapshot) if snapshot is not None else None, + error=run.error, + ) + + +def _acquisition_recovery_evidence( + session: Session, + *, + result: SanctionsAcquisitionResult, + tenant_id: str, +) -> dict[str, object]: + run = session.get(ConnectorSanctionsAcquisitionRun, result.run_id) + snapshot = ( + session.get(ConnectorSanctionsSnapshot, run.snapshot_id) + if run is not None and run.snapshot_id + else None + ) + checksum_matches = bool( + snapshot is None + or hashlib.sha256(snapshot.raw_content).hexdigest() == snapshot.sha256 + ) + successful = result.status in {"succeeded", "not_modified"} + verified = bool( + run is not None + and run.tenant_id == tenant_id + and run.status == result.status + and run.finished_at is not None + and checksum_matches + and (not successful or snapshot is not None) + ) + return { + "verified": verified, + "checks": { + "run_present": run is not None, + "run_status": result.status, + "run_finished": bool(run is not None and run.finished_at is not None), + "snapshot_present": snapshot is not None, + "snapshot_checksum_matches": checksum_matches, + "provider_mutation": False, + }, + "run_id": result.run_id, + "snapshot_sha256": snapshot.sha256 if snapshot is not None else None, + "source_version": snapshot.source_version if snapshot is not None else None, + } + + def _snapshot_record( session: Session, *, diff --git a/src/govoplan_connectors/backend/tabular_sources.py b/src/govoplan_connectors/backend/tabular_sources.py index eb72af9..a8842c5 100644 --- a/src/govoplan_connectors/backend/tabular_sources.py +++ b/src/govoplan_connectors/backend/tabular_sources.py @@ -130,6 +130,7 @@ class SqlTabularSourceProvider: principal: object, *, snapshot: TabularSnapshotInput, + source_id: str | None = None, ) -> TabularSource: db, api_principal = _context(session, principal, WRITE_SCOPE) name = snapshot.name.strip() @@ -178,6 +179,8 @@ class SqlTabularSourceProvider: created_by=actor_id, updated_by=actor_id, ) + if source_id: + item.id = source_id db.add(item) db.flush() return _source_dto(item) @@ -296,6 +299,7 @@ def _context( if required_scope == READ_SCOPE: accepted_scopes.update( { + WRITE_SCOPE, "datasources:catalogue:read", "datasources:source:write", "datasources:source:admin", diff --git a/tests/test_recovery.py b/tests/test_recovery.py new file mode 100644 index 0000000..a2aabdd --- /dev/null +++ b/tests/test_recovery.py @@ -0,0 +1,263 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +import unittest +from unittest.mock import patch + +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session + +from govoplan_connectors.backend.db.models import ConnectorTabularSource +from govoplan_connectors.backend.feeds import ConnectorFeedProvider +from govoplan_connectors.backend.recovery import ( + CONNECTOR_RECOVERY_OPERATIONS, + ConnectorRecoveryError, + begin_connector_external_mutation, + begin_connector_read_snapshot, +) +from govoplan_connectors.backend.router import api_import_feed_snapshot +from govoplan_connectors.backend.schemas import FeedImportRequest +from govoplan_connectors.backend.tabular_sources import WRITE_SCOPE +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.recovery import ( + RecoveryCheckpoint, + RecoveryOperation, + RecoveryStatus, +) +from govoplan_core.core.recovery_runtime import ( + RecoveryOperationStateConflict, + claim_durable_recovery_operation, +) +from govoplan_core.core.tabular_sources import TabularSnapshotInput +from govoplan_core.core.runtime_coordination import ( + DistributedLease, + RuntimeIdentity, + bind_process_runtime_identity, +) +from govoplan_core.db.base import Base +from govoplan_connectors.backend.tabular_sources import SqlTabularSourceProvider + + +RSS = b""" +Updates +https://example.test/Updates +1One""" + + +def _identity(node: str, incarnation: str) -> RuntimeIdentity: + return RuntimeIdentity( + installation_id="connector-recovery-tests", + node_id=node, + incarnation=incarnation, + role="worker", + software_version="test", + composition_hash="a" * 64, + ) + + +def _principal() -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id="tenant-1", + scopes=frozenset({WRITE_SCOPE}), + ), + account=object(), + user=object(), + ) + + +class ConnectorRecoveryTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=( + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, + ConnectorTabularSource.__table__, + ), + ) + self.session = Session(self.engine, expire_on_commit=False) + bind_process_runtime_identity(_identity("node-1", "incarnation-1")) + + def tearDown(self) -> None: + bind_process_runtime_identity(None) + self.session.close() + self.engine.dispose() + + def test_feed_snapshot_and_recovery_checkpoint_commit_atomically_and_replay(self) -> None: + document = ConnectorFeedProvider().parse( + RSS, + source_url="https://example.test/feed.xml", + ) + payload = FeedImportRequest( + url="https://example.test/feed.xml", + name="Updates", + source_name="updates", + ) + with ( + patch( + "govoplan_connectors.backend.router.feed_transport.fetch", + return_value=document, + ) as fetch, + patch("govoplan_connectors.backend.router.audit_event"), + ): + first = api_import_feed_snapshot( + payload, + session=self.session, + principal=_principal(), + idempotency_key="feed-import-1", + ) + replay = api_import_feed_snapshot( + payload, + session=self.session, + principal=_principal(), + idempotency_key="feed-import-1", + ) + + self.assertEqual(first.ref, replay.ref) + fetch.assert_called_once() + operation = self.session.scalar(select(RecoveryOperation)) + assert operation is not None + self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status) + self.assertEqual( + first.ref.removeprefix("snapshot:"), + operation.resource_id, + ) + + def test_recovery_metadata_distinguishes_reads_from_external_mutations(self) -> None: + declarations = { + item.operation_type: item for item in CONNECTOR_RECOVERY_OPERATIONS + } + + self.assertFalse(declarations["read-snapshot"].provider_mutation) + self.assertTrue(declarations["read-snapshot"].implemented) + self.assertTrue(declarations["external-mutation"].provider_mutation) + self.assertFalse(declarations["external-mutation"].implemented) + + def test_stale_atomic_connector_fence_fails_without_claiming_an_effect(self) -> None: + recovery = begin_connector_read_snapshot( + self.session, + tenant_id="tenant-1", + provider_id="provider-1", + idempotency_key="read-1", + source_revision="revision-1", + cursor="cursor-1", + dry_run_evidence={"performed": True, "approved": True}, + ) + lease = self.session.scalar(select(DistributedLease)) + assert lease is not None + lease.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1) + self.session.commit() + bind_process_runtime_identity(_identity("node-2", "incarnation-2")) + + with self.assertRaises(RecoveryOperationStateConflict): + claim_durable_recovery_operation( + recovery.operation.session_factory, + identity=_identity("node-2", "incarnation-2"), + operation_id=recovery.operation_id, + ) + + operation = self.session.get(RecoveryOperation, recovery.operation_id) + self.session.refresh(operation) + self.assertEqual(RecoveryStatus.FAILED.value, operation.status) + + def test_external_mutation_unknown_outcome_blocks_blind_retry(self) -> None: + kwargs = { + "tenant_id": "tenant-1", + "provider_id": "provider-1", + "idempotency_key": "publish-1", + "request_sha256": "b" * 64, + "source_revision": "revision-1", + "cursor": None, + "dry_run_evidence": {"performed": True, "approved": True}, + "resource_type": "external_record", + "resource_id": "record-1", + } + recovery = begin_connector_external_mutation(self.session, **kwargs) + recovery.outcome_unknown( + summary="The provider connection closed after dispatch", + provider_code="connection_closed", + ) + + with self.assertRaises(ConnectorRecoveryError): + begin_connector_external_mutation(self.session, **kwargs) + operation = self.session.get(RecoveryOperation, recovery.operation_id) + self.session.refresh(operation) + self.assertEqual(RecoveryStatus.OUTCOME_UNKNOWN.value, operation.status) + + def test_tampered_chain_rolls_back_the_atomic_connector_projection(self) -> None: + recovery = begin_connector_read_snapshot( + self.session, + tenant_id="tenant-1", + provider_id="provider-1", + idempotency_key="tampered-read", + source_revision=None, + cursor=None, + dry_run_evidence={"performed": False, "reason": "read-only"}, + ) + checkpoint = self.session.scalar( + select(RecoveryCheckpoint) + .where(RecoveryCheckpoint.operation_id == recovery.operation_id) + .order_by(RecoveryCheckpoint.sequence) + .limit(1) + ) + assert checkpoint is not None + checkpoint.summary = "tampered" + self.session.commit() + source = SqlTabularSourceProvider().create_snapshot( + self.session, + _principal(), + snapshot=TabularSnapshotInput( + name="Tampered", + source_name="tampered", + rows=({"id": 1},), + ), + source_id=recovery.resource_id, + ) + + with self.assertRaises(ConnectorRecoveryError): + recovery.commit_success( + self.session, + evidence={ + "verified": True, + "checks": {"snapshot_ref": source.ref}, + }, + ) + + self.assertIsNone( + self.session.get(ConnectorTabularSource, recovery.resource_id) + ) + operation = self.session.get(RecoveryOperation, recovery.operation_id) + self.session.refresh(operation) + self.assertEqual(RecoveryStatus.RUNNING.value, operation.status) + + def test_definitive_external_rejection_is_terminal(self) -> None: + recovery = begin_connector_external_mutation( + self.session, + tenant_id="tenant-1", + provider_id="provider-1", + idempotency_key="publish-rejected", + request_sha256="c" * 64, + source_revision="revision-1", + cursor=None, + dry_run_evidence={"performed": True, "approved": True}, + resource_type="external_record", + resource_id="record-2", + ) + recovery.reject( + summary="The provider rejected the requested revision", + provider_code="revision_conflict", + ) + + operation = self.session.get(RecoveryOperation, recovery.operation_id) + self.session.refresh(operation) + self.assertEqual(RecoveryStatus.REJECTED.value, operation.status) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sanctions_sources.py b/tests/test_sanctions_sources.py index bb32009..70bf430 100644 --- a/tests/test_sanctions_sources.py +++ b/tests/test_sanctions_sources.py @@ -27,6 +27,16 @@ from govoplan_connectors.backend.sanctions_sources import ( ) from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.recovery import ( + RecoveryCheckpoint, + RecoveryOperation, + RecoveryStatus, +) +from govoplan_core.core.runtime_coordination import ( + DistributedLease, + RuntimeIdentity, + bind_process_runtime_identity, +) from govoplan_core.db.base import Base, utcnow @@ -89,13 +99,27 @@ class SanctionsSourcesTests(unittest.TestCase): Base.metadata.create_all( self.engine, tables=( + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, ConnectorSanctionsAcquisitionRun.__table__, ConnectorSanctionsSnapshot.__table__, ), ) self.session = Session(self.engine) + bind_process_runtime_identity( + RuntimeIdentity( + installation_id="connectors-tests", + node_id="connectors-test-node", + incarnation="connectors-test-incarnation", + role="worker", + software_version="test", + composition_hash="a" * 64, + ) + ) def tearDown(self) -> None: + bind_process_runtime_identity(None) self.session.close() self.engine.dispose() @@ -234,6 +258,7 @@ class SanctionsSourcesTests(unittest.TestCase): first.snapshot.ref.removeprefix("sanctions-snapshot:"), ) record.acquired_at = utcnow() - timedelta(days=3) + self.session.commit() result = provider.refresh_source( self.session, @@ -244,6 +269,48 @@ class SanctionsSourcesTests(unittest.TestCase): self.assertEqual("stale", result.status) self.assertEqual(first.snapshot.ref, result.snapshot.ref) + def test_idempotent_refresh_replays_the_committed_acquisition(self) -> None: + transport = _Transport((response(),)) + provider = SqlSanctionsSnapshotProvider(transport) + + first = provider.refresh_source( + self.session, + principal(), + provider_id=UNSC_PROVIDER_ID, + idempotency_key="scheduled-refresh-1", + ) + replay = provider.refresh_source( + self.session, + principal(), + provider_id=UNSC_PROVIDER_ID, + idempotency_key="scheduled-refresh-1", + ) + + self.assertEqual(first.run_id, replay.run_id) + self.assertEqual(first.snapshot.ref, replay.snapshot.ref) + self.assertEqual([], transport.responses) + operation = self.session.query(RecoveryOperation).one() + self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status) + + def test_provider_failure_commits_failed_run_and_terminal_recovery(self) -> None: + provider = SqlSanctionsSnapshotProvider( + _Transport((SanctionsSourceError("offline"),)) + ) + + result = provider.refresh_source( + self.session, + principal(), + provider_id=UNSC_PROVIDER_ID, + ) + + self.assertEqual("unavailable", result.status) + operation = self.session.query(RecoveryOperation).one() + self.assertEqual(RecoveryStatus.FAILED.value, operation.status) + self.assertEqual( + result.run_id, + self.session.query(ConnectorSanctionsAcquisitionRun).one().id, + ) + def test_snapshot_access_is_tenant_and_scope_isolated(self) -> None: provider = SqlSanctionsSnapshotProvider() created = provider.refresh_source(