diff --git a/.github/workflows/engineering.yml b/.github/workflows/engineering.yml index 9fa49d2..2ef3f22 100644 --- a/.github/workflows/engineering.yml +++ b/.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 \ diff --git a/backend/tests/test_delivery_io.py b/backend/tests/test_delivery_io.py index 34f7796..73f3bae 100644 --- a/backend/tests/test_delivery_io.py +++ b/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 == [] diff --git a/scripts/release/README.md b/scripts/release/README.md index 8b78840..2108c81 100644 --- a/scripts/release/README.md +++ b/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 diff --git a/scripts/release/file_copy.py b/scripts/release/file_copy.py index 9d3dbf6..a571115 100644 --- a/scripts/release/file_copy.py +++ b/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, ...]: diff --git a/scripts/release/install.py b/scripts/release/install.py index b73398c..163d690 100644 --- a/scripts/release/install.py +++ b/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) diff --git a/scripts/release/manifest.py b/scripts/release/manifest.py index 7705fc6..99c28a6 100644 --- a/scripts/release/manifest.py +++ b/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 diff --git a/scripts/release/restore.py b/scripts/release/restore.py index 0fea5d7..c3dcfa6 100644 --- a/scripts/release/restore.py +++ b/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 ):