From ad55d47645e239c96f9b87c60a8a510045759c2e Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 19 Aug 2026 19:20:05 +0200 Subject: [PATCH] fix(recovery): support batched SQLite blob writes --- README.md | 15 +- docs/FILES_HANDBOOK.md | 18 +- src/govoplan_files/backend/manifest.py | 2 +- .../backend/storage/recovery.py | 346 +++++++++++++++--- tests/test_storage_recovery.py | 58 ++- 5 files changed, 374 insertions(+), 65 deletions(-) diff --git a/README.md b/README.md index 60c428f..d36aa6e 100644 --- a/README.md +++ b/README.md @@ -165,11 +165,16 @@ the original archive and stores only the selected members. Password-protected ZIP passwords remain request-only and are never included in the preview token. Managed blob writes and applied orphan cleanup use Core's durable recovery -ledger. Intent, request digests, recovery mode, and a distributed lease are -committed before physical storage effects; the Files session commit verifies -database and streamed object evidence, while rollback compensates only a newly -reserved unreferenced key. New object keys are opaque and do not retain the -uploaded filename. Uncertain or mismatched effects remain visible through Ops. +ledger. On PostgreSQL, intent, request digests, recovery mode, and a distributed +lease are committed before physical storage effects; the Files session commit +verifies database and streamed object evidence, while rollback compensates only +a newly reserved unreferenced key. Development SQLite records blob intent in +the caller transaction to avoid its second-writer deadlock, then verifies on +commit or reconstructs compensation evidence after handled rollback. Because a +hard loss before that commit can leave an unrecorded object, SQLite requires a +complete integrity scan after a crash and is not a production recovery profile. +New object keys are opaque and do not retain the uploaded filename. Uncertain +or mismatched effects remain visible through Ops. Operators with `files:file:admin` can run bounded, resumable integrity scans in Administration. Scan batches and finding actions carry monotonic revisions; diff --git a/docs/FILES_HANDBOOK.md b/docs/FILES_HANDBOOK.md index 5a7d22b..606194b 100644 --- a/docs/FILES_HANDBOOK.md +++ b/docs/FILES_HANDBOOK.md @@ -573,15 +573,27 @@ candidates until those controls are implemented. ### Recovery ledger for object effects -Every managed blob creation or integrity repair starts a Core recovery -operation in an independent committed transaction before Files protects or -writes bytes. The operation records tenant/blob identifiers, an opaque object +On PostgreSQL, every managed blob creation or integrity repair starts a Core +recovery operation in an independent committed transaction before Files +protects or writes bytes. The operation records tenant/blob identifiers, an opaque object locator or locator digest, semantic SHA-256/size evidence, the recovery mode, and a distributed lease fence. It never records file contents, ZIP passwords, connector credentials, or a newly uploaded filename. New object keys are opaque; legacy filename-bearing keys remain readable but repair operations record only their digest and recover through the blob ID. +SQLite is a supported local-development database but permits only one writer. +Files therefore uses an explicit reduced-durability mode there: recovery intent +and its lease are written in the caller transaction, while a process-local +fence prevents competing effects in the same runtime. Commit makes the intent +durable before independent verification. A handled rollback reconstructs a +durable recovery operation and verifies compensation or forward completion. +A hard process loss before commit can leave an object without a surviving +ledger row, so SQLite is not a production recovery profile; after such a loss, +run a complete Files integrity scan and reconcile every reported orphan before +resuming writes. PostgreSQL retains the independent pre-effect durability +guarantee. + The Files business transaction then creates or updates the blob, version, and asset rows. Its actual SQLAlchemy commit or rollback settles every pending operation: diff --git a/src/govoplan_files/backend/manifest.py b/src/govoplan_files/backend/manifest.py index 9183ecb..3e523b5 100644 --- a/src/govoplan_files/backend/manifest.py +++ b/src/govoplan_files/backend/manifest.py @@ -1178,7 +1178,7 @@ manifest = ModuleManifest( title="Operate Files integrity, recovery, and connector transport safety", summary="Back up database evidence, blob ciphertext, and Encryption custody as one recovery unit, and pin every SDK-managed connector peer.", body=( - "Local durable storage is the operational baseline. Recover Files from a coordinated database/blob snapshot with the matching Encryption tables and original deployment master key, then run the bounded resumable integrity scan from Administration and verify representative protected and unprotected access paths. Each scan batch and finding action requires the revision shown to the operator, so a stale screen cannot recheck or delete after concurrent reconciliation. Protected scans verify stored ciphertext before decryption and then verify plaintext semantic evidence. Managed blob creation/repair and applied orphan cleanup commit lease-fenced Core recovery intent before object effects; success, compensation, and forward completion require independent database and object checks, while mismatch is quarantined and unresolved work remains visible in Ops. Missing or mismatched blobs are quarantined; orphan objects are reported before dry-run-first, explicitly authorized cleanup. " + "Local durable storage is the operational baseline. Recover Files from a coordinated database/blob snapshot with the matching Encryption tables and original deployment master key, then run the bounded resumable integrity scan from Administration and verify representative protected and unprotected access paths. Each scan batch and finding action requires the revision shown to the operator, so a stale screen cannot recheck or delete after concurrent reconciliation. Protected scans verify stored ciphertext before decryption and then verify plaintext semantic evidence. On PostgreSQL, managed blob creation/repair and applied orphan cleanup commit lease-fenced Core recovery intent before object effects; success, compensation, and forward completion require independent database and object checks, while mismatch is quarantined and unresolved work remains visible in Ops. Development SQLite instead records blob intent in the caller transaction to avoid a second-writer deadlock, uses a process-local fence, verifies after commit, and reconstructs durable compensation evidence after handled rollback. A hard process loss before the SQLite caller commits can therefore leave an unrecorded object; SQLite is not a production recovery profile and operators must run an integrity scan after such a loss. Missing or mismatched blobs are quarantined; orphan objects are reported before dry-run-first, explicitly authorized cleanup. " "S3 connector pools pin every retry, redirect, discovered endpoint, and provider alias while retaining the configured TLS authority; outbound proxies and ambient credential discovery are disabled. SMB initial connections, reconnects, aliases, and DFS referrals use a Files-owned pinned transport and cache. Both apply the deployment private-network policy immediately before each socket opens and fail closed if an SDK no longer exposes the verified transport seam. Installer-owned Garage storage is supported only at the exact deployment service endpoint with its explicit trust marker. Destructive module retirement drops database tables but does not remove backend blob objects." ), layer="configured", diff --git a/src/govoplan_files/backend/storage/recovery.py b/src/govoplan_files/backend/storage/recovery.py index 4acaeef..207723e 100644 --- a/src/govoplan_files/backend/storage/recovery.py +++ b/src/govoplan_files/backend/storage/recovery.py @@ -1,26 +1,39 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass from datetime import UTC, datetime import hashlib +from threading import Lock from typing import Protocol -from sqlalchemy import event +from sqlalchemy import event, select from sqlalchemy.orm import Session from govoplan_core.core.recovery import ( RecoveryGuaranteeError, RecoveryMode, + RecoveryOperation, RecoveryPlan, RecoveryStatus, + plan_recovery_operation, + prepare_recovery_operation, + start_recovery_operation, + verify_recovery_evidence_chain, ) from govoplan_core.core.recovery_runtime import ( DurableRecoveryOperation, + DurableRecoveryStart, RecoveryOperationBusy, RecoveryOperationStateConflict, begin_durable_recovery_operation, ) -from govoplan_core.core.runtime_coordination import process_runtime_identity +from govoplan_core.core.runtime_coordination import ( + RuntimeIdentity, + acquire_lease, + process_runtime_identity, + release_lease, +) from govoplan_core.db.session import get_database from govoplan_files.backend.db.models import ( FileBlob, @@ -38,6 +51,8 @@ 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" +_SQLITE_FENCES: set[str] = set() +_SQLITE_FENCES_LOCK = Lock() class _PendingEffect(Protocol): @@ -58,6 +73,8 @@ class PendingBlobWrite: expected_storage_checksum_sha256: str | None = None expected_storage_size_bytes: int | None = None expected_envelope_id: str | None = None + sqlite_fence_key: str | None = None + restart_after_rollback: Callable[[], DurableRecoveryOperation] | None = None def prepare_stored_bytes( self, @@ -72,6 +89,19 @@ class PendingBlobWrite: self.expected_envelope_id = envelope_id def settle(self, *, committed: bool) -> None: + if not committed and self.restart_after_rollback is not None: + try: + self.operation = self.restart_after_rollback() + except Exception: + # SQLite cannot make the pre-effect ledger row independent of + # the caller transaction. If reconstructing the evidence row + # also fails, still avoid leaving a known new orphan behind. + if self.created_new: + try: + self.backend.delete(self.storage_key) + except (StorageBackendError, OSError): + pass + raise evidence = _blob_write_evidence(self) if _blob_write_complete(evidence): self.operation.succeed(evidence=evidence) @@ -219,77 +249,145 @@ def begin_blob_write_recovery( f"files-blob-{disposition}:{blob_id}:{state_token[:48]}" ) try: - started = begin_durable_recovery_operation( - get_database().SessionLocal, - identity=process_runtime_identity(), + identity = process_runtime_identity() + except RuntimeError as exc: + raise FileStorageError( + "The Files recovery ledger is unavailable; no object was written" + ) from exc + lease_resource_key = f"files:blob:{tenant_id}:{blob_id}" + 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, + } + metadata = { + "resources": ["postgresql", "object-storage"], + "storage_backend": backend.name, + "durability_mode": "independent_transaction", + } + session_factory = get_database().SessionLocal + sqlite_mode = session.get_bind().dialect.name == "sqlite" + sqlite_fence_key: str | None = None + + def start_independent(*, reconstructed_after_rollback: bool = False) -> DurableRecoveryStart: + return begin_durable_recovery_operation( + session_factory, + identity=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", - ), - ), + request=request, + recovery_plan=recovery_plan, 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, + **precondition_evidence, + **( + { + "sqlite_caller_transaction_rolled_back": True, + "effect_may_have_preceded_durable_intent": True, + } + if reconstructed_after_rollback + else {} + ), }, - lease_resource_key=( - f"files:blob:{tenant_id}:{blob_id}" - ), + lease_resource_key=lease_resource_key, lease_ttl_seconds=15 * 60, resource_type="file_blob", resource_id=blob_id, metadata={ - "resources": ["postgresql", "object-storage"], - "storage_backend": backend.name, + **metadata, + "durability_mode": ( + "sqlite_post_rollback_reconstruction" + if reconstructed_after_rollback + else "independent_transaction" + ), }, ) + + try: + if sqlite_mode: + _reserve_sqlite_fence(lease_resource_key) + sqlite_fence_key = lease_resource_key + started = _begin_caller_transaction_recovery_operation( + session, + session_factory=session_factory, + identity=identity, + module_id="files", + operation_type=f"blob-{disposition}", + idempotency_key=idempotency_key, + request=request, + recovery_plan=recovery_plan, + precondition_evidence={ + **precondition_evidence, + "sqlite_caller_transaction": True, + "reduced_crash_durability": True, + }, + lease_resource_key=lease_resource_key, + lease_ttl_seconds=15 * 60, + resource_type="file_blob", + resource_id=blob_id, + metadata={ + **metadata, + "resources": ["sqlite", "object-storage"], + "durability_mode": "sqlite_caller_transaction", + }, + ) + else: + started = start_independent() except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: + if sqlite_fence_key is not None: + _release_sqlite_fence(sqlite_fence_key) raise FileStorageError( "This managed blob is already owned by another recovery operation" ) from exc except (RecoveryGuaranteeError, RuntimeError) as exc: + if sqlite_fence_key is not None: + _release_sqlite_fence(sqlite_fence_key) raise FileStorageError( "The Files recovery ledger is unavailable; no object was written" ) from exc if started.replayed or started.operation is None: + if sqlite_fence_key is not None: + _release_sqlite_fence(sqlite_fence_key) raise FileStorageError( "The matching Files blob operation was already completed; reload before retrying" ) @@ -303,6 +401,14 @@ def begin_blob_write_recovery( semantic_size_bytes=semantic_size_bytes, protection_discriminator=protection_discriminator, created_new=created_new, + sqlite_fence_key=sqlite_fence_key, + restart_after_rollback=( + lambda: _require_started_operation( + start_independent(reconstructed_after_rollback=True) + ) + if sqlite_mode + else None + ), ) _register_pending_effect(session, pending) return pending @@ -398,10 +504,17 @@ def _register_pending_effect(session: Session, effect: _PendingEffect) -> None: def _after_session_commit(session: Session) -> None: + # SQLAlchemy also emits after_commit for a released SAVEPOINT. A Files + # batch may open one while creating a later recovery row; settle only when + # the outer business transaction has actually become visible. + if session.in_nested_transaction(): + return _settle_pending_effects(session, committed=True) def _after_session_rollback(session: Session) -> None: + if session.in_nested_transaction(): + return try: _settle_pending_effects(session, committed=False) except RecoveryGuaranteeError as exc: @@ -422,12 +535,139 @@ def _settle_pending_effects(session: Session, *, committed: bool) -> None: operation.release_unresolved() except Exception: pass + finally: + sqlite_fence_key = getattr(effect, "sqlite_fence_key", None) + if sqlite_fence_key: + _release_sqlite_fence(sqlite_fence_key) if failures: raise RecoveryGuaranteeError( f"{len(failures)} Files recovery operation(s) could not be finalized" ) from failures[0] +def _reserve_sqlite_fence(resource_key: str) -> None: + with _SQLITE_FENCES_LOCK: + if resource_key in _SQLITE_FENCES: + raise RecoveryOperationBusy( + f"Another local SQLite transaction owns {resource_key}" + ) + _SQLITE_FENCES.add(resource_key) + + +def _release_sqlite_fence(resource_key: str) -> None: + with _SQLITE_FENCES_LOCK: + _SQLITE_FENCES.discard(resource_key) + + +def _require_started_operation(started: DurableRecoveryStart) -> DurableRecoveryOperation: + if started.replayed or started.operation is None: + raise RecoveryOperationStateConflict(started.operation_id, started.status) + return started.operation + + +def _begin_caller_transaction_recovery_operation( + session: Session, + *, + session_factory: Callable[[], Session], + identity: RuntimeIdentity, + module_id: str, + operation_type: str, + idempotency_key: str, + request: dict[str, object], + recovery_plan: RecoveryPlan, + precondition_evidence: dict[str, object], + lease_resource_key: str, + lease_ttl_seconds: int, + resource_type: str, + resource_id: str, + metadata: dict[str, object], +) -> DurableRecoveryStart: + """Start SQLite recovery evidence inside the caller transaction. + + SQLite permits only one writer, so opening the normal independent ledger + transaction after an earlier member of a batch has written will deadlock + until the busy timeout. The caller-transaction mode preserves fencing, + request hashing, checkpoints, commit verification, and rollback + compensation, but cannot make the intent survive a hard process loss + before the caller commits. PostgreSQL never uses this reduced mode. + """ + + claim = acquire_lease( + session, + installation_id=identity.installation_id, + resource_key=lease_resource_key, + holder_node_id=identity.node_id, + holder_incarnation=identity.incarnation, + ttl_seconds=lease_ttl_seconds, + metadata={ + "module_id": module_id, + "operation_type": operation_type, + "durability_mode": "sqlite_caller_transaction", + }, + ) + if claim is None: + raise RecoveryOperationBusy( + f"Another runtime owns the recovery fence for {lease_resource_key}" + ) + existing = session.execute( + select(RecoveryOperation).where( + RecoveryOperation.installation_id == identity.installation_id, + RecoveryOperation.module_id == module_id, + RecoveryOperation.idempotency_key == idempotency_key, + ) + ).scalar_one_or_none() + operation = plan_recovery_operation( + session, + installation_id=identity.installation_id, + module_id=module_id, + operation_type=operation_type, + idempotency_key=idempotency_key, + request=request, + recovery_plan=recovery_plan, + resource_type=resource_type, + resource_id=resource_id, + lease_claim=claim, + metadata=metadata, + ) + if existing is not None: + if operation.status == RecoveryStatus.SUCCEEDED.value: + release_lease(session, claim) + return DurableRecoveryStart( + operation_id=operation.id, + status=operation.status, + replayed=True, + operation=None, + ) + raise RecoveryOperationStateConflict(operation.id, operation.status) + prepare_recovery_operation( + session, + operation, + evidence=precondition_evidence, + lease_claim=claim, + ) + start_recovery_operation( + session, + operation, + evidence={"lease_resource_key": lease_resource_key}, + lease_claim=claim, + ) + if not verify_recovery_evidence_chain(session, operation.id): + raise RecoveryGuaranteeError( + "Recovery checkpoint chain verification failed before side effects" + ) + return DurableRecoveryStart( + operation_id=operation.id, + status=operation.status, + replayed=False, + operation=DurableRecoveryOperation( + session_factory=session_factory, + operation_id=operation.id, + lease_claim=claim, + lease_ttl_seconds=lease_ttl_seconds, + ), + ) + + def _blob_write_evidence(effect: PendingBlobWrite) -> dict[str, object]: database_blob_present: bool | None database_matches: bool | None diff --git a/tests/test_storage_recovery.py b/tests/test_storage_recovery.py index c2e34e4..61a0401 100644 --- a/tests/test_storage_recovery.py +++ b/tests/test_storage_recovery.py @@ -115,9 +115,10 @@ class StorageRecoveryTests(unittest.TestCase): def put_bytes(backend_self, key, data, *, content_type=None): with self.Session() as evidence_session: - operation = evidence_session.query(RecoveryOperation).one() + operation = evidence_session.query(RecoveryOperation).one_or_none() observed_running_operation.append( - operation.status == RecoveryStatus.RUNNING.value + operation is not None + and operation.status == RecoveryStatus.RUNNING.value and len(operation.request_sha256) == 64 ) put_bytes(key, data, content_type=content_type) @@ -140,7 +141,14 @@ class StorageRecoveryTests(unittest.TestCase): operation = self._only_operation() self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status) - self.assertEqual([True], observed_running_operation) + # SQLite uses an explicit caller-transaction mode because it permits + # only one writer. The intent becomes visible with the business commit; + # PostgreSQL retains the independent pre-effect commit guarantee. + self.assertEqual([False], observed_running_operation) + self.assertEqual( + "sqlite_caller_transaction", + operation.metadata_["durability_mode"], + ) self.assertTrue(self.backend.exists(blob.storage_key)) self.assertNotIn("private-name", blob.storage_key) self.assertEqual(".blob", Path(blob.storage_key).suffix) @@ -164,10 +172,54 @@ class StorageRecoveryTests(unittest.TestCase): operation = self._only_operation() self.assertEqual(RecoveryStatus.RECOVERED.value, operation.status) + self.assertEqual( + "sqlite_post_rollback_reconstruction", + operation.metadata_["durability_mode"], + ) self.assertFalse(self.backend.exists(storage_key)) with self.Session() as evidence_session: self.assertIsNone(evidence_session.get(FileBlob, blob.id)) + def test_multiple_blob_writes_share_one_sqlite_business_transaction(self) -> None: + with patch( + "govoplan_files.backend.storage.files.get_storage_backend", + return_value=self.backend, + ): + first = _get_or_create_blob( + self.session, + tenant_id=TENANT_ID, + data=b"first archive member", + filename="first.txt", + content_type="text/plain", + actor_id=USER_ID, + ) + second = _get_or_create_blob( + self.session, + tenant_id=TENANT_ID, + data=b"second archive member", + filename="second.txt", + content_type="text/plain", + actor_id=USER_ID, + ) + self.session.commit() + + self.assertTrue(self.backend.exists(first.storage_key)) + self.assertTrue(self.backend.exists(second.storage_key)) + with self.Session() as evidence_session: + operations = evidence_session.query(RecoveryOperation).all() + self.assertEqual(2, len(operations)) + self.assertEqual( + {RecoveryStatus.SUCCEEDED.value}, + {operation.status for operation in operations}, + ) + self.assertEqual( + {"sqlite_caller_transaction"}, + { + operation.metadata_["durability_mode"] + for operation in operations + }, + ) + def test_post_write_tamper_is_quarantined_and_recovery_required(self) -> None: with patch( "govoplan_files.backend.storage.files.get_storage_backend",