perf(delivery): bound copy I/O and streamline metadata scans

.github/workflows/engineering.yml

@@ -108,6 +108,7 @@ jobs:
             tests/test_dependency_errors.py \
             tests/test_production_logging.py \
             tests/test_delivery.py \
+            tests/test_delivery_io.py \
             tests/test_operator_imports.py \
             tests/test_health.py \
             tests/test_identity.py \

backend/tests/test_delivery_io.py

@@ -6,7 +6,9 @@ import importlib
 import json
 import os
 import sys
+import threading
 import time
+from collections.abc import Sequence
 from pathlib import Path
 
 import pytest
@@ -200,3 +202,114 @@ def test_fixed_regular_transfer_preserves_exact_bytes_and_offsets(
         file_copy.transfer(incoming, outgoing, length, deadline=time.monotonic() + 30)
     assert target.read_bytes() == content
     assert policy.digest(target) == policy.digest(source)
+
+
+def test_copy_inventory_prefetch_is_bounded_and_results_keep_input_order():
+    # Trusted callbacks only: ordinary concurrency control, no VM/Root acceptance.
+    release, saturated, second = (threading.Event() for _ in range(3))
+    lock = threading.Lock()
+    consumed, results, errors = [], [], []
+    active = maximum = 0
+
+    class Inventory(Sequence):
+        def __len__(self):
+            return 32
+
+        def __getitem__(self, index):
+            if not 0 <= index < len(self):
+                raise IndexError
+            with lock:
+                consumed.append(index)
+            return index
+
+    def copy(index):
+        nonlocal active, maximum
+        with lock:
+            active += 1
+            maximum = max(active, maximum)
+            if active == 4:
+                saturated.set()
+        try:
+            assert release.wait(5)
+            if index == 0:
+                assert second.wait(5)
+            if index == 1:
+                second.set()
+            return index * 3
+        finally:
+            with lock:
+                active -= 1
+
+    def collect():
+        try:
+            results.extend(file_copy.bounded_copies(Inventory(), copy))
+        except BaseException as error:
+            errors.append(error)
+
+    collector = threading.Thread(target=collect)
+    collector.start()
+    try:
+        assert saturated.wait(5)
+        with lock:
+            assert consumed == [0, 1, 2, 3]
+            assert maximum == 4
+    finally:
+        release.set()
+        collector.join(10)
+    assert not collector.is_alive() and not errors
+    assert results == [index * 3 for index in range(32)]
+    assert active == 0 and maximum == 4
+
+
+def test_copy_failure_joins_dispatched_calls_before_propagating():
+    # No files/privileges: observe failure quiescence of trusted callbacks.
+    together = threading.Barrier(4)
+    release, failing, returned = (threading.Event() for _ in range(3))
+    lock = threading.Lock()
+    active = 0
+    errors, active_when_returned = [], []
+
+    def copy(index):
+        nonlocal active
+        with lock:
+            active += 1
+        try:
+            if index < 4:
+                together.wait(timeout=5)
+            if index == 0:
+                failing.set()
+                raise policy.DeliveryError("ordinary callback refusal model")
+            assert release.wait(5)
+            return index
+        finally:
+            with lock:
+                active -= 1
+
+    def collect():
+        try:
+            file_copy.bounded_copies(list(range(32)), copy)
+        except BaseException as error:
+            errors.append(error)
+        finally:
+            with lock:
+                active_when_returned.append(active)
+            returned.set()
+
+    collector = threading.Thread(target=collect)
+    collector.start()
+    try:
+        assert failing.wait(5)
+        assert not returned.wait(0.05)
+    finally:
+        release.set()
+        collector.join(10)
+    assert not collector.is_alive() and active_when_returned == [0]
+    assert len(errors) == 1 and isinstance(errors[0], policy.DeliveryError)
+    assert str(errors[0]) == "ordinary callback refusal model"
+
+
+def test_expired_copy_deadline_refuses_before_dispatch():
+    calls = []
+    with pytest.raises(policy.DeliveryError, match="copy deadline exceeded"):
+        file_copy.bounded_copies(list(range(8)), calls.append, deadline=time.monotonic() - 1)
+    assert calls == []

scripts/release/README.md

@@ -408,6 +408,12 @@ failure. Further bundles receive full verification; separate commands start fres
 Fixed-size regular-file copies can use Linux kernel transfer with bounded chunks.
 Unsupported transfer falls back at the same offsets. Source size/EOF, file fsync
 and independent destination byte verification remain required.
+Verified installer and stopped-node recovery inventories can keep at most four
+file operations in flight. Each file retains its durability and byte checks;
+directory finalization and success receipts wait for all dispatched work. On
+failure, queued work is cancelled and running calls are joined before the error
+returns. Private partial data is retained for inspection. The original operator
+CPU, memory, PID and deadline limits still apply.
 Old configuration and maintenance receipts go into inactive
 `recovery/ID/history/`. The new boot's Sandbox preparation owns cgroup recreation.
 Unknown/nonempty targets and differing existing files are refused; partial restores

scripts/release/file_copy.py

@@ -4,12 +4,63 @@ import errno
 import os
 import stat
 import time
-from typing import BinaryIO
+from collections.abc import Callable, Sequence
+from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
+from itertools import islice
+from typing import BinaryIO, cast
 
 from .policy import MAX_FILE_BYTES, DeliveryError
 
 CHUNK = 1024 * 1024
 UNSUPPORTED = {errno.ENOSYS, errno.EINVAL, errno.EOPNOTSUPP, errno.EXDEV}
+IN_FLIGHT = 4
+
+
+def bounded_copies[Item, Result](
+    rows: Sequence[Item],
+    operation: Callable[[Item], Result],
+    *,
+    deadline: float | None = None,
+) -> list[Result]:
+    """Join a finite verified inventory, with at most four operations in flight."""
+
+    def remaining() -> float | None:
+        if deadline is None:
+            return None
+        value = deadline - time.monotonic()
+        if value <= 0:
+            raise DeliveryError("copy deadline exceeded; private partial target retained")
+        return value
+
+    remaining()
+    if len(rows) < 2:
+        return [operation(row) for row in rows]
+    results: list[Result | None] = [None] * len(rows)
+    source = iter(enumerate(rows))
+    with ThreadPoolExecutor(max_workers=min(IN_FLIGHT, len(rows))) as executor:
+        pending = {
+            executor.submit(operation, row): index for index, row in islice(source, IN_FLIGHT)
+        }
+        try:
+            while pending:
+                ready, _ = wait(pending, timeout=remaining(), return_when=FIRST_COMPLETED)
+                if not ready:
+                    raise DeliveryError("copy deadline exceeded; private partial target retained")
+                # Observe every completed failure before dispatching more work.
+                completed = [(pending.pop(future), future.result()) for future in ready]
+                for index, result in completed:
+                    results[index] = result
+                    following = next(source, None)
+                    if following is not None:
+                        index, row = following
+                        remaining()
+                        pending[executor.submit(operation, row)] = index
+        finally:
+            for future in pending:
+                future.cancel()
+            # The executor joins running calls before this function can return
+            # or propagate an error. No success receipt can outrun file I/O.
+    return cast(list[Result], results)
 
 
 def _identity(info: os.stat_result) -> tuple[int, ...]:

scripts/release/install.py

@@ -235,7 +235,8 @@ def frozen_copy(
     target.mkdir(mode=0o700)
     for name in sorted(directories, key=lambda name: (name.count("/"), name)):
         (target / name).mkdir(mode=0o700)
-    for record in records:
+
+    def copy(record: dict) -> None:
         # The complete sealed tree is read back immediately below. Avoid
         # hashing each destination both here and again in that inventory.
         _copy_verified(
@@ -246,6 +247,8 @@ def frozen_copy(
             readback=False,
             expected_bytes=record["bytes"],
         )
+
+    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)

scripts/release/manifest.py

@@ -45,24 +45,24 @@ _MAX_SCOPED_METADATA = 2 * MAX_METADATA_BYTES
 
 def _bundle_signature(bundle: Path) -> dict[str, tuple[int, ...]]:
     """Observe topology and all metadata that can invalidate a prior byte read."""
-    if not bundle.is_dir() or bundle.is_symlink():
-        raise DeliveryError("bundle directory required")
-    members = set()
-    for path in bundle.iterdir():
-        members.add(path.name)
-        if len(members) > 2:
-            raise DeliveryError("unexpected bundle member")
-    if members != {"manifest.json", "payload"}:
+    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
+    descriptor = os.open(bundle, flags)
+    members = {}
+    try:
+        root = os.fstat(descriptor)
+        with os.scandir(descriptor) as entries:
+            for entry in entries:
+                members[entry.name] = entry.stat(follow_symlinks=False)
+                if len(members) > 2:
+                    raise DeliveryError("unexpected bundle member")
+    finally:
+        os.close(descriptor)
+    if set(members) != {"manifest.json", "payload"}:
         raise DeliveryError("unexpected bundle member")
     result = {}
 
-    def add(path: Path, *, directory: bool) -> None:
-        info = path.lstat()
-        if not (stat.S_ISDIR(info.st_mode) if directory else stat.S_ISREG(info.st_mode)):
-            raise DeliveryError("bundle proof requires ordinary directories and files")
-        if not directory and info.st_nlink != 1:
-            raise DeliveryError("bundle proof requires single-link files")
-        result[path.relative_to(bundle).as_posix()] = (
+    def identity(info: os.stat_result) -> tuple[int, ...]:
+        return (
             info.st_dev,
             info.st_ino,
             info.st_mode,
@@ -73,25 +73,42 @@ def _bundle_signature(bundle: Path) -> dict[str, tuple[int, ...]]:
             info.st_mtime_ns,
             info.st_ctime_ns,
         )
+
+    def add(name: str, info: os.stat_result, *, directory: bool) -> None:
+        if not (stat.S_ISDIR(info.st_mode) if directory else stat.S_ISREG(info.st_mode)):
+            raise DeliveryError("bundle proof requires ordinary directories and files")
+        if not directory and info.st_nlink != 1:
+            raise DeliveryError("bundle proof requires single-link files")
+        result[name] = identity(info)
         if len(result) > MAX_FILES + 3:
             raise DeliveryError("bundle proof entry count limit")
 
-    def failed(error: OSError) -> None:
-        raise error
-
-    add(bundle, directory=True)
-    add(bundle / "manifest.json", directory=False)
+    add(".", root, directory=True)
+    add("manifest.json", members["manifest.json"], directory=False)
     payload = bundle / "payload"
-    add(payload, directory=True)
-    for directory, names, files in os.walk(payload, followlinks=False, onerror=failed):
-        for name in names:
-            path = Path(directory) / name
-            relative_name(path.relative_to(payload).as_posix() + "/.__directory__")
-            add(path, directory=True)
-        for name in files:
-            path = Path(directory) / name
-            relative_name(path.relative_to(payload).as_posix())
-            add(path, directory=False)
+    add("payload", members["payload"], directory=True)
+    pending = [(str(payload), "", result["payload"])]
+    while pending:
+        directory, prefix, expected = pending.pop()
+        # Close each iterator before visiting another directory. Wide or deep
+        # verified inventories never require one open descriptor per directory.
+        descriptor = os.open(directory, flags)
+        try:
+            if identity(os.fstat(descriptor)) != expected:
+                raise DeliveryError("verified bundle changed during the operator operation")
+            with os.scandir(descriptor) as entries:
+                for entry in entries:
+                    name = prefix + entry.name
+                    info = entry.stat(follow_symlinks=False)
+                    is_directory = stat.S_ISDIR(info.st_mode)
+                    relative_name(name + "/.__directory__" if is_directory else name)
+                    add("payload/" + name, info, directory=is_directory)
+                    if is_directory:
+                        pending.append(
+                            (os.path.join(directory, entry.name), name + "/", identity(info))
+                        )
+        finally:
+            os.close(descriptor)
     return result
 
 

scripts/release/restore.py

@@ -121,7 +121,7 @@ def target(record: dict, recovery: Path) -> tuple[Path, int, int, int]:
     return path, 0, 0, 0o700 if "bytes" not in record else 0o600
 
 
-def member(path: Path, uid: int, gid: int, mode: int, *, directory: bool) -> None:
+def member(path: Path, uid: int, gid: int, mode: int, *, directory: bool) -> os.stat_result:
     info = path.lstat()
     if (
         (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (uid, gid, mode)
@@ -129,6 +129,7 @@ def member(path: Path, uid: int, gid: int, mode: int, *, directory: bool) -> Non
         or (not directory and info.st_nlink != 1)
     ):
         raise DeliveryError("existing recovery destination differs; no overwrite")
+    return info
 
 
 def file_witness(info: os.stat_result) -> tuple[int, ...]:
@@ -148,8 +149,7 @@ def file_witness(info: os.stat_result) -> tuple[int, ...]:
 def readback(
     path: Path, item: dict, uid: int, gid: int, mode: int, deadline: float
 ) -> tuple[int, ...]:
-    member(path, uid, gid, mode, directory=False)
-    before = file_witness(path.lstat())
+    before = file_witness(member(path, uid, gid, mode, directory=False))
     descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
     with os.fdopen(descriptor, "rb") as stream:
         if file_witness(os.fstat(stream.fileno())) != before:
@@ -169,8 +169,7 @@ def readback(
 
 
 def unchanged_file(path: Path, witness: tuple[int, ...], uid: int, gid: int, mode: int) -> None:
-    member(path, uid, gid, mode, directory=False)
-    if file_witness(path.lstat()) != witness:
+    if file_witness(member(path, uid, gid, mode, directory=False)) != witness:
         raise DeliveryError("verified recovery file changed before completion")
 
 
@@ -290,12 +289,15 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict:
         free = os.statvfs(Path("/var/lib/kf-onlinejudge"))
         if free.f_frsize * free.f_bavail < value["payload_bytes"] + 128 * 1024**2:
             raise DeliveryError("cold recovery destination reserve insufficient")
-        verified = {}
-        for item in value["files"]:
+
+        def copy(item: dict) -> tuple[int, ...]:
             path, uid, gid, mode = target(item, recovery)
-            verified[item["path"]] = write_file(
-                root / "payload" / item["path"], path, item, uid, gid, mode, deadline
-            )
+            return write_file(root / "payload" / item["path"], path, item, uid, gid, mode, 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)
+        }
         for item in sorted(
             value["directories"], key=lambda row: row["path"].count("/"), reverse=True
         ):