382 lines
13 KiB
Python
382 lines
13 KiB
Python
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",
|
|
]
|