fix: complete backup recovery safeguards

core/backup.py

@@ -306,50 +306,31 @@ def _publish_no_clobber(source, destination):
 
 def _publish_pair(root, pairs):
     publications = []
-    guards = []
     try:
         for source, destination in pairs:
             _publish_no_clobber(source, destination)
             publications.append((destination, source))
         _fsync_directory(root)
-
-        for source, _destination in pairs:
-            guard = Path(f"{source}.guard")
-            os.link(source, guard)
-            guards.append(guard)
-        for source, _destination in pairs:
-            source.unlink()
-        for guard in guards:
-            guard.unlink()
-        _fsync_directory(root)
     except Exception:
-        ownership_sources = [*guards, *(source for source, _destination in pairs if source.exists())]
-        _remove_owned_publications(publications, ownership_sources)
+        _remove_owned_publications(publications)
         for source, _destination in pairs:
             _unlink_private(source)
-        for guard in guards:
-            _unlink_private(guard)
         raise BackupError("publication_failed") from None
 
-
-def _remove_owned_publications(publications, owned_sources):
-    for destination, original_source in reversed(publications):
-        sources = [source for source in owned_sources if source.exists()]
-        if original_source.exists():
-            sources.append(original_source)
-        try:
-            if destination.exists() and any(
-                os.path.samefile(destination, source) for source in sources
-            ):
-                destination.unlink()
-        except OSError:
-            pass
+    cleanup_failed = False
+    for source, _destination in pairs:
+        if not _unlink_private(source):
+            cleanup_failed = True
+    if cleanup_failed:
+        raise BackupError("cleanup_failed")
 
 
-def _remove_our_publications(publications):
+def _remove_owned_publications(publications):
     for destination, source in reversed(publications):
         try:
-            if destination.exists() and os.path.samefile(destination, source):
+            if source.exists() and destination.exists() and os.path.samefile(
+                destination, source
+            ):
                 destination.unlink()
         except OSError:
             pass
@@ -412,7 +393,11 @@ def _load_manifest(path):
         raise BackupError("invalid_manifest") from None
     if not isinstance(manifest, dict) or set(manifest) != _MANIFEST_KEYS:
         raise BackupError("invalid_manifest")
-    if manifest["format"] != BACKUP_FORMAT or manifest["version"] != BACKUP_VERSION:
+    if (
+        manifest["format"] != BACKUP_FORMAT
+        or type(manifest["version"]) is not int
+        or manifest["version"] != BACKUP_VERSION
+    ):
         raise BackupError("invalid_manifest")
     if (
         not isinstance(manifest["application_version"], str)
@@ -570,11 +555,16 @@ def run_synthetic_restore_verification():
         or _same_database(restore_target, live_target)
     ):
         raise BackupError("live_target")
+    try:
         if _read_snapshot(source_target).row_counts:
             raise BackupError("source_not_empty")
         restore_before = _read_snapshot(restore_target)
         if restore_before.migrations or restore_before.row_counts:
             raise BackupError("target_not_empty")
+    except BackupError:
+        raise
+    except Exception:
+        raise BackupError("integration_verification_failed") from None
 
     default_connection = connections["default"]
     original_database = dict(default_connection.settings_dict)
@@ -684,6 +674,8 @@ def _clear_disposable_database(target):
             autocommit=True,
         ) as database:
             with database.cursor() as cursor:
+                cursor.execute("SET lock_timeout = '5s'")
+                cursor.execute("SET statement_timeout = '15s'")
                 cursor.execute("DROP SCHEMA public CASCADE")
                 cursor.execute("CREATE SCHEMA public")
     except Exception:

core/management/commands/test_backup_restore.py

@@ -9,6 +9,7 @@ class Command(BaseCommand):
     def handle(self, *args, **options):
         try:
             result = run_synthetic_restore_verification()
+            result = run_synthetic_restore_verification()
         except BackupError as error:
             raise CommandError(error.code) from None
         self.stdout.write(f"backup={result.name} format={result.version} verified")

tests/test_backup.py

@@ -7,6 +7,7 @@ from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
+import psycopg
 from django.db import connection
 from psycopg import connect, sql
 
@@ -180,7 +181,7 @@ def test_backup_sanitizes_subprocess_failure_and_captures_diagnostics(
     assert list(backup_root.iterdir()) == []
 
 
-def test_second_temporary_unlink_failure_leaves_no_partial_final_pair(
+def test_second_temporary_unlink_failure_preserves_complete_committed_pair(
     backup_root, database_settings, synthetic_snapshot, monkeypatch
 ):
     original_unlink = Path.unlink
@@ -195,6 +196,25 @@ def test_second_temporary_unlink_failure_leaves_no_partial_final_pair(
 
     monkeypatch.setattr(Path, "unlink", fail_once)
 
+    with pytest.raises(BackupError, match="cleanup_failed"):
+        create_backup("synthetic.dump", runner=RecordingRunner())
+
+    assert (backup_root / "synthetic.dump").read_bytes() == b"PGDMP synthetic"
+    assert (backup_root / "synthetic.dump.json").is_file()
+    assert sorted(path.name for path in backup_root.iterdir()) == [
+        "synthetic.dump",
+        "synthetic.dump.json",
+    ]
+
+
+def test_directory_fsync_failure_rolls_back_both_owned_finals(
+    backup_root, database_settings, synthetic_snapshot, monkeypatch
+):
+    monkeypatch.setattr(
+        "core.backup._fsync_directory",
+        lambda _root: (_ for _ in ()).throw(OSError("synthetic fsync failure")),
+    )
+
     with pytest.raises(BackupError, match="publication_failed"):
         create_backup("synthetic.dump", runner=RecordingRunner())
 
@@ -409,6 +429,47 @@ def test_manifest_rejects_boolean_integer_fields(
         verify_backup(result.name, target, runner=RecordingRunner())
 
 
+def test_manifest_rejects_boolean_version(
+    backup_root, database_settings, synthetic_snapshot, settings
+):
+    settings.BACKUP_VERIFY_ALLOWED = True
+    result = create_backup("synthetic.dump", runner=RecordingRunner())
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    manifest["version"] = True
+    result.manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+    target = DatabaseTarget(
+        host="restore-db", port="5432", name="restore", user="user", password="secret"
+    )
+
+    with pytest.raises(BackupError, match="invalid_manifest"):
+        verify_backup(result.name, target, runner=RecordingRunner())
+
+
+def test_synthetic_preflight_database_error_is_sanitized(monkeypatch, settings):
+    import core.backup as backup
+
+    settings.BACKUP_VERIFY_ALLOWED = True
+    for name, value in {
+        "BACKUP_SOURCE_HOST": "source",
+        "BACKUP_SOURCE_NAME": "source",
+        "BACKUP_SOURCE_USER": "user",
+        "BACKUP_SOURCE_PASSWORD": "SYNTHETIC PRIVATE PASSWORD",
+    }.items():
+        monkeypatch.setenv(name, value)
+    monkeypatch.setattr(
+        backup,
+        "_read_snapshot",
+        lambda _target: (_ for _ in ()).throw(
+            psycopg.OperationalError("SYNTHETIC PRIVATE DIAGNOSTIC")
+        ),
+    )
+
+    with pytest.raises(BackupError, match="integration_verification_failed") as error:
+        backup.run_synthetic_restore_verification()
+
+    assert "PRIVATE" not in str(error.value)
+
+
 def test_cleanup_attempts_every_resource_and_reports_safe_failure(monkeypatch, tmp_path):
     import core.backup as backup
 

tests/test_backup_commands.py

@@ -116,18 +116,26 @@ def test_verify_backup_converts_service_error_to_safe_code(monkeypatch, settings
 
 
 def test_backup_restore_acceptance_command_prints_only_safe_summary(monkeypatch):
-    monkeypatch.setattr(
-        "core.management.commands.test_backup_restore.run_synthetic_restore_verification",
-        lambda: VerificationResult(
+    calls = 0
+
+    def verify():
+        nonlocal calls
+        calls += 1
+        return VerificationResult(
             name="synthetic-restore-check.dump",
             version=1,
             row_counts={"finance_account": 1, "finance_transaction": 1},
-        ),
+        )
+
+    monkeypatch.setattr(
+        "core.management.commands.test_backup_restore.run_synthetic_restore_verification",
+        verify,
     )
     stdout = StringIO()
 
     call_command("test_backup_restore", stdout=stdout)
 
+    assert calls == 2
     assert stdout.getvalue().splitlines() == [
         "backup=synthetic-restore-check.dump format=1 verified",
         "table=finance_account rows=1",

tests/test_backup_restore.py

@@ -1,8 +1,19 @@
+import os
 import subprocess
 
+import pytest
+
+from core.backup import (
+    DatabaseSnapshot,
+    DatabaseTarget,
+    _read_snapshot,
+    run_synthetic_restore_verification,
+)
+
 
 def test_backup_configuration_is_allow_gated(settings):
     assert settings.BACKUP_ROOT == "/app/backups"
+    if not os.environ.get("BACKUP_SOURCE_HOST"):
         assert settings.BACKUP_VERIFY_ALLOWED is False
 
 
@@ -12,3 +23,35 @@ def test_application_image_installs_matching_postgresql_client():
     )
 
     assert result.stdout.startswith("pg_dump (PostgreSQL) 17.")
+
+
+@pytest.mark.skipif(
+    not all(
+        os.environ.get(name)
+        for name in (
+            "BACKUP_SOURCE_HOST",
+            "BACKUP_SOURCE_NAME",
+            "BACKUP_SOURCE_USER",
+            "BACKUP_SOURCE_PASSWORD",
+        )
+    ),
+    reason="requires the backup profile's disposable databases",
+)
+def test_synthetic_restore_is_repeatable_and_leaves_both_databases_empty(
+    settings, django_db_blocker
+):
+    with django_db_blocker.unblock():
+        first = run_synthetic_restore_verification()
+        second = run_synthetic_restore_verification()
+        source = DatabaseTarget(
+            host=os.environ["BACKUP_SOURCE_HOST"],
+            port=os.environ.get("BACKUP_SOURCE_PORT", "5432"),
+            name=os.environ["BACKUP_SOURCE_NAME"],
+            user=os.environ["BACKUP_SOURCE_USER"],
+            password=os.environ["BACKUP_SOURCE_PASSWORD"],
+        )
+        target = DatabaseTarget.from_default(settings.DATABASES["default"])
+        empty = DatabaseSnapshot(migrations=(), row_counts={})
+        assert first.row_counts == second.row_counts
+        assert _read_snapshot(source) == empty
+        assert _read_snapshot(target) == empty