perf(delivery): batch installs and parallelize cold recovery

backend/tests/test_delivery.py

@@ -365,7 +365,28 @@ def test_changed_copy_retains_incomplete_file_and_never_overwrites_it(tmp_path,
     assert target.read_bytes() == b"changed"
 
 
-def test_frozen_copy_reuses_verified_records_and_checks_full_destination(tmp_path, monkeypatch):
+@pytest.fixture
+def frozen_filesystem_model(tmp_path, monkeypatch):
+    """Only fsync tiny owned fixtures; real filesystem flush remains VM-only."""
+
+    class Sync:
+        def __call__(self, descriptor):
+            assert stat.S_ISDIR(os.fstat(descriptor).st_mode)
+            for path in tmp_path.rglob("*"):
+                if path.is_file():
+                    assert path.stat().st_uid == os.getuid() and path.stat().st_size < 4096
+                    with path.open("rb") as stream:
+                        os.fsync(stream.fileno())
+            return 0
+
+    monkeypatch.setattr(
+        installer.restore_io.ctypes, "CDLL", lambda *_a, **_kw: SimpleNamespace(syncfs=Sync())
+    )
+
+
+def test_frozen_copy_reuses_verified_records_and_checks_full_destination(
+    tmp_path, monkeypatch, frozen_filesystem_model
+):
     monkeypatch.setattr(installer.preflight, "protected", lambda *args, **kwargs: None)
     source, target = tmp_path / "source", tmp_path / "target"
     source.mkdir()
@@ -390,7 +411,9 @@ def test_frozen_copy_reuses_verified_records_and_checks_full_destination(tmp_pat
     assert scanned == [target, target]
 
 
-def test_frozen_copy_rejects_destination_io_mismatch_before_acceptance(tmp_path, monkeypatch):
+def test_frozen_copy_rejects_destination_io_mismatch_before_acceptance(
+    tmp_path, monkeypatch, frozen_filesystem_model
+):
     # Simulated I/O mismatch in an ordinary owned fixture; Root protection and
     # actual disk/error isolation are still separate VM acceptance.
     monkeypatch.setattr(installer.preflight, "protected", lambda *args, **kwargs: None)
@@ -416,7 +439,9 @@ def test_frozen_copy_rejects_destination_io_mismatch_before_acceptance(tmp_path,
         installer.frozen_copy(source, target, verified_records=records)
 
 
-def test_frozen_copy_rejects_changed_source_against_trusted_records(tmp_path, monkeypatch):
+def test_frozen_copy_rejects_changed_source_against_trusted_records(
+    tmp_path, monkeypatch, frozen_filesystem_model
+):
     # Ordinary fixture bytes exercise the deferred batch validation. Actual
     # Root protection and installation are separately verified in the VM.
     monkeypatch.setattr(installer.preflight, "protected", lambda *args, **kwargs: None)
@@ -439,6 +464,30 @@ def test_frozen_copy_rejects_changed_source_against_trusted_records(tmp_path, mo
         installer.frozen_copy(source, target, verified_records=records)
 
 
+def test_frozen_copy_rejects_sync_failure_even_when_destination_bytes_match(tmp_path, monkeypatch):
+    class FailedSync:
+        def __call__(self, _descriptor):
+            return -1
+
+    monkeypatch.setattr(installer.preflight, "protected", lambda *args, **kwargs: None)
+    monkeypatch.setattr(
+        installer.restore_io.ctypes,
+        "CDLL",
+        lambda *_a, **_kw: SimpleNamespace(syncfs=FailedSync()),
+    )
+    source, target = tmp_path / "source", tmp_path / "target"
+    source.mkdir()
+    (source / "backend").mkdir()
+    payload = source / "backend/ordinary.txt"
+    payload.write_bytes(b"known ordinary bytes")
+    payload.chmod(0o444)
+    (source / "backend").chmod(0o555)
+    records = manifest.inventory(source)
+    with pytest.raises(policy.DeliveryError, match="filesystem synchronization failed"):
+        installer.frozen_copy(source, target, verified_records=records)
+    assert (target / "backend/ordinary.txt").read_bytes() == payload.read_bytes()
+
+
 def test_kernel_observation_keeps_cumulative_counters_and_rejects_duplicate_rows():
     raw = (
         "usage_usec 17\nuser_usec 11\nsystem_usec 6\nnr_periods 9\nnr_throttled 2\nthrottled_usec 3"

backend/tests/test_delivery_restore_io.py

@@ -2,6 +2,7 @@
 
 import errno
 import hashlib
+import multiprocessing
 import os
 import stat
 import time
@@ -65,6 +66,19 @@ def assert_closed(descriptors):
         assert failure.value.errno == errno.EBADF
 
 
+class ReadAfterSync:
+    """Picklable observation of actual tiny fixture reads in the child processes."""
+
+    def __init__(self, synced, reads, descriptors):
+        self.synced, self.reads, self.descriptors = synced, reads, descriptors
+
+    def __call__(self, path, item, uid, gid, mode, deadline):
+        assert self.synced.read_bytes() == b"sync completed"
+        assert all(os.fstat(fd).st_ino > 0 for fd in self.descriptors)
+        (self.reads / item["path"]).touch()
+        return restore.readback(path, item, uid, gid, mode, deadline)
+
+
 def test_bounded_batches_group_identical_objects_without_losing_paths():
     rows = [
         {"path": str(index), "sha256": str(index % 2), "bytes": 1}
@@ -89,22 +103,31 @@ def test_repeated_objects_expand_to_independent_durable_verified_destinations(
     rows, source, destination = fixture_rows(tmp_path, [b"first", b"other", b"first", b"first"])
     events = []
     target = destination(rows[0])[0].parent
+    synced, reads = tmp_path / "synced", tmp_path / "reads"
+    reads.mkdir()
     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)
+        original_sync = filesystems.sync
+
+        def sync(deadline):
+            original_sync(deadline)
+            synced.write_bytes(b"sync completed")
+
+        monkeypatch.setattr(filesystems, "sync", sync)
 
         verified = restore_io.materialize(
-            rows, source, destination, readback, filesystems, time.monotonic() + 120
+            rows,
+            source,
+            destination,
+            ReadAfterSync(synced, reads, descriptors),
+            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)
+    assert len(list(reads.iterdir())) == len(rows)
     inodes = set()
     for row in rows:
         path, uid, gid, mode = destination(row)
@@ -117,6 +140,7 @@ def test_repeated_objects_expand_to_independent_durable_verified_destinations(
 
 
 def test_existing_wrong_bytes_are_refused_without_overwrite(tmp_path, monkeypatch):
+    prior_children = {child.pid for child in multiprocessing.active_children()}
     rows, source, destination = fixture_rows(tmp_path, [b"new"])
     path, _, _, mode = destination(rows[0])
     path.write_bytes(b"bad")
@@ -128,6 +152,7 @@ def test_existing_wrong_bytes_are_refused_without_overwrite(tmp_path, monkeypatc
                 rows, source, destination, restore.readback, filesystems, time.monotonic() + 120
             )
     assert path.read_bytes() == b"bad" and restore.file_witness(path.stat()) == before
+    assert {child.pid for child in multiprocessing.active_children()} == prior_children
 
 
 def test_failed_sync_keeps_private_files_and_cannot_return_verified_witnesses(
@@ -259,3 +284,46 @@ def test_constructor_closes_earlier_cursor_when_later_anchor_is_invalid(tmp_path
         restore_io.Filesystems([valid, invalid], time.monotonic() + 120)
     assert len(opened) == 1
     assert_closed(opened)
+
+
+def test_actual_two_copy_processes_join_before_return_and_preserve_all_bytes(tmp_path, monkeypatch):
+    rows, source, destination = fixture_rows(
+        tmp_path, [b"known first", b"known second", b"known third", b"known fourth"]
+    )
+    prior_children = {child.pid for child in multiprocessing.active_children()}
+    markers = tmp_path / "worker-identities"
+    markers.mkdir()
+    original = restore_io.pending_file
+
+    def record_worker(*args):
+        (markers / str(os.getpid())).touch()
+        deadline = time.monotonic() + 5
+        while len(list(markers.iterdir())) < 2:
+            assert time.monotonic() < deadline
+            time.sleep(0.01)
+        return original(*args)
+
+    monkeypatch.setattr(restore_io, "pending_file", record_worker)
+    with closing(filesystem_model(monkeypatch, destination(rows[0])[0].parent, [])) as fs:
+        verified = restore_io.materialize(
+            rows, source, destination, restore.readback, fs, time.monotonic() + 120
+        )
+    assert len(list(markers.iterdir())) == 2
+    assert {child.pid for child in multiprocessing.active_children()} == prior_children
+    assert all(not (restore_io.Path("/proc") / p.name).exists() for p in markers.iterdir())
+    assert set(verified) == {row["path"] for row in rows}
+    for row in rows:
+        path, uid, gid, mode = destination(row)
+        assert path.read_bytes() == source(row).read_bytes()
+        restore.unchanged_file(path, verified[row["path"]], uid, gid, mode)
+
+
+def test_process_copy_refuses_fork_from_a_multithreaded_owner(tmp_path, monkeypatch):
+    rows, source, destination = fixture_rows(tmp_path, [b"known fixture"])
+    monkeypatch.setattr(restore_io.threading, "active_count", lambda: 2)
+    with closing(filesystem_model(monkeypatch, destination(rows[0])[0].parent, [])) as fs:
+        with pytest.raises(DeliveryError, match="one POSIX owner thread"):
+            restore_io.materialize(
+                rows, source, destination, restore.readback, fs, time.monotonic() + 120
+            )
+    assert not destination(rows[0])[0].exists()

scripts/release/install.py

@@ -8,7 +8,7 @@ import stat
 import subprocess
 from pathlib import Path
 
-from . import file_copy, manifest, preflight
+from . import file_copy, manifest, preflight, restore_io
 from .layout import RELEASES as RELEASES
 from .policy import DeliveryError, canonical, digest, fields, read_json, sha256
 
@@ -101,6 +101,7 @@ def _copy_verified(
     *,
     readback: bool,
     expected_bytes: int | None = None,
+    synchronize: bool = True,
 ) -> None:
     preflight.protected(source)
     sha256(expected)
@@ -137,7 +138,8 @@ def _copy_verified(
                     "source changed after bundle verification; incomplete copy retained"
                 )
             os.fchmod(outgoing, mode)
-            os.fsync(outgoing)
+            if synchronize:
+                os.fsync(outgoing)
         if readback and digest(target) != expected:
             raise DeliveryError("installed bytes differ; incomplete release retained")
     finally:
@@ -242,9 +244,6 @@ def frozen_copy(
 
 def _copy_new_tree(source: Path, target: Path, records: list[dict], directories: list[str]) -> None:
     """Create a new sealed tree; the caller must verify the complete destination."""
-    target.mkdir(mode=0o700)
-    for name in sorted(directories, key=lambda name: (name.count("/"), name)):
-        (target / name).mkdir(mode=0o700)
 
     def copy(record: dict) -> None:
         # frozen_copy verifies this image immediately; a release staging tree
@@ -256,14 +255,25 @@ def _copy_new_tree(source: Path, target: Path, records: list[dict], directories:
             int(record["mode"], 8),
             readback=False,
             expected_bytes=record["bytes"],
+            synchronize=False,
         )
 
-    file_copy.bounded_copies(records, copy)
-    for name in sorted(directories, key=lambda name: name.count("/"), reverse=True):
-        (target / name).chmod(0o555)
-        sync_directory(target / name)
-    target.chmod(0o555)
-    sync_directory(target)
+    # Retain the filesystem error cursor before creating any destination.
+    # The enclosing installation service/caller owns its existing deadline.
+    filesystems = restore_io.Filesystems([target], None)
+    try:
+        target.mkdir(mode=0o700)
+        for name in sorted(directories, key=lambda name: (name.count("/"), name)):
+            (target / name).mkdir(mode=0o700)
+        for batch in restore_io.batches(records):
+            file_copy.bounded_copies(batch, copy)
+            filesystems.sync(None)
+        for name in sorted(directories, key=lambda name: name.count("/"), reverse=True):
+            (target / name).chmod(0o555)
+        target.chmod(0o555)
+        filesystems.sync(None)
+    finally:
+        filesystems.close()
 
 
 def files(bundle: Path, expected: str) -> dict:

scripts/release/restore_io.py

@@ -1,11 +1,13 @@
 """Bounded cold-copy batches, durable before full destination byte readback."""
 
 import ctypes
+import multiprocessing
 import os
 import stat
+import threading
 import time
 from collections.abc import Callable, Iterator, Sequence
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ProcessPoolExecutor
 from pathlib import Path
 
 from . import file_copy
@@ -13,10 +15,11 @@ from .policy import DeliveryError
 
 BATCH_FILES = 256
 BATCH_BYTES = 64 * 1024**2
+COPY_PROCESSES = 2
 
 
-def remaining(deadline: float) -> None:
-    if time.monotonic() >= deadline:
+def remaining(deadline: float | None) -> None:
+    if deadline is not None and time.monotonic() >= deadline:
         raise DeliveryError("recovery copy deadline exceeded; private partial target retained")
 
 
@@ -41,7 +44,7 @@ def batches(rows: Sequence[dict]) -> Iterator[list[dict]]:
 class Filesystems:
     """Hold filesystem error cursors from before the first destination write."""
 
-    def __init__(self, paths: Sequence[Path], deadline: float):
+    def __init__(self, paths: Sequence[Path], deadline: float | None):
         self.handles: dict[int, tuple[Path, int, int]] = {}
         self.closed = False
         remaining(deadline)
@@ -82,7 +85,7 @@ class Filesystems:
         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:
+    def sync(self, deadline: float | None) -> None:
         if self.closed:
             raise DeliveryError("recovery filesystem cursor is closed")
         for device, (path, descriptor, inode) in self.handles.items():
@@ -148,6 +151,19 @@ def pending_file(
     return witness
 
 
+def _copy_group(group: list[tuple], deadline: float) -> list[tuple[int, ...]]:
+    return [
+        pending_file(source, path, item, uid, gid, mode, deadline)
+        for item, source, path, uid, gid, mode in group
+    ]
+
+
+def _read_group(readback: Callable, group: list[tuple], deadline: float) -> list[tuple[int, ...]]:
+    return [
+        readback(path, item, uid, gid, mode, deadline) for item, path, uid, gid, mode, _ in group
+    ]
+
+
 def materialize(
     rows: Sequence[dict],
     source: Callable[[dict], Path],
@@ -158,42 +174,53 @@ def materialize(
 ) -> 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.
+    Two joined child processes use the existing exclusive node CPU ceiling;
+    no private service/cgroup or extra resource budget is introduced. Each
+    worker processes a bounded half-batch without one Future per file. The
+    parent retains the pre-write error cursors and flushes only after both
+    copy groups join, then dispatches full byte readback. A batch retains at
+    most 256 file records and 64 MiB; 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)
+    if os.name != "posix" or threading.active_count() != 1:
+        raise DeliveryError("recovery process workers require one POSIX owner thread")
     verified = {}
-    with ThreadPoolExecutor(max_workers=file_copy.IN_FLIGHT) as executor:
+    executor = ProcessPoolExecutor(
+        max_workers=COPY_PROCESSES, mp_context=multiprocessing.get_context("fork")
+    )
+    # Shutdown joins every child on either success or failure. The fixed unit
+    # deadline additionally bounds a blocked syscall or unresponsive child.
+    with 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]
+            remaining(deadline)
+            group_size = max(1, (len(batch) + COPY_PROCESSES - 1) // COPY_PROCESSES)
+            groups = [
+                [
+                    (item, source(item), *destination(item))
+                    for item in batch[offset : offset + group_size]
                 ]
-                for _, path, _, _, _ in group:
+                for offset in range(0, len(batch), group_size)
+            ]
+            for group in groups:
+                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()))
+            calls = [executor.submit(_copy_group, group, deadline) for group in groups]
+            for group, call in zip(groups, calls, strict=True):
+                for (item, _, path, uid, gid, mode), witness in zip(
+                    group, call.result(), strict=True
+                ):
+                    pending.append((item, path, uid, gid, mode, witness))
             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()
+            groups = [
+                pending[offset : offset + group_size]
+                for offset in range(0, len(pending), group_size)
+            ]
+            calls = [executor.submit(_read_group, readback, group, deadline) for group in groups]
+            for group, call in zip(groups, calls, strict=True):
+                for (item, _, _, _, _, before), actual in zip(group, call.result(), strict=True):
                     if actual != before:
                         raise DeliveryError("recovery destination changed across synchronization")
                     verified[item["path"]] = actual