from __future__ import annotations from dataclasses import dataclass from datetime import UTC, datetime import hashlib from typing import Protocol from sqlalchemy import event from sqlalchemy.orm import Session 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 from govoplan_core.db.session import get_database from govoplan_files.backend.db.models import ( FileBlob, FileIntegrityFinding, FileIntegrityScan, ) from govoplan_files.backend.storage.backends import ( StorageBackend, StorageBackendError, StorageObjectMissing, ) from govoplan_files.backend.storage.common import FileStorageError _PENDING_EFFECTS_KEY = "govoplan_files_pending_recovery_effects" _HOOKS_INSTALLED_KEY = "govoplan_files_recovery_hooks_installed" _ROLLBACK_ERRORS_KEY = "govoplan_files_recovery_rollback_errors" class _PendingEffect(Protocol): def settle(self, *, committed: bool) -> None: ... @dataclass(slots=True) class PendingBlobWrite: operation: DurableRecoveryOperation backend: StorageBackend tenant_id: str blob_id: str storage_key: str semantic_checksum_sha256: str semantic_size_bytes: int protection_discriminator: str created_new: bool expected_storage_checksum_sha256: str | None = None expected_storage_size_bytes: int | None = None expected_envelope_id: str | None = None def prepare_stored_bytes( self, data: bytes, *, envelope_id: str | None, ) -> None: """Retain process-local verification evidence without recording content.""" self.expected_storage_checksum_sha256 = hashlib.sha256(data).hexdigest() self.expected_storage_size_bytes = len(data) self.expected_envelope_id = envelope_id def settle(self, *, committed: bool) -> None: evidence = _blob_write_evidence(self) if _blob_write_complete(evidence): self.operation.succeed(evidence=evidence) return if self.created_new and evidence.get("database_blob_present") is False: self._settle_unreferenced_new_object(evidence) return if not self.created_new and _object_matches(evidence): if _forward_complete_blob_repair(self): completed = _blob_write_evidence(self) if _blob_write_complete(completed): self.operation.succeed(evidence=completed) return if evidence.get("database_blob_present") is True and not _object_matches( evidence ): _quarantine_blob_after_failed_verification(self, evidence) evidence = _blob_write_evidence(self) status = ( RecoveryStatus.OUTCOME_UNKNOWN if not evidence.get("verified") else RecoveryStatus.RECOVERY_REQUIRED ) transaction = "committed" if committed else "rolled back" self.operation.unresolved( status=status, summary=f"Managed Files blob write remained unresolved after the database transaction {transaction}", evidence=evidence, failure_summary=( "The managed object and Files blob metadata require reconciliation" ), ) def _settle_unreferenced_new_object(self, evidence: dict[str, object]) -> None: object_present = evidence.get("object_present") if object_present is False: self.operation.reject( summary="The Files blob write left no durable object or database row", evidence=evidence, ) return if object_present is not True: self.operation.unresolved( status=RecoveryStatus.OUTCOME_UNKNOWN, summary="The unreferenced Files object could not be probed", evidence=evidence, failure_summary="Object storage availability prevented upload compensation", ) return try: self.backend.delete(self.storage_key) except (StorageBackendError, OSError): self.operation.unresolved( status=RecoveryStatus.RECOVERY_REQUIRED, summary="An unreferenced Files object could not be compensated", evidence=evidence, failure_summary="Delete the unreferenced managed object after verifying that no FileBlob references it", ) return recovered = _blob_write_evidence(self) if ( recovered.get("verified") is True and recovered.get("database_blob_present") is False and recovered.get("object_present") is False ): self.operation.compensate( failure_summary="The Files database transaction did not retain the new blob", failure_evidence=evidence, recovery_evidence=recovered, ) return self.operation.unresolved( status=RecoveryStatus.RECOVERY_REQUIRED, summary="Files upload compensation could not be verified", evidence=recovered, failure_summary="The unreferenced managed object requires operator reconciliation", ) @dataclass(slots=True) class PendingOrphanCleanup: operation: DurableRecoveryOperation backend: StorageBackend finding_id: str tenant_id: str storage_key: str resolved_by_user_id: str def settle(self, *, committed: bool) -> None: del committed evidence = _orphan_cleanup_evidence(self) if _orphan_cleanup_complete(evidence): self.operation.succeed(evidence=evidence) return if ( evidence.get("verified") is True and evidence.get("object_present") is True and evidence.get("finding_deleted") is False ): self.operation.reject( summary="The orphan object was retained and the cleanup finding stayed open", evidence=evidence, ) return if evidence.get("object_present") is False: if _forward_complete_orphan_finding(self): completed = _orphan_cleanup_evidence(self) if _orphan_cleanup_complete(completed): self.operation.succeed(evidence=completed) return status = ( RecoveryStatus.OUTCOME_UNKNOWN if not evidence.get("verified") else RecoveryStatus.RECOVERY_REQUIRED ) self.operation.unresolved( status=status, summary="Files orphan cleanup requires reconciliation", evidence=evidence, failure_summary="Recheck the object and integrity-finding state before another cleanup attempt", ) def begin_blob_write_recovery( session: Session, *, backend: StorageBackend, tenant_id: str, blob_id: str, storage_key: str, semantic_checksum_sha256: str, semantic_size_bytes: int, protection_discriminator: str, created_new: bool, repair_token: str | None = None, ) -> PendingBlobWrite: disposition = "create" if created_new else "repair" state_token = repair_token or blob_id key_digest = hashlib.sha256(storage_key.encode("utf-8")).hexdigest() idempotency_key = ( f"files-blob-{disposition}:{blob_id}:{state_token[:48]}" ) try: started = begin_durable_recovery_operation( get_database().SessionLocal, identity=process_runtime_identity(), module_id="files", operation_type=f"blob-{disposition}", idempotency_key=idempotency_key, request={ "tenant_id": tenant_id, "blob_id": blob_id, "storage_key": storage_key if created_new else None, "storage_key_sha256": key_digest, "semantic_checksum_sha256": semantic_checksum_sha256, "semantic_size_bytes": semantic_size_bytes, "protection_discriminator": protection_discriminator, "disposition": disposition, }, recovery_plan=RecoveryPlan( mode=( RecoveryMode.COMPENSATION if created_new else RecoveryMode.FORWARD_RECOVERY ), preconditions=( "the caller has Files write authority for the target owner", "the storage key belongs to the tenant Files namespace", "the request records digests rather than file contents", ), compensation_steps=( "verify that no FileBlob references the newly reserved key", "delete only that unreferenced key and verify absence", ) if created_new else (), forward_recovery_steps=( "verify the expected bytes at the existing blob key", "forward-complete matching integrity metadata or quarantine the blob", ) if not created_new else (), verification_steps=( "reload FileBlob metadata through an independent session", "stream and hash the managed object independently", ), ), precondition_evidence={ "blob_id": blob_id, "storage_key_sha256": key_digest, "semantic_checksum_sha256": semantic_checksum_sha256, "semantic_size_bytes": semantic_size_bytes, "created_new": created_new, }, lease_resource_key=( f"files:blob:{tenant_id}:{blob_id}" ), lease_ttl_seconds=15 * 60, resource_type="file_blob", resource_id=blob_id, metadata={ "resources": ["postgresql", "object-storage"], "storage_backend": backend.name, }, ) except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: raise FileStorageError( "This managed blob is already owned by another recovery operation" ) from exc except (RecoveryGuaranteeError, RuntimeError) as exc: raise FileStorageError( "The Files recovery ledger is unavailable; no object was written" ) from exc if started.replayed or started.operation is None: raise FileStorageError( "The matching Files blob operation was already completed; reload before retrying" ) pending = PendingBlobWrite( operation=started.operation, backend=backend, tenant_id=tenant_id, blob_id=blob_id, storage_key=storage_key, semantic_checksum_sha256=semantic_checksum_sha256, semantic_size_bytes=semantic_size_bytes, protection_discriminator=protection_discriminator, created_new=created_new, ) _register_pending_effect(session, pending) return pending def begin_orphan_cleanup_recovery( session: Session, finding: FileIntegrityFinding, *, backend: StorageBackend, user_id: str, ) -> PendingOrphanCleanup: key_digest = hashlib.sha256(finding.storage_key.encode("utf-8")).hexdigest() try: started = begin_durable_recovery_operation( get_database().SessionLocal, identity=process_runtime_identity(), module_id="files", operation_type="integrity-orphan-cleanup", idempotency_key=f"files-orphan-cleanup:{finding.id}", request={ "tenant_id": finding.tenant_id, "finding_id": finding.id, "storage_key_sha256": key_digest, }, recovery_plan=RecoveryPlan( mode=RecoveryMode.FORWARD_RECOVERY, preconditions=( "the integrity finding identifies an unreferenced object", "the object key remains inside the completed scan scope", "a fresh database reference check found no FileBlob", ), forward_recovery_steps=( "verify object absence independently", "mark the durable finding deleted only after absence is proven", ), verification_steps=( "reload the finding and its scan through an independent session", "probe the original storage key through the configured backend", ), ), precondition_evidence={ "finding_id": finding.id, "scan_id": finding.scan_id, "storage_key_sha256": key_digest, "finding_state": finding.state, }, lease_resource_key=( f"files:orphan-cleanup:{finding.tenant_id}:{key_digest[:40]}" ), lease_ttl_seconds=15 * 60, resource_type="file_integrity_finding", resource_id=finding.id, metadata={ "resources": ["postgresql", "object-storage"], "storage_backend": backend.name, }, ) except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: raise FileStorageError( "This orphan object is already owned by another recovery operation" ) from exc except (RecoveryGuaranteeError, RuntimeError) as exc: raise FileStorageError( "The Files recovery ledger is unavailable; no object was deleted" ) from exc if started.replayed or started.operation is None: raise FileStorageError( "This orphan cleanup was already completed; reload the finding" ) pending = PendingOrphanCleanup( operation=started.operation, backend=backend, finding_id=finding.id, tenant_id=finding.tenant_id, storage_key=finding.storage_key, resolved_by_user_id=user_id, ) _register_pending_effect(session, pending) return pending def _register_pending_effect(session: Session, effect: _PendingEffect) -> None: if not session.in_transaction(): session.begin() pending = session.info.setdefault(_PENDING_EFFECTS_KEY, []) pending.append(effect) if session.info.get(_HOOKS_INSTALLED_KEY): return event.listen(session, "after_commit", _after_session_commit) event.listen(session, "after_rollback", _after_session_rollback) session.info[_HOOKS_INSTALLED_KEY] = True def _after_session_commit(session: Session) -> None: _settle_pending_effects(session, committed=True) def _after_session_rollback(session: Session) -> None: try: _settle_pending_effects(session, committed=False) except RecoveryGuaranteeError as exc: session.info.setdefault(_ROLLBACK_ERRORS_KEY, []).append(str(exc)) def _settle_pending_effects(session: Session, *, committed: bool) -> None: pending = list(session.info.pop(_PENDING_EFFECTS_KEY, [])) failures: list[Exception] = [] for effect in pending: try: effect.settle(committed=committed) except Exception as exc: # preserve every effect's chance to settle failures.append(exc) operation = getattr(effect, "operation", None) if operation is not None and not operation.closed: try: operation.release_unresolved() except Exception: pass if failures: raise RecoveryGuaranteeError( f"{len(failures)} Files recovery operation(s) could not be finalized" ) from failures[0] def _blob_write_evidence(effect: PendingBlobWrite) -> dict[str, object]: database_blob_present: bool | None database_matches: bool | None database_integrity_verified: bool | None try: with effect.operation.session_factory() as session: blob = session.get(FileBlob, effect.blob_id) database_blob_present = blob is not None database_matches = bool( blob is not None and blob.tenant_id == effect.tenant_id and blob.storage_key == effect.storage_key and blob.checksum_sha256 == effect.semantic_checksum_sha256 and blob.size_bytes == effect.semantic_size_bytes and blob.protection_discriminator == effect.protection_discriminator and blob.encryption_envelope_id == effect.expected_envelope_id and ( effect.expected_storage_checksum_sha256 == ( blob.storage_checksum_sha256 or blob.checksum_sha256 ) ) ) if blob is not None else False database_integrity_verified = bool( blob is not None and blob.integrity_status == "verified" and blob.quarantined_at is None ) if blob is not None else False except Exception: database_blob_present = None database_matches = None database_integrity_verified = None object_present: bool | None observed_size: int | None = None observed_checksum: str | None = None try: digest = hashlib.sha256() observed_size = 0 for chunk in effect.backend.iter_bytes(effect.storage_key): observed_size += len(chunk) digest.update(chunk) observed_checksum = digest.hexdigest() object_present = True except StorageObjectMissing: object_present = False except (StorageBackendError, OSError): object_present = None object_size_matches = ( observed_size == effect.expected_storage_size_bytes if object_present is True and effect.expected_storage_size_bytes is not None else False if object_present is False else None ) object_checksum_matches = ( observed_checksum == effect.expected_storage_checksum_sha256 if object_present is True and effect.expected_storage_checksum_sha256 is not None else False if object_present is False else None ) verified = database_blob_present is not None and object_present is not None return { "verified": verified, "checks": { "database_reloaded": database_blob_present is not None, "object_probed": object_present is not None, "database_matches_request": database_matches, "database_integrity_verified": database_integrity_verified, "object_size_matches": object_size_matches, "object_checksum_matches": object_checksum_matches, }, "blob_id": effect.blob_id, "database_blob_present": database_blob_present, "database_matches_request": database_matches, "database_integrity_verified": database_integrity_verified, "object_present": object_present, "observed_size_bytes": observed_size, "observed_checksum_sha256": observed_checksum, "expected_storage_size_bytes": effect.expected_storage_size_bytes, "expected_storage_checksum_sha256": effect.expected_storage_checksum_sha256, } def _blob_write_complete(evidence: dict[str, object]) -> bool: return bool( evidence.get("verified") is True and evidence.get("database_matches_request") is True and evidence.get("database_integrity_verified") is True and _object_matches(evidence) ) def _object_matches(evidence: dict[str, object]) -> bool: checks = evidence.get("checks") return bool( isinstance(checks, dict) and evidence.get("object_present") is True and checks.get("object_size_matches") is True and checks.get("object_checksum_matches") is True ) def _forward_complete_blob_repair(effect: PendingBlobWrite) -> bool: try: with effect.operation.session_factory() as session: blob = session.get(FileBlob, effect.blob_id) if ( blob is None or blob.tenant_id != effect.tenant_id or blob.storage_key != effect.storage_key or blob.checksum_sha256 != effect.semantic_checksum_sha256 or blob.size_bytes != effect.semantic_size_bytes or blob.protection_discriminator != effect.protection_discriminator or blob.encryption_envelope_id != effect.expected_envelope_id ): return False blob.storage_checksum_sha256 = ( effect.expected_storage_checksum_sha256 if effect.expected_envelope_id else None ) blob.storage_size_bytes = ( effect.expected_storage_size_bytes if effect.expected_envelope_id else None ) blob.integrity_status = "verified" blob.integrity_checked_at = datetime.now(UTC) blob.integrity_failure = None blob.quarantined_at = None session.add(blob) session.commit() return True except Exception: return False def _quarantine_blob_after_failed_verification( effect: PendingBlobWrite, evidence: dict[str, object], ) -> None: try: with effect.operation.session_factory() as session: blob = session.get(FileBlob, effect.blob_id) if blob is None or blob.storage_key != effect.storage_key: return blob.integrity_status = ( "missing" if evidence.get("object_present") is False else "checksum_mismatch" ) blob.integrity_checked_at = datetime.now(UTC) blob.integrity_failure = "recovery_verification_failed" blob.quarantined_at = datetime.now(UTC) session.add(blob) session.commit() except Exception: return def _orphan_cleanup_evidence(effect: PendingOrphanCleanup) -> dict[str, object]: try: object_present: bool | None = effect.backend.exists(effect.storage_key) except (StorageBackendError, OSError): object_present = None finding_present: bool | None finding_deleted: bool | None still_unreferenced: bool | None try: with effect.operation.session_factory() as session: finding = session.get(FileIntegrityFinding, effect.finding_id) finding_present = finding is not None finding_deleted = bool( finding is not None and finding.state == "deleted" ) if finding is not None else False referenced = ( session.query(FileBlob.id) .filter( FileBlob.tenant_id == effect.tenant_id, FileBlob.storage_key == effect.storage_key, ) .first() ) still_unreferenced = referenced is None except Exception: finding_present = None finding_deleted = None still_unreferenced = None verified = object_present is not None and finding_present is not None return { "verified": verified, "checks": { "database_reloaded": finding_present is not None, "object_probed": object_present is not None, "finding_marked_deleted": finding_deleted, "object_absent": ( not object_present if object_present is not None else None ), "still_unreferenced": still_unreferenced, }, "finding_id": effect.finding_id, "finding_present": finding_present, "finding_deleted": finding_deleted, "object_present": object_present, "still_unreferenced": still_unreferenced, } def _orphan_cleanup_complete(evidence: dict[str, object]) -> bool: return bool( evidence.get("verified") is True and evidence.get("finding_deleted") is True and evidence.get("object_present") is False and evidence.get("still_unreferenced") is True ) def _forward_complete_orphan_finding(effect: PendingOrphanCleanup) -> bool: try: with effect.operation.session_factory() as session: finding = session.get(FileIntegrityFinding, effect.finding_id) if ( finding is None or finding.tenant_id != effect.tenant_id or finding.storage_key != effect.storage_key ): return False scan = session.get(FileIntegrityScan, finding.scan_id) if scan is None or not finding.storage_key.startswith( scan.storage_prefix ): return False referenced = ( session.query(FileBlob.id) .filter( FileBlob.tenant_id == effect.tenant_id, FileBlob.storage_key == effect.storage_key, ) .first() ) if referenced: return False finding.state = "deleted" finding.resolved_at = datetime.now(UTC) finding.resolved_by_user_id = effect.resolved_by_user_id session.add(finding) session.commit() return True except Exception: return False __all__ = [ "PendingBlobWrite", "PendingOrphanCleanup", "begin_blob_write_recovery", "begin_orphan_cleanup_recovery", ]