From 08bac1ad848b21d279ad445efbdc736fab6a8cd3 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Tue, 27 Oct 2026 23:12:07 +0800 Subject: [PATCH] perf(delivery): batch durable cold recovery copies --- .github/workflows/engineering.yml | 13 +- backend/tests/test_delivery_restore_io.py | 261 ++++++++++++++++++++++ scripts/release/restore.py | 48 +++- scripts/release/restore_io.py | 201 +++++++++++++++++ 4 files changed, 509 insertions(+), 14 deletions(-) create mode 100644 backend/tests/test_delivery_restore_io.py create mode 100644 scripts/release/restore_io.py diff --git a/.github/workflows/engineering.yml b/.github/workflows/engineering.yml index 30bf575..585f6ee 100644 --- a/.github/workflows/engineering.yml +++ b/.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 \ diff --git a/backend/tests/test_delivery_restore_io.py b/backend/tests/test_delivery_restore_io.py new file mode 100644 index 0000000..586ca32 --- /dev/null +++ b/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) diff --git a/scripts/release/restore.py b/scripts/release/restore.py index 4ebf7b9..73cbd4c 100644 --- a/scripts/release/restore.py +++ b/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: diff --git a/scripts/release/restore_io.py b/scripts/release/restore_io.py new file mode 100644 index 0000000..1b43f39 --- /dev/null +++ b/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