feat(delivery): deduplicate bounded cold backups

scripts/release/backup.py

@@ -15,12 +15,16 @@ from pathlib import Path
 from . import (
     availability,
     backup_manifest,
+    backup_objects,
     configuration,
     install,
     lifecycle,
     preflight,
     service_install,
 )
+from . import (
+    manifest as release_manifest,
+)
 from .policy import DeliveryError, canonical, digest, read_json, sha256
 from .services import python_environment
 
@@ -237,6 +241,7 @@ def guarded() -> None:
             raise DeliveryError("backup process limit differs")
 
 
+@release_manifest.verified_operation
 def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
     from .recovery_database import bootstrap_fingerprint
 
@@ -246,6 +251,8 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
     for name in (
         "backup.py",
         "backup_manifest.py",
+        "backup_objects.py",
+        "file_copy.py",
         "recovery_database.py",
         "recovery_identity.py",
         "dependency_initialize.py",
@@ -282,9 +289,10 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
         if target.exists() or target.is_symlink():
             raise DeliveryError("backup target already exists; incomplete backups retained")
         target.mkdir(mode=0o700)
-        tree = target / "payload"
+        tree = target / "objects"
         tree.mkdir(mode=0o700)
-        dirs, files, total = [], [], 0
+        dirs, files, total, stored = [], [], 0, 0
+        objects = {}
         for domain, source_root in SOURCES.items():
             if source_root.is_symlink() or not source_root.is_dir():
                 raise DeliveryError("a required backup source domain is absent or linked")
@@ -310,10 +318,8 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
                 backup_manifest._member(entry, set(), directory=True)
                 if not stat.S_ISDIR(info.st_mode):
                     raise DeliveryError("backup source directory changed type")
-                destination = tree / relative
-                destination.mkdir(mode=0o700)
                 dirs.append(entry)
-                if len(dirs) + len(files) > backup_manifest.MAX_ENTRIES:
+                if len(dirs) + len(files) > backup_manifest.MAX_LOGICAL_ENTRIES:
                     raise DeliveryError("backup directory inventory exceeds bound")
                 for member in names:
                     if (source_dir / member).is_symlink():
@@ -335,29 +341,22 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
                         raise DeliveryError("backup individual file exceeds bound")
                     total += before.st_size
                     if (
-                        total > backup_manifest.MAX_BYTES
-                        or len(dirs) + len(files) >= backup_manifest.MAX_ENTRIES
+                        total > backup_manifest.MAX_LOGICAL_BYTES
+                        or len(dirs) + len(files) >= backup_manifest.MAX_LOGICAL_ENTRIES
                         or os.statvfs(ROOT).f_frsize * os.statvfs(ROOT).f_bavail
                         < before.st_size + 64 * 1024**2
                     ):
                         raise DeliveryError("private backup capacity exhausted")
-                    h = hashlib.sha256()
-                    with os.fdopen(
-                        os.open(p, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC),
-                        "rb",
-                    ) as incoming:
-                        with os.fdopen(
-                            os.open(
-                                destination / member,
-                                os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC,
-                                0o600,
-                            ),
-                            "wb",
-                        ) as outgoing:
-                            copy_file(incoming, outgoing, before, deadline, h)
-                    entry["sha256"] = h.hexdigest()
+                    entry["sha256"], added = backup_objects.store(
+                        p,
+                        before,
+                        tree,
+                        objects,
+                        deadline,
+                        remaining_bytes=backup_manifest.MAX_BYTES - stored,
+                    )
+                    stored += added
                     files.append(entry)
-                install.sync_directory(destination)
         stopped(selected, cut)
         if install.selection() != selected or read_json(lifecycle.LAST, 32768) != cut:
             raise DeliveryError("selected cut changed during backup")
@@ -369,8 +368,8 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
         now = int(time.time())
         value = backup_manifest.validate(
             {
-                "schema_version": 3,
-                "format": backup_manifest.FORMAT,
+                "schema_version": 4,
+                "format": backup_manifest.OBJECT_FORMAT,
                 "backup_id": backup_id,
                 "source_sha": manifest["source_sha"],
                 "manifest_sha256": selected["manifest_sha256"],
@@ -385,7 +384,8 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
                 "images": manifest["images"],
                 "directories": sorted(dirs, key=lambda x: x["path"]),
                 "files": sorted(files, key=lambda x: x["path"]),
-                "payload_bytes": total,
+                "payload_bytes": stored,
+                "logical_bytes": total,
                 "restored": False,
                 "execution_verified": False,
             }
@@ -393,16 +393,20 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
         raw = canonical(value)
         if len(raw) > backup_manifest.MAX_METADATA_BYTES:
             raise DeliveryError("private backup manifest exceeds bound")
+        backup_objects.sync_filesystem(tree, deadline)
         install.exclusive_file(target / "manifest.json", raw, 0o600)
         install.sync_directory(tree)
         install.sync_directory(target)
         install.sync_directory(ROOT)
         fingerprint = digest(target / "manifest.json")
+        # Release construction-only inventories before the independent byte read.
+        del value, raw, dirs, files, objects
         backup_manifest.verify(target, fingerprint)
         return {
             "backup_id": backup_id,
             "private_manifest_sha256": fingerprint,
-            "payload_bytes": total,
+            "payload_bytes": stored,
+            "logical_bytes": total,
             "source_sha": manifest["source_sha"],
             "captured": True,
             "restored": False,
@@ -412,26 +416,6 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict:
         os.close(descriptor)
 
 
-def copy_file(incoming, outgoing, before, deadline: float, h) -> None:
-    keys = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
-    observed = os.fstat(incoming.fileno())
-    if any(getattr(before, key) != getattr(observed, key) for key in keys):
-        raise DeliveryError("backup source changed before capture")
-    count = 0
-    while block := incoming.read(65536):
-        count += len(block)
-        if count > before.st_size or time.monotonic() >= deadline:
-            raise DeliveryError("backup source or deadline changed")
-        h.update(block)
-        outgoing.write(block)
-    after = os.fstat(incoming.fileno())
-    if count != before.st_size or any(getattr(before, key) != getattr(after, key) for key in keys):
-        raise DeliveryError("backup source changed during capture")
-    outgoing.flush()
-    os.fsync(outgoing.fileno())
-    os.fchmod(outgoing.fileno(), 0o600)
-
-
 def retention_eligible(expired: dict, retained: dict, now: int) -> None:
     """An explicit expired snapshot never removes the last verified recovery copy."""
     if (
@@ -461,10 +445,11 @@ def prune(backup_id: str, expected: str, keep_id: str, keep_expected: str) -> di
         retention_eligible(value, keep, int(time.time()))
         # Remove exactly the already verified inventory, never traverse or delete
         # an unknown tree. Any interruption leaves a private partial target.
-        for item in value["files"]:
+        storage = target / backup_manifest.storage_directory(value)
+        for item in backup_manifest.physical_files(value):
             if time.monotonic() >= deadline:
                 raise DeliveryError("private backup pruning deadline")
-            path = target / "payload" / item["path"]
+            path = storage / item["path"]
             info = path.lstat()
             if (
                 not stat.S_ISREG(info.st_mode)
@@ -476,15 +461,17 @@ def prune(backup_id: str, expected: str, keep_id: str, keep_expected: str) -> di
             path.unlink()
             install.sync_directory(path.parent)
         for item in sorted(
-            value["directories"], key=lambda item: item["path"].count("/"), reverse=True
+            value["directories"] if value["schema_version"] == 3 else [],
+            key=lambda item: item["path"].count("/"),
+            reverse=True,
         ):
-            path = target / "payload" / item["path"]
+            path = storage / item["path"]
             preflight.protected(path, directory=True)
             if stat.S_IMODE(path.stat().st_mode) != 0o700:
                 raise DeliveryError("private backup directory changed before pruning")
             path.rmdir()
             install.sync_directory(path.parent)
-        (target / "payload").rmdir()
+        storage.rmdir()
         configuration.private(target / "manifest.json")
         if digest(target / "manifest.json") != sha256(expected):
             raise DeliveryError("private backup receipt changed before pruning")
@@ -591,7 +578,11 @@ def launch(args) -> dict:
     result = subprocess.run(command, capture_output=True, timeout=1820, check=False)
     if result.returncode or len(result.stdout) > 4096:
         raise DeliveryError(
-            "private backup action failed; source unchanged and partial target retained"
+            "private backup action failed; exit="
+            + str(result.returncode)
+            + "; stderr_sha256="
+            + hashlib.sha256(result.stderr).hexdigest()
+            + "; source unchanged and partial target retained"
         )
     return json.loads(result.stdout)
 
@@ -599,13 +590,19 @@ def launch(args) -> dict:
 def selected_operator() -> None:
     selected = install.selection()
     payload = install.RELEASES / selected["release_id"] / "payload"
-    for name in ("backup.py", "backup_manifest.py"):
+    for name in (
+        "backup.py",
+        "backup_manifest.py",
+        "backup_objects.py",
+        "file_copy.py",
+    ):
         if digest(payload / "delivery/scripts/release" / name) != digest(
             Path(__file__).parent / name
         ):
             raise DeliveryError("backup operator differs from selected release")
 
 
+@release_manifest.verified_operation
 def main() -> None:
     parser = argparse.ArgumentParser(description="Private bounded cold backup operator")
     sub = parser.add_subparsers(dest="action", required=True)

scripts/release/backup_manifest.py

@@ -17,9 +17,13 @@ from .policy import (
 )
 
 FORMAT = "kf.private.cold-backup.v3"
+OBJECT_FORMAT = "kf.private.cold-backup.v4"
 MAX_BYTES = 7 * 1024**3
 MAX_FILE_BYTES = 256 * 1024**2
 MAX_ENTRIES = 65536
+# v4 preserves logical versions independently of their deduplicated storage.
+MAX_LOGICAL_BYTES = 12 * 1024**3
+MAX_LOGICAL_ENTRIES = 81920
 MAX_METADATA_BYTES = 16 * 1024**2
 DOMAINS = {
     "postgres",
@@ -130,6 +134,8 @@ def validate_references(value: dict) -> dict:
 
 
 def validate(value: dict) -> dict:
+    version = value.get("schema_version") if isinstance(value, dict) else None
+    object_store = type(version) is int and version == 4
     fields(
         value,
         {
@@ -152,13 +158,14 @@ def validate(value: dict) -> dict:
             "payload_bytes",
             "restored",
             "execution_verified",
-        },
+        }
+        | ({"logical_bytes"} if object_store else set()),
         "private backup",
     )
     if (
         type(value["schema_version"]) is not int
-        or value["schema_version"] != 3
-        or value["format"] != FORMAT
+        or value["schema_version"] not in {3, 4}
+        or value["format"] != (OBJECT_FORMAT if object_store else FORMAT)
         or not isinstance(value["backup_id"], str)
         or re.fullmatch(r"bk-[0-9a-f]{32}", value["backup_id"]) is None
         or not isinstance(value["source_sha"], str)
@@ -238,23 +245,35 @@ def validate(value: dict) -> dict:
     if (
         not isinstance(dirs, list)
         or not isinstance(files, list)
-        or not 1 <= len(dirs) + len(files) <= MAX_ENTRIES
+        or not 1 <= len(dirs) + len(files) <= (MAX_LOGICAL_ENTRIES if object_store else MAX_ENTRIES)
     ):
         raise DeliveryError("backup inventory count exceeds bound")
     seen = set()
     for item in dirs:
         fields(item, {"path", "uid", "gid", "mode"}, "backup directory")
         _member(item, seen, directory=True)
-    total = 0
+    total, objects = 0, {}
     for item in files:
         fields(item, {"path", "uid", "gid", "mode", "bytes", "sha256"}, "backup file")
         _member(item, seen, directory=False)
         total += integer(item["bytes"], 0, MAX_FILE_BYTES, "backup file bytes")
         sha256(item["sha256"])
+        if item["sha256"] in objects and objects[item["sha256"]] != item["bytes"]:
+            raise DeliveryError("private backup object size conflict")
+        objects[item["sha256"]] = item["bytes"]
+    physical = sum(objects.values()) if object_store else total
     if (
         type(value["payload_bytes"]) is not int
-        or value["payload_bytes"] != total
-        or total > MAX_BYTES
+        or value["payload_bytes"] != physical
+        or physical > MAX_BYTES
+        or (
+            object_store
+            and (
+                type(value["logical_bytes"]) is not int
+                or value["logical_bytes"] != total
+                or total > MAX_LOGICAL_BYTES
+            )
+        )
     ):
         raise DeliveryError("private backup aggregate bytes differ")
     directories = {item["path"] for item in dirs}
@@ -278,6 +297,36 @@ def validate(value: dict) -> dict:
     return value
 
 
+def logical_bytes(value: dict) -> int:
+    return value["logical_bytes"] if value["schema_version"] == 4 else value["payload_bytes"]
+
+
+def storage_directory(value: dict) -> str:
+    return "objects" if value["schema_version"] == 4 else "payload"
+
+
+def member_path(root: Path, value: dict, item: dict) -> Path:
+    return (
+        root
+        / storage_directory(value)
+        / (item["sha256"] if value["schema_version"] == 4 else item["path"])
+    )
+
+
+def named_member(root: Path, value: dict, name: str) -> Path:
+    for item in value["files"]:
+        if item["path"] == name:
+            return member_path(root, value, item)
+    raise DeliveryError("private backup required logical member missing")
+
+
+def physical_files(value: dict) -> list[dict]:
+    if value["schema_version"] == 3:
+        return value["files"]
+    objects = {item["sha256"]: item["bytes"] for item in value["files"]}
+    return [{"path": key, "sha256": key, "bytes": size} for key, size in objects.items()]
+
+
 def _member(item: dict, seen: set[str], *, directory: bool) -> None:
     path = name(item["path"])
     if path in seen:
@@ -307,14 +356,14 @@ def verify(root: Path, expected: str) -> dict:
         raise DeliveryError("private backup directory required")
     path = root / "manifest.json"
     configuration.private(path)
-    if digest(path) != sha256(expected) or {p.name for p in root.iterdir()} != {
-        "manifest.json",
-        "payload",
-    }:
+    if digest(path) != sha256(expected):
         raise DeliveryError("private backup receipt or layout differs")
     value = validate(read_metadata(path))
+    storage = storage_directory(value)
+    if {p.name for p in root.iterdir()} != {"manifest.json", storage}:
+        raise DeliveryError("private backup receipt or layout differs")
     found_files, found_dirs = set(), set()
-    payload = root / "payload"
+    payload = root / storage
     preflight.protected(payload, directory=True)
     if stat.S_IMODE(payload.stat().st_mode) != 0o700:
         raise DeliveryError("private backup payload directory required")
@@ -322,7 +371,11 @@ def verify(root: Path, expected: str) -> dict:
         for member in names + files:
             p = Path(directory) / member
             info = p.lstat()
-            relative = name(p.relative_to(payload).as_posix())
+            relative = p.relative_to(payload).as_posix()
+            if value["schema_version"] == 4:
+                sha256(relative)
+            else:
+                name(relative)
             if info.st_uid != 0 or info.st_gid != 0 or p.is_symlink():
                 raise DeliveryError("private backup member ownership differs")
             if stat.S_ISDIR(info.st_mode) and stat.S_IMODE(info.st_mode) == 0o700:
@@ -335,14 +388,39 @@ def verify(root: Path, expected: str) -> dict:
                 found_files.add(relative)
             else:
                 raise DeliveryError("private backup special or non-private member")
-            if len(found_files) + len(found_dirs) > MAX_ENTRIES:
+            if len(found_files) + len(found_dirs) > (
+                MAX_LOGICAL_ENTRIES if value["schema_version"] == 4 else MAX_ENTRIES
+            ):
                 raise DeliveryError("private backup traversal exceeds bound")
-    if found_dirs != {x["path"] for x in value["directories"]} or found_files != {
-        x["path"] for x in value["files"]
-    }:
+    physical = physical_files(value)
+    if found_dirs != (
+        {x["path"] for x in value["directories"]} if value["schema_version"] == 3 else set()
+    ) or found_files != {x["path"] for x in physical}:
         raise DeliveryError("private backup inventory differs")
-    for item in value["files"]:
+    from . import file_copy
+
+    for item in physical:
         p = payload / item["path"]
-        if p.stat().st_size != item["bytes"] or digest(p) != item["sha256"]:
+        before = p.lstat()
+        if (
+            not stat.S_ISREG(before.st_mode)
+            or before.st_nlink != 1
+            or (before.st_uid, before.st_gid, stat.S_IMODE(before.st_mode)) != (0, 0, 0o600)
+            or before.st_size != item["bytes"]
+        ):
+            raise DeliveryError("private backup object protection changed")
+        with os.fdopen(os.open(p, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC), "rb") as stream:
+            if file_copy._identity(os.fstat(stream.fileno())) != file_copy._identity(before):
+                raise DeliveryError("private backup object changed before verification")
+            from hashlib import file_digest
+
+            observed = file_digest(stream, "sha256").hexdigest()
+            after = os.fstat(stream.fileno())
+        if (
+            before.st_size != item["bytes"]
+            or observed != item["sha256"]
+            or file_copy._identity(after) != file_copy._identity(before)
+            or file_copy._identity(p.lstat()) != file_copy._identity(before)
+        ):
             raise DeliveryError("private backup byte verification failed")
     return value

scripts/release/backup_objects.py

@@ -0,0 +1,117 @@
+"""Private content storage; every logical source byte is read before deduplication."""
+
+import ctypes
+import hashlib
+import os
+import stat
+import time
+from pathlib import Path
+
+from . import file_copy
+from .policy import DeliveryError
+
+CHUNK = 1024 * 1024
+
+
+def identity(info: os.stat_result) -> tuple[int, ...]:
+    return file_copy._identity(info)
+
+
+def read_stream(incoming, before, deadline: float, outgoing=None) -> str:
+    if identity(os.fstat(incoming.fileno())) != identity(before):
+        raise DeliveryError("backup source changed before capture")
+    count, digest = 0, hashlib.sha256()
+    while block := incoming.read(CHUNK):
+        count += len(block)
+        if count > before.st_size or time.monotonic() >= deadline:
+            raise DeliveryError("backup source or deadline changed")
+        digest.update(block)
+        if outgoing is not None:
+            outgoing.write(block)
+    if (
+        count != before.st_size
+        or time.monotonic() >= deadline
+        or identity(os.fstat(incoming.fileno())) != identity(before)
+    ):
+        raise DeliveryError("backup source changed during capture")
+    return digest.hexdigest()
+
+
+def store(
+    source: Path,
+    before,
+    objects: Path,
+    seen: dict[str, int],
+    deadline: float,
+    *,
+    remaining_bytes: int,
+) -> tuple[str, int]:
+    """Hash every source; copy only new content, with a second full byte check.
+
+    The one fixed pending file is exclusive. An interrupted copy remains private
+    and prevents adoption of the incomplete snapshot. Publishing never replaces
+    an existing object, and the final object has exactly one link.
+    """
+    if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1:
+        raise DeliveryError("special or linked backup source refused")
+    with os.fdopen(
+        os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC),
+        "rb",
+    ) as incoming:
+        digest = read_stream(incoming, before, deadline)
+        added = 0
+        if digest in seen:
+            if seen[digest] != before.st_size:
+                raise DeliveryError("backup content size conflict")
+        else:
+            if before.st_size > remaining_bytes:
+                raise DeliveryError("private backup physical capacity exhausted")
+            pending = objects / ".pending"
+            with os.fdopen(
+                os.open(
+                    pending,
+                    os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC,
+                    0o600,
+                ),
+                "wb",
+            ) as outgoing:
+                os.fchmod(outgoing.fileno(), 0o600)
+                incoming.seek(0)
+                if read_stream(incoming, before, deadline, outgoing) != digest:
+                    raise DeliveryError("backup source bytes changed between reads")
+                outgoing.flush()
+                written = os.fstat(outgoing.fileno())
+            if identity(pending.lstat()) != identity(written):
+                raise DeliveryError("private backup pending object changed")
+            os.link(pending, objects / digest, follow_symlinks=False)
+            # Only this operation's completed, exclusively created staging link.
+            pending.unlink()
+            seen[digest] = before.st_size
+            added = before.st_size
+        if identity(source.lstat()) != identity(before):
+            raise DeliveryError("backup source path changed during capture")
+        return digest, added
+
+
+def sync_filesystem(directory: Path, deadline: float) -> None:
+    """Flush the dedicated backup filesystem before publishing its manifest.
+
+    Linux syncfs reports delayed write errors and waits for data and metadata.
+    The existing service deadline bounds a blocked filesystem operation too.
+    """
+    if time.monotonic() >= deadline:
+        raise DeliveryError("private backup capture deadline before synchronization")
+    descriptor = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)
+    try:
+        libc = ctypes.CDLL(None, use_errno=True)
+        try:
+            syncfs = libc.syncfs
+        except AttributeError as error:
+            raise DeliveryError("private backup filesystem synchronization unavailable") from error
+        syncfs.argtypes, syncfs.restype = [ctypes.c_int], ctypes.c_int
+        if syncfs(descriptor) != 0:
+            raise DeliveryError("private backup filesystem synchronization failed")
+        if time.monotonic() >= deadline:
+            raise DeliveryError("private backup capture deadline after synchronization")
+    finally:
+        os.close(descriptor)

scripts/release/recovery_database.py

@@ -152,7 +152,11 @@ def admission_guard():
             if stat.S_IMODE(directory.stat().st_mode) != 0o700:
                 raise DeliveryError("private recovery record directory required")
             records = {}
-            for name in ("prepared.json", "data-restored.json", "identity-recovered.json"):
+            for name in (
+                "prepared.json",
+                "data-restored.json",
+                "identity-recovered.json",
+            ):
                 path = directory / name
                 preflight.protected(path)
                 configuration.private(path)
@@ -231,10 +235,18 @@ def validate_data_receipt(
         raise DeliveryError("cold data restoration is not this boot's selected complete data step")
     backup.identifier(value["backup_id"])
     sha256(value["private_backup_sha256"])
-    integer(value["restored_payload_bytes"], 1, 7 * 1024**3, "restored bytes")
+    integer(
+        value["restored_payload_bytes"],
+        1,
+        backup_manifest.MAX_LOGICAL_BYTES,
+        "restored bytes",
+    )
     for key in ("restored_files", "restored_directories"):
-        integer(value[key], 1, 50000, "restored entries")
-    if value["restored_files"] + value["restored_directories"] > 50000:
+        integer(value[key], 1, backup_manifest.MAX_LOGICAL_ENTRIES, "restored entries")
+    if (
+        value["restored_files"] + value["restored_directories"]
+        > backup_manifest.MAX_LOGICAL_ENTRIES
+    ):
         raise DeliveryError("restored inventory limit differs")
     return value
 
@@ -395,7 +407,11 @@ def recover_locked(recovery_id: str, data_sha: str, password_file: Path, passwor
     try:
         dispatched = True
         completed = subprocess.run(
-            command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=290, check=False
+            command,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            timeout=290,
+            check=False,
         )
         if completed.returncode or len(completed.stdout) > 16384:
             raise DeliveryError("DB identity recovery failed; target remains stopped")
@@ -403,7 +419,9 @@ def recover_locked(recovery_id: str, data_sha: str, password_file: Path, passwor
     finally:
         if dispatched:
             active = backup.manager(
-                "show", unit, "--property=LoadState,ActiveState,MainPID,Description,Transient"
+                "show",
+                unit,
+                "--property=LoadState,ActiveState,MainPID,Description,Transient",
             )
             if active.get("LoadState") != "not-found":
                 if active.get("Description") != description or active.get("Transient") != "yes":
@@ -491,10 +509,17 @@ def read_result(raw: bytes) -> dict:
     ):
         raise DeliveryError("ordinary DB recovery result is incomplete or unproven")
     integer(
-        value["data_filesystem_bytes"], 2 * 1024**3 * 9 // 10, 2 * 1024**3, "DB recovery volume"
+        value["data_filesystem_bytes"],
+        2 * 1024**3 * 9 // 10,
+        2 * 1024**3,
+        "DB recovery volume",
     )
     sha256(value["role_authority_sha256"])
-    fields(value["effective_settings"], set(recovery_identity.recovery_settings()), "PG settings")
+    fields(
+        value["effective_settings"],
+        set(recovery_identity.recovery_settings()),
+        "PG settings",
+    )
     for item in value["effective_settings"].values():
         fields(item, {"setting", "unit"}, "PG effective setting")
         if (
@@ -520,7 +545,10 @@ def main() -> None:
     args = parser.parse_args()
     try:
         value = recover(
-            args.recovery_id, args.data_receipt_sha256, args.password_file, args.password_sha256
+            args.recovery_id,
+            args.data_receipt_sha256,
+            args.password_file,
+            args.password_sha256,
         )
         print(json.dumps(value, sort_keys=True))
     except (OSError, ValueError, subprocess.SubprocessError) as error:

scripts/release/restore.py

@@ -62,7 +62,10 @@ def guarded() -> None:
         raise DeliveryError("fixed recovery operator unit required")
     for path, expected in (
         (group, ["536870912", "0", "32", f"{CPU_PERCENT * 1000} 100000"]),
-        (Path("/sys/fs/cgroup/kfonlinejudge.slice"), ["4026531840", "0", "384", "200000 100000"]),
+        (
+            Path("/sys/fs/cgroup/kfonlinejudge.slice"),
+            ["4026531840", "0", "384", "200000 100000"],
+        ),
     ):
         observed = [
             (path / key).read_text().strip()
@@ -86,7 +89,9 @@ def stopped() -> None:
     dependency_install.require_stopped()
     for name in (*lifecycle.APPS, *lifecycle.DEPENDENCIES):
         values = backup.manager(
-            "show", "kf-" + name + ".service", "--property=MainPID,ActiveState,ControlGroup"
+            "show",
+            "kf-" + name + ".service",
+            "--property=MainPID,ActiveState,ControlGroup",
         )
         if values.get("MainPID") != "0" or values.get("ActiveState") != "inactive":
             raise DeliveryError("cold recovery requires stopped required services")
@@ -100,7 +105,8 @@ def stopped() -> None:
         except FileNotFoundError:
             continue
         row = next(
-            (line.split()[1:] for line in status.splitlines() if line.startswith("Uid:")), None
+            (line.split()[1:] for line in status.splitlines() if line.startswith("Uid:")),
+            None,
         )
         if row is None or len(row) != 4:
             raise DeliveryError("recovery process identity unavailable")
@@ -182,18 +188,22 @@ def write_file(
     if path.exists() or path.is_symlink():
         # Existing destinations are read and checked, never overwritten.
         return readback(path, item, uid, gid, mode, deadline)
-    with os.fdopen(os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC), "rb") as incoming:
-        with os.fdopen(
+    with (
+        os.fdopen(os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC), "rb") as incoming,
+        os.fdopen(
             os.open(
-                path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600
+                path,
+                os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC,
+                0o600,
             ),
             "wb",
-        ) as outgoing:
-            file_copy.transfer(incoming, outgoing, item["bytes"], deadline=deadline)
-            outgoing.flush()
-            os.fchown(outgoing.fileno(), uid, gid)
-            os.fchmod(outgoing.fileno(), mode)
-            os.fsync(outgoing.fileno())
+        ) as outgoing,
+    ):
+        file_copy.transfer(incoming, outgoing, item["bytes"], deadline=deadline)
+        outgoing.flush()
+        os.fchown(outgoing.fileno(), uid, gid)
+        os.fchmod(outgoing.fileno(), mode)
+        os.fsync(outgoing.fileno())
     # The backup was fully verified before copying. Independently read every
     # flushed destination while its pages are still local, then bind that proof
     # to its unchanged inode/metadata through completion on the stopped node.
@@ -211,6 +221,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
         "restore_policy.py",
         "backup.py",
         "backup_manifest.py",
+        "backup_objects.py",
         "manifest.py",
         "file_copy.py",
     ):
@@ -271,9 +282,10 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
                 )
         for domain in ("postgres", "artifacts", "imports", "factory"):
             path = LIVE[domain]
-            if path.exists() or path.is_symlink():
-                if path.is_symlink() or not path.is_dir() or next(path.iterdir(), None) is not None:
-                    raise DeliveryError("cold recovery target is not clean")
+            if (path.exists() or path.is_symlink()) and (
+                path.is_symlink() or not path.is_dir() or next(path.iterdir(), None) is not None
+            ):
+                raise DeliveryError("cold recovery target is not clean")
         recovery.mkdir(mode=0o700)
         (recovery / "history").mkdir(mode=0o700)
         install.exclusive_file(recovery / "prepared.json", canonical(record), 0o600)
@@ -290,12 +302,20 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
                 # Keep new directories Root-owned and writable until verified
                 # children are complete. Final ownership/mode is applied below.
         free = os.statvfs(Path("/var/lib/kf-onlinejudge"))
-        if free.f_frsize * free.f_bavail < value["payload_bytes"] + 128 * 1024**2:
+        if free.f_frsize * free.f_bavail < backup_manifest.logical_bytes(value) + 128 * 1024**2:
             raise DeliveryError("cold recovery destination reserve insufficient")
 
         def copy(item: dict) -> tuple[int, ...]:
             path, uid, gid, mode = target(item, recovery)
-            return write_file(root / "payload" / item["path"], path, item, uid, gid, mode, deadline)
+            return write_file(
+                backup_manifest.member_path(root, value, item),
+                path,
+                item,
+                uid,
+                gid,
+                mode,
+                deadline,
+            )
 
         witnesses = file_copy.bounded_copies(value["files"], copy, deadline=deadline)
         verified = {
@@ -326,7 +346,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
             unchanged_file(path, verified[item["path"]], uid, gid, mode)
         record.update(
             data_restored=True,
-            restored_payload_bytes=value["payload_bytes"],
+            restored_payload_bytes=backup_manifest.logical_bytes(value),
             restored_files=len(value["files"]),
             restored_directories=len(value["directories"]),
             historical_configuration_active=False,

scripts/release/restore_policy.py

@@ -4,7 +4,7 @@ import json
 from pathlib import Path
 from urllib.parse import urlsplit
 
-from . import configuration, install, manifest
+from . import backup_manifest, configuration, install, manifest
 from .policy import DeliveryError, digest, fields, read_json, sha256
 
 
@@ -104,10 +104,12 @@ def compatible(backup: dict, payload: Path, target_data: dict, private_root: Pat
     never a raw cold copy into an incompatible target schema/runtime.
     """
     selected = backup["selected"]
-    old_root = private_root / "payload/releases" / selected["release_id"]
-    old = manifest.validate(read_json(old_root / "manifest.json"))
+    old_manifest = backup_manifest.named_member(
+        private_root, backup, "releases/" + selected["release_id"] + "/manifest.json"
+    )
+    old = manifest.validate(read_json(old_manifest))
     if (
-        digest(old_root / "manifest.json") != backup["manifest_sha256"]
+        digest(old_manifest) != backup["manifest_sha256"]
         or old["source_sha"] != backup["source_sha"]
         or old["images"] != backup["images"]
         or old["images"] != target_data["images"]
@@ -125,7 +127,10 @@ def compatible(backup: dict, payload: Path, target_data: dict, private_root: Pat
 
     if not migrations(old) or migrations(old) != migrations(target_data):
         raise DeliveryError("cold recovery requires identical Backend migration inventory")
-    if (private_root / "payload/postgres/PG_VERSION").read_bytes() != b"18\n":
+    if (
+        backup_manifest.named_member(private_root, backup, "postgres/PG_VERSION").read_bytes()
+        != b"18\n"
+    ):
         raise DeliveryError("cold recovery PostgreSQL major version differs")
     # Every old data byte was verified by the closed private inventory. This
     # check also binds the independently verified target public payload.
@@ -142,18 +147,28 @@ def identity_check(backup: dict, selected: dict, root: Path) -> dict:
     }
     old = backup["selected"]
     prior = {
-        name: root
-        / "payload"
-        / ("judge-config" if name == "runtime.json" else "app-config")
-        / configuration.installed_path(old["release_id"], old["configuration_id"], name).name
+        name: ("judge-config" if name == "runtime.json" else "app-config")
+        + "/"
+        + configuration.installed_path(old["release_id"], old["configuration_id"], name).name
         for name in configuration.CONFIG_FILES
     }
     inventory = {item["path"]: item for item in backup["files"]}
-    for path in prior.values():
-        relative = path.relative_to(root / "payload").as_posix()
-        if relative not in inventory or digest(path) != inventory[relative]["sha256"]:
+    for relative in prior.values():
+        if (
+            relative not in inventory
+            or digest(backup_manifest.member_path(root, backup, inventory[relative]))
+            != inventory[relative]["sha256"]
+        ):
             raise DeliveryError("old protected recovery configuration is missing")
-    previous, current = snapshot(prior), snapshot(paths)
+    previous, current = (
+        snapshot(
+            {
+                name: backup_manifest.member_path(root, backup, inventory[relative])
+                for name, relative in prior.items()
+            }
+        ),
+        snapshot(paths),
+    )
     fresh_identities(previous, current)
     cookie = Path("/var/lib/kf-onlinejudge/rabbitmq/.erlang.cookie")
     if digest(cookie) == backup["broker_cookie_sha256"]: