perf(delivery): batch durable cold recovery copies

.github/workflows/engineering.yml

@@ -20,14 +20,14 @@ jobs:
       run:
         shell: bash --noprofile --norc -euo pipefail {0}
     env:
-      CI: 'true'
+      CI: "true"
       RUSTUP_TOOLCHAIN: 1.99.0
-      CARGO_BUILD_JOBS: '2'
-      CARGO_NET_RETRY: '2'
+      CARGO_BUILD_JOBS: "2"
+      CARGO_NET_RETRY: "2"
       UV_PYTHON: 3.13.16
-      UV_NO_PROGRESS: '1'
-      UV_HTTP_TIMEOUT: '30'
-      UV_HTTP_RETRIES: '2'
+      UV_NO_PROGRESS: "1"
+      UV_HTTP_TIMEOUT: "30"
+      UV_HTTP_RETRIES: "2"
     steps:
       - name: Check out the pushed commit
         uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
@@ -110,6 +110,7 @@ jobs:
             tests/test_delivery.py \
             tests/test_delivery_io.py \
             tests/test_delivery_backup.py \
+            tests/test_delivery_restore_io.py \
             tests/test_operator_imports.py \
             tests/test_health.py \
             tests/test_identity.py \

backend/tests/test_delivery_restore_io.py

@@ -0,0 +1,261 @@
+"""Trusted ordinary byte fixtures; real Root/cgroup/syncfs/RTO remains VM-only."""
+
+import errno
+import hashlib
+import os
+import stat
+import time
+from contextlib import closing
+from types import SimpleNamespace
+
+import pytest
+from scripts.release import restore, restore_io
+from scripts.release.policy import DeliveryError
+
+
+def fixture_rows(tmp_path, contents):
+    source = tmp_path / "source"
+    target = tmp_path / "target"
+    source.mkdir()
+    target.mkdir()
+    rows = []
+    for index, raw in enumerate(contents):
+        digest = hashlib.sha256(raw).hexdigest()
+        path = source / digest
+        if not path.exists():
+            path.write_bytes(raw)
+        rows.append({"path": str(index), "bytes": len(raw), "sha256": digest})
+    return (
+        rows,
+        lambda row: source / row["sha256"],
+        lambda row: (
+            target / row["path"],
+            os.getuid(),
+            os.getgid(),
+            0o600 if int(row["path"]) % 2 else 0o400,
+        ),
+    )
+
+
+def filesystem_model(monkeypatch, target, events, callback=None):
+    """Model syncfs using only tiny owned fixture fsyncs; never host syncfs."""
+
+    class Sync:
+        def __call__(self, descriptor):
+            assert stat.S_ISDIR(os.fstat(descriptor).st_mode)
+            if callback is not None:
+                callback()
+            files = [p for p in target.iterdir() if p.is_file()]
+            for path in files:
+                with path.open("rb") as stream:
+                    os.fsync(stream.fileno())
+            events.append(("sync", descriptor))
+            return 0
+
+    monkeypatch.setattr(
+        restore_io.ctypes, "CDLL", lambda *_a, **_kw: SimpleNamespace(syncfs=Sync())
+    )
+    return restore_io.Filesystems([target], time.monotonic() + 120)
+
+
+def assert_closed(descriptors):
+    for descriptor in descriptors:
+        with pytest.raises(OSError) as failure:
+            os.fstat(descriptor)
+        assert failure.value.errno == errno.EBADF
+
+
+def test_bounded_batches_group_identical_objects_without_losing_paths():
+    rows = [
+        {"path": str(index), "sha256": str(index % 2), "bytes": 1}
+        for index in range(restore_io.BATCH_FILES + 4)
+    ]
+    rows.append({"path": "large", "sha256": "2", "bytes": restore_io.BATCH_BYTES + 1})
+    groups = list(restore_io.batches(rows))
+    flattened = [row for group in groups for row in group]
+    assert {row["path"] for row in flattened} == {row["path"] for row in rows}
+    assert len(flattened) == len(rows)
+    assert [row["sha256"] for row in flattened] == sorted(row["sha256"] for row in rows)
+    assert all(len(group) <= restore_io.BATCH_FILES for group in groups)
+    assert all(
+        sum(row["bytes"] for row in group) <= restore_io.BATCH_BYTES or len(group) == 1
+        for group in groups
+    )
+
+
+def test_repeated_objects_expand_to_independent_durable_verified_destinations(
+    tmp_path, monkeypatch
+):
+    rows, source, destination = fixture_rows(tmp_path, [b"first", b"other", b"first", b"first"])
+    events = []
+    target = destination(rows[0])[0].parent
+    with closing(filesystem_model(monkeypatch, target, events)) as filesystems:
+        descriptors = [descriptor for _, descriptor, _ in filesystems.handles.values()]
+        assert list(target.iterdir()) == []
+
+        def readback(path, item, uid, gid, mode, deadline):
+            assert events and events[0][0] == "sync"
+            assert all(os.fstat(fd).st_ino > 0 for fd in descriptors)
+            events.append(("read", item["path"]))
+            return restore.readback(path, item, uid, gid, mode, deadline)
+
+        verified = restore_io.materialize(
+            rows, source, destination, readback, filesystems, time.monotonic() + 120
+        )
+    assert_closed(descriptors)
+    assert set(verified) == {row["path"] for row in rows}
+    assert len([event for event in events if event[0] == "read"]) == len(rows)
+    inodes = set()
+    for row in rows:
+        path, uid, gid, mode = destination(row)
+        assert path.read_bytes() == source(row).read_bytes()
+        assert stat.S_IMODE(path.stat().st_mode) == mode
+        assert path.stat().st_nlink == 1
+        inodes.add(path.stat().st_ino)
+        restore.unchanged_file(path, verified[row["path"]], uid, gid, mode)
+    assert len(inodes) == len(rows)
+
+
+def test_existing_wrong_bytes_are_refused_without_overwrite(tmp_path, monkeypatch):
+    rows, source, destination = fixture_rows(tmp_path, [b"new"])
+    path, _, _, mode = destination(rows[0])
+    path.write_bytes(b"bad")
+    path.chmod(mode)
+    before = restore.file_witness(path.stat())
+    with closing(filesystem_model(monkeypatch, path.parent, [])) as filesystems:
+        with pytest.raises(DeliveryError, match="byte verification"):
+            restore_io.materialize(
+                rows, source, destination, restore.readback, filesystems, time.monotonic() + 120
+            )
+    assert path.read_bytes() == b"bad" and restore.file_witness(path.stat()) == before
+
+
+def test_failed_sync_keeps_private_files_and_cannot_return_verified_witnesses(
+    tmp_path, monkeypatch
+):
+    rows, source, destination = fixture_rows(tmp_path, [b"a", b"b", b"c"])
+    reads = []
+
+    def failed_sync():
+        raise DeliveryError("synthetic synchronization failure")
+
+    with closing(
+        filesystem_model(monkeypatch, destination(rows[0])[0].parent, [], failed_sync)
+    ) as filesystems:
+        descriptors = [fd for _, fd, _ in filesystems.handles.values()]
+        with pytest.raises(DeliveryError, match="synthetic synchronization failure"):
+            restore_io.materialize(
+                rows,
+                source,
+                destination,
+                lambda *args: reads.append(args),
+                filesystems,
+                time.monotonic() + 120,
+            )
+    assert_closed(descriptors)
+    assert reads == []
+    assert all(destination(row)[0].is_file() for row in rows)
+
+
+def test_same_bytes_inode_replacement_across_sync_cannot_adopt_new_file(tmp_path, monkeypatch):
+    rows, source, destination = fixture_rows(tmp_path, [b"same"])
+
+    def replace_inode():
+        path, _, _, mode = destination(rows[0])
+        replacement = path.parent / "replacement"
+        replacement.write_bytes(path.read_bytes())
+        replacement.chmod(mode)
+        replacement.replace(path)
+
+    with closing(
+        filesystem_model(monkeypatch, destination(rows[0])[0].parent, [], replace_inode)
+    ) as filesystems:
+        with pytest.raises(DeliveryError, match="changed across synchronization"):
+            restore_io.materialize(
+                rows, source, destination, restore.readback, filesystems, time.monotonic() + 120
+            )
+
+
+def test_expired_batch_deadline_copies_no_files(tmp_path, monkeypatch):
+    rows, source, destination = fixture_rows(tmp_path, [b"data"])
+    with closing(filesystem_model(monkeypatch, destination(rows[0])[0].parent, [])) as filesystems:
+        with pytest.raises(DeliveryError, match="deadline exceeded"):
+            restore_io.materialize(
+                rows, source, destination, restore.readback, filesystems, time.monotonic() - 1
+            )
+    assert not destination(rows[0])[0].exists()
+
+
+def test_syncfs_failure_is_propagated_in_the_ordinary_syscall_model(tmp_path, monkeypatch):
+    class FailedSync:
+        def __call__(self, _descriptor):
+            return -1
+
+    monkeypatch.setattr(
+        restore_io.ctypes, "CDLL", lambda *_a, **_kw: SimpleNamespace(syncfs=FailedSync())
+    )
+    with closing(restore_io.Filesystems([tmp_path], time.monotonic() + 120)) as filesystems:
+        descriptors = [fd for _, fd, _ in filesystems.handles.values()]
+        with pytest.raises(DeliveryError, match="filesystem synchronization failed"):
+            filesystems.sync(time.monotonic() + 120)
+    assert_closed(descriptors)
+
+
+def test_cursor_is_open_before_first_write_and_held_across_batches(tmp_path, monkeypatch):
+    rows, source, destination = fixture_rows(tmp_path, [b"a", b"b", b"c", b"d", b"e"])
+    target = destination(rows[0])[0].parent
+    monkeypatch.setattr(restore_io, "BATCH_FILES", 2)
+    events = []
+    with closing(filesystem_model(monkeypatch, target, events)) as filesystems:
+        descriptors = [fd for _, fd, _ in filesystems.handles.values()]
+        before = {fd: (os.fstat(fd).st_dev, os.fstat(fd).st_ino) for fd in descriptors}
+        real_write = restore_io.pending_file
+
+        def write(*args):
+            assert {fd: (os.fstat(fd).st_dev, os.fstat(fd).st_ino) for fd in descriptors} == before
+            return real_write(*args)
+
+        monkeypatch.setattr(restore_io, "pending_file", write)
+        verified = restore_io.materialize(
+            rows, source, destination, restore.readback, filesystems, time.monotonic() + 120
+        )
+        assert len(verified) == len(rows)
+        assert len(events) == 3
+        assert {fd for event, fd in events if event == "sync"} == set(descriptors)
+    assert_closed(descriptors)
+
+
+def test_cursor_can_anchor_before_target_directories_exist(tmp_path, monkeypatch):
+    target = tmp_path / "later" / "nested"
+
+    class Sync:
+        def __call__(self, descriptor):
+            assert os.fstat(descriptor).st_ino == tmp_path.stat().st_ino
+            return 0
+
+    monkeypatch.setattr(
+        restore_io.ctypes, "CDLL", lambda *_a, **_kw: SimpleNamespace(syncfs=Sync())
+    )
+    with closing(restore_io.Filesystems([target], time.monotonic() + 120)) as filesystems:
+        target.mkdir(parents=True)
+        filesystems.contains(target / "file")
+        filesystems.sync(time.monotonic() + 120)
+
+
+def test_constructor_closes_earlier_cursor_when_later_anchor_is_invalid(tmp_path, monkeypatch):
+    valid, invalid = tmp_path / "valid", tmp_path / "invalid"
+    valid.mkdir()
+    invalid.write_bytes(b"ordinary owned fixture")
+    opened = []
+    real_open = restore_io.os.open
+
+    def record_open(*args, **kwargs):
+        descriptor = real_open(*args, **kwargs)
+        opened.append(descriptor)
+        return descriptor
+
+    monkeypatch.setattr(restore_io.os, "open", record_open)
+    with pytest.raises(DeliveryError, match="anchor differs"):
+        restore_io.Filesystems([valid, invalid], time.monotonic() + 120)
+    assert len(opened) == 1
+    assert_closed(opened)

scripts/release/restore.py

@@ -22,6 +22,7 @@ from . import (
     lifecycle,
     manifest,
     preflight,
+    restore_io,
     restore_policy,
 )
 from .policy import DeliveryError, canonical, digest, sha256
@@ -219,6 +220,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
     for name in (
         "restore.py",
         "restore_policy.py",
+        "restore_io.py",
         "backup.py",
         "backup_manifest.py",
         "backup_objects.py",
@@ -268,6 +270,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
         "execution_verified": False,
     }
     deadline = time.monotonic() + 1700
+    filesystems = None
     try:
         fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
         stopped()
@@ -286,6 +289,8 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
                 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 value["schema_version"] == 4:
+            filesystems = restore_io.Filesystems([recovery, *LIVE.values()], deadline)
         recovery.mkdir(mode=0o700)
         (recovery / "history").mkdir(mode=0o700)
         install.exclusive_file(recovery / "prepared.json", canonical(record), 0o600)
@@ -317,10 +322,22 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
                 deadline,
             )
 
-        witnesses = file_copy.bounded_copies(value["files"], copy, deadline=deadline)
-        verified = {
-            item["path"]: witness for item, witness in zip(value["files"], witnesses, strict=True)
-        }
+        if value["schema_version"] == 4:
+            assert filesystems is not None
+            verified = restore_io.materialize(
+                value["files"],
+                lambda item: backup_manifest.member_path(root, value, item),
+                lambda item: target(item, recovery),
+                readback,
+                filesystems,
+                deadline,
+            )
+        else:
+            witnesses = file_copy.bounded_copies(value["files"], copy, deadline=deadline)
+            verified = {
+                item["path"]: witness
+                for item, witness in zip(value["files"], witnesses, strict=True)
+            }
         for item in sorted(
             value["directories"], key=lambda row: row["path"].count("/"), reverse=True
         ):
@@ -330,12 +347,20 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
                 os.chown(path, uid, gid, follow_symlinks=False)
             if stat.S_IMODE(current.st_mode) != mode:
                 path.chmod(mode)
-            install.sync_directory(path)
+            if value["schema_version"] != 4:
+                install.sync_directory(path)
             member(path, uid, gid, mode, directory=True)
         # Flush directory entries once after their children, including the two
         # parents whose new domain roots are outside the backup's directory list.
-        install.sync_directory(recovery / "history")
-        install.sync_directory(LIVE["postgres"].parent)
+        if value["schema_version"] == 4:
+            assert filesystems is not None
+            filesystems.sync(deadline)
+            for item in value["directories"]:
+                path, uid, gid, mode = target(item, recovery)
+                member(path, uid, gid, mode, directory=True)
+        else:
+            install.sync_directory(recovery / "history")
+            install.sync_directory(LIVE["postgres"].parent)
         stopped()
         if install.selection() != selected:
             raise DeliveryError("selected recovery identity changed")
@@ -344,6 +369,9 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
                 raise DeliveryError("recovery verification deadline exceeded")
             path, uid, gid, mode = target(item, recovery)
             unchanged_file(path, verified[item["path"]], uid, gid, mode)
+        if filesystems is not None:
+            filesystems.close()
+            filesystems = None
         record.update(
             data_restored=True,
             restored_payload_bytes=backup_manifest.logical_bytes(value),
@@ -357,7 +385,11 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
         install.sync_directory(STATE)
         return record
     finally:
-        os.close(descriptor)
+        try:
+            if filesystems is not None:
+                filesystems.close()
+        finally:
+            os.close(descriptor)
 
 
 def launch(recovery_id: str, backup_id: str, expected: str) -> dict:

scripts/release/restore_io.py

@@ -0,0 +1,201 @@
+"""Bounded cold-copy batches, durable before full destination byte readback."""
+
+import ctypes
+import os
+import stat
+import time
+from collections.abc import Callable, Iterator, Sequence
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+from . import file_copy
+from .policy import DeliveryError
+
+BATCH_FILES = 256
+BATCH_BYTES = 64 * 1024**2
+
+
+def remaining(deadline: float) -> None:
+    if time.monotonic() >= deadline:
+        raise DeliveryError("recovery copy deadline exceeded; private partial target retained")
+
+
+def batches(rows: Sequence[dict]) -> Iterator[list[dict]]:
+    """Keep repeated objects adjacent; large files have an exclusive batch."""
+    current: list[dict] = []
+    size = 0
+    for row in sorted(rows, key=lambda item: (item["sha256"], item["path"])):
+        count = row["bytes"]
+        if current and (len(current) >= BATCH_FILES or size + count > BATCH_BYTES):
+            yield current
+            current, size = [], 0
+        current.append(row)
+        size += count
+        if size >= BATCH_BYTES:
+            yield current
+            current, size = [], 0
+    if current:
+        yield current
+
+
+class Filesystems:
+    """Hold filesystem error cursors from before the first destination write."""
+
+    def __init__(self, paths: Sequence[Path], deadline: float):
+        self.handles: dict[int, tuple[Path, int, int]] = {}
+        self.closed = False
+        remaining(deadline)
+        libc = ctypes.CDLL(None, use_errno=True)
+        try:
+            self._syncfs = libc.syncfs
+        except AttributeError as error:
+            raise DeliveryError("recovery filesystem synchronization unavailable") from error
+        self._syncfs.argtypes, self._syncfs.restype = [ctypes.c_int], ctypes.c_int
+        try:
+            for path in paths:
+                remaining(deadline)
+                while not path.exists() and not path.is_symlink():
+                    if path.parent == path:
+                        raise DeliveryError("recovery filesystem ancestor unavailable")
+                    path = path.parent
+                before = path.lstat()
+                if not stat.S_ISDIR(before.st_mode):
+                    raise DeliveryError("recovery filesystem anchor differs")
+                if before.st_dev in self.handles:
+                    continue
+                descriptor = os.open(
+                    path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
+                )
+                try:
+                    if file_copy._identity(os.fstat(descriptor)) != file_copy._identity(before):
+                        raise DeliveryError("recovery filesystem anchor changed")
+                except BaseException:
+                    os.close(descriptor)
+                    raise
+                self.handles[before.st_dev] = (path, descriptor, before.st_ino)
+        except BaseException:
+            self.close()
+            raise
+
+    def contains(self, path: Path) -> None:
+        parent = path.parent.lstat()
+        if self.closed or not stat.S_ISDIR(parent.st_mode) or parent.st_dev not in self.handles:
+            raise DeliveryError("recovery destination has no pre-write filesystem cursor")
+
+    def sync(self, deadline: float) -> None:
+        if self.closed:
+            raise DeliveryError("recovery filesystem cursor is closed")
+        for device, (path, descriptor, inode) in self.handles.items():
+            remaining(deadline)
+            current, opened = path.lstat(), os.fstat(descriptor)
+            # This operator legitimately changes directory entries and modes.
+            # The retained descriptor must still refer to the same directory.
+            if any(
+                not stat.S_ISDIR(info.st_mode) or (info.st_dev, info.st_ino) != (device, inode)
+                for info in (current, opened)
+            ):
+                raise DeliveryError("recovery filesystem anchor changed")
+            if self._syncfs(descriptor) != 0:
+                raise DeliveryError("recovery filesystem synchronization failed")
+            remaining(deadline)
+
+    def close(self) -> None:
+        if not self.closed:
+            self.closed = True
+            error = None
+            for _, descriptor, _ in self.handles.values():
+                try:
+                    os.close(descriptor)
+                except OSError as failure:
+                    error = failure
+            if error is not None:
+                raise error
+
+
+def pending_file(
+    source: Path, path: Path, item: dict, uid: int, gid: int, mode: int, deadline: float
+) -> tuple[int, ...]:
+    """Materialize only; this witness cannot certify bytes before batch readback."""
+    remaining(deadline)
+    if path.exists() or path.is_symlink():
+        info = path.lstat()
+        if (
+            not stat.S_ISREG(info.st_mode)
+            or info.st_nlink != 1
+            or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode), info.st_size)
+            != (uid, gid, mode, item["bytes"])
+        ):
+            raise DeliveryError("existing recovery destination differs; no overwrite")
+        return file_copy._identity(info)
+    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,
+            ),
+            "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)
+        witness = file_copy._identity(os.fstat(outgoing.fileno()))
+    if file_copy._identity(path.lstat()) != witness:
+        raise DeliveryError("recovery destination changed before synchronization")
+    return witness
+
+
+def materialize(
+    rows: Sequence[dict],
+    source: Callable[[dict], Path],
+    destination: Callable[[dict], tuple[Path, int, int, int]],
+    readback: Callable[[Path, dict, int, int, int, float], tuple[int, ...]],
+    filesystems: Filesystems,
+    deadline: float,
+) -> dict[str, tuple[int, ...]]:
+    """Copy, join, sync and independently read each complete destination.
+
+    Only four calls are dispatched at a time. A batch retains at most 256 file
+    records and 64 MiB of logical contents; a larger file is handled alone.
+    Every returned witness follows an actual filesystem flush and full SHA256
+    destination read. Exceptions join outstanding work and retain partial files.
+    """
+    remaining(deadline)
+    verified = {}
+    with ThreadPoolExecutor(max_workers=file_copy.IN_FLIGHT) as executor:
+        for batch in batches(rows):
+            pending = []
+            for offset in range(0, len(batch), file_copy.IN_FLIGHT):
+                remaining(deadline)
+                group = [
+                    (item, *destination(item))
+                    for item in batch[offset : offset + file_copy.IN_FLIGHT]
+                ]
+                for _, path, _, _, _ in group:
+                    filesystems.contains(path)
+                calls = [
+                    executor.submit(
+                        pending_file, source(item), path, item, uid, gid, mode, deadline
+                    )
+                    for item, path, uid, gid, mode in group
+                ]
+                for record, call in zip(group, calls, strict=True):
+                    pending.append((*record, call.result()))
+            filesystems.sync(deadline)
+            for offset in range(0, len(pending), file_copy.IN_FLIGHT):
+                remaining(deadline)
+                group = pending[offset : offset + file_copy.IN_FLIGHT]
+                calls = [
+                    executor.submit(readback, path, item, uid, gid, mode, deadline)
+                    for item, path, uid, gid, mode, _ in group
+                ]
+                for (item, _, _, _, _, before), call in zip(group, calls, strict=True):
+                    actual = call.result()
+                    if actual != before:
+                        raise DeliveryError("recovery destination changed across synchronization")
+                    verified[item["path"]] = actual
+    remaining(deadline)
+    return verified