From db834dfc73d23c9de3ffe68a75f9a89bffdf9d6e Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Wed, 28 Oct 2026 04:56:18 +0800 Subject: [PATCH] perf(delivery): reuse verified readback bytes within batches --- backend/tests/test_delivery_restore_io.py | 192 +++++++++++++++++++++- scripts/release/README.md | 15 +- scripts/release/restore.py | 85 +++++++++- scripts/release/restore_io.py | 15 +- 4 files changed, 288 insertions(+), 19 deletions(-) diff --git a/backend/tests/test_delivery_restore_io.py b/backend/tests/test_delivery_restore_io.py index 53809c1..e886456 100644 --- a/backend/tests/test_delivery_restore_io.py +++ b/backend/tests/test_delivery_restore_io.py @@ -7,6 +7,7 @@ import os import stat import time from contextlib import closing +from pathlib import Path from types import SimpleNamespace import pytest @@ -149,7 +150,12 @@ def test_existing_wrong_bytes_are_refused_without_overwrite(tmp_path, monkeypatc 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 + 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 @@ -197,7 +203,12 @@ def test_same_bytes_inode_replacement_across_sync_cannot_adopt_new_file(tmp_path ) as filesystems: with pytest.raises(DeliveryError, match="changed across synchronization"): restore_io.materialize( - rows, source, destination, restore.readback, filesystems, time.monotonic() + 120 + rows, + source, + destination, + restore.readback, + filesystems, + time.monotonic() + 120, ) @@ -206,7 +217,12 @@ def test_expired_batch_deadline_copies_no_files(tmp_path, monkeypatch): 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 + rows, + source, + destination, + restore.readback, + filesystems, + time.monotonic() - 1, ) assert not destination(rows[0])[0].exists() @@ -217,7 +233,9 @@ def test_syncfs_failure_is_propagated_in_the_ordinary_syscall_model(tmp_path, mo return -1 monkeypatch.setattr( - restore_io.ctypes, "CDLL", lambda *_a, **_kw: SimpleNamespace(syncfs=FailedSync()) + 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()] @@ -242,7 +260,12 @@ def test_cursor_is_open_before_first_write_and_held_across_batches(tmp_path, mon monkeypatch.setattr(restore_io, "pending_file", write) verified = restore_io.materialize( - rows, source, destination, restore.readback, filesystems, time.monotonic() + 120 + rows, + source, + destination, + restore.readback, + filesystems, + time.monotonic() + 120, ) assert len(verified) == len(rows) assert len(events) == 3 @@ -327,3 +350,162 @@ def test_process_copy_refuses_fork_from_a_multithreaded_owner(tmp_path, monkeypa rows, source, destination, restore.readback, fs, time.monotonic() + 120 ) assert not destination(rows[0])[0].exists() + + +def group_files(tmp_path, values, expected=None): + rows = [] + for i, raw in enumerate(values): + path = tmp_path / str(i) + path.write_bytes(raw) + path.chmod(0o600) + wanted = raw if expected is None else expected[i] + item = { + "path": str(i), + "bytes": len(wanted), + "sha256": hashlib.sha256(wanted).hexdigest(), + } + rows.append( + ( + item, + path, + os.getuid(), + os.getgid(), + 0o600, + restore.file_witness(path.stat()), + ) + ) + return rows + + +def read_group(group): + return restore.readback_group(group, time.monotonic() + 120) + + +def test_each_repeated_destination_is_fully_read_with_one_actual_reference_hash( + tmp_path, monkeypatch +): + raw = b"a" * 1048576 + b"b" * 1048576 + b"last-byte-proof" + group = group_files(tmp_path, [raw] * 4) + original_open, original_hash = os.fdopen, hashlib.sha256 + observed, hashes = {}, [] + + class Reader: + def __init__(self, stream, path): + self.stream, self.path = stream, path + + def __enter__(self): + return self + + def __exit__(self, *args): + return self.stream.__exit__(*args) + + def fileno(self): + return self.stream.fileno() + + def read(self, count): + assert count == 1048576 + block = self.stream.read(count) + observed[self.path] = observed.get(self.path, 0) + len(block) + return block + + def opened(fd, mode): + return Reader(original_open(fd, mode), Path(os.readlink(f"/proc/self/fd/{fd}"))) + + def hashed(*args, **kwargs): + hashes.append(True) + return original_hash(*args, **kwargs) + + monkeypatch.setattr(restore.os, "fdopen", opened) + monkeypatch.setattr(restore.hashlib, "sha256", hashed) + witnesses = read_group(group) + assert observed == {row[1]: len(raw) for row in group} + assert len(hashes) == 1 + assert witnesses == [row[-1] for row in group] + + +def test_last_byte_difference_in_repeated_target_is_refused(tmp_path): + good = b"x" * 1048576 + b"ending" + bad = good[:-1] + b"!" + group = group_files(tmp_path, [good, bad, good], [good] * 3) + with pytest.raises(DeliveryError, match="byte verification"): + read_group(group) + assert group[1][1].read_bytes() == bad + + +def test_identically_wrong_reference_and_duplicates_cannot_match_expected_digest( + tmp_path, +): + group = group_files(tmp_path, [b"bad"] * 3, [b"new"] * 3) + with pytest.raises(DeliveryError, match="byte verification"): + read_group(group) + + +def test_unique_large_and_empty_files_keep_exact_bytes(tmp_path): + values = [b"", b"", b"one", b"other", b"z" * (8 * 1024**2 + 1)] + group = group_files(tmp_path, values) + assert read_group(group) == [row[-1] for row in group] + + +def test_reference_is_rebound_when_digest_changes(tmp_path): + group = group_files(tmp_path, [b"aa", b"aa", b"bb", b"bb", b"aa", b"aa"]) + assert read_group(group) == [row[-1] for row in group] + + +def test_changed_file_after_complete_read_is_refused(tmp_path, monkeypatch): + group = group_files(tmp_path, [b"known", b"known"]) + original = os.fdopen + + class ChangedAtEnd: + def __init__(self, stream): + self.stream = stream + + def __enter__(self): + return self + + def __exit__(self, *args): + return self.stream.__exit__(*args) + + def fileno(self): + return self.stream.fileno() + + def read(self, count): + value = self.stream.read(count) + if not value: + group[0][1].write_bytes(b"other") + return value + + monkeypatch.setattr(restore.os, "fdopen", lambda *a, **kw: ChangedAtEnd(original(*a, **kw))) + with pytest.raises(DeliveryError, match="changed during readback"): + read_group(group) + + +def test_group_rejects_metadata_bounds_and_expired_deadline(tmp_path): + group = group_files(tmp_path, [b"known"]) + with pytest.raises(DeliveryError, match="fixed batch bounds"): + read_group(group * (restore_io.BATCH_FILES + 1)) + with pytest.raises(DeliveryError, match="deadline"): + restore.readback_group(group, time.monotonic() - 1) + + +def test_grouped_real_copy_workers_join_and_return_independent_verified_files( + tmp_path, monkeypatch +): + children = {c.pid for c in multiprocessing.active_children()} + rows, source, destination = fixture_rows(tmp_path, [b"shared"] * 8) + with closing(filesystem_model(monkeypatch, destination(rows[0])[0].parent, [])) as filesystems: + result = restore_io.materialize( + rows, + source, + destination, + restore.readback, + filesystems, + time.monotonic() + 120, + group_readback=restore.readback_group, + ) + assert {c.pid for c in multiprocessing.active_children()} == children + assert set(result) == {row["path"] for row in rows} + assert len({destination(row)[0].stat().st_ino for row in rows}) == len(rows) + for row in rows: + path, uid, gid, mode = destination(row) + assert path.read_bytes() == b"shared" + restore.unchanged_file(path, result[row["path"]], uid, gid, mode) diff --git a/scripts/release/README.md b/scripts/release/README.md index 9047b3e..1f44f2a 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -411,8 +411,13 @@ all normal services and tasks are stopped. The stopped-state checks precede data verification and copying and run again before completion. Original identities, modes and replay IDs are preserved. Its PostgreSQL target is writable; the fresh RabbitMQ volume is checked through a read-only mount and receives no copied broker state. -Each restored file is flushed and independently read back against its trusted -digest. Its inode, ownership, mode, link count, size and nanosecond timestamps +Each restored file is flushed and read back in full. Within a joined half-batch, +repeated digest/size groups can retain at most 8 MiB of reference bytes per reader. +The first target receives an actual SHA-256 check; every later target is fully +read and compared byte for byte with those verified bytes. Unique and larger +files each receive streaming SHA-256 verification. The reference is discarded +before another group or half-batch; it never certifies a file from metadata alone. +Every target's inode, ownership, mode, link count, size and nanosecond timestamps must remain unchanged until the final receipt. These observations are local to one stopped-node recovery and are never reused after an interrupted attempt. Directory entries are flushed after their children before completion is recorded. @@ -424,8 +429,10 @@ Changes are refused. These memory-only proofs are bound to the process, capped a four bundles, 100006 entries and 16 MiB of metadata, and discarded on return or failure. Further bundles receive full verification; separate commands start fresh. 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. +Unsupported transfer falls back at the same offsets. Source size/EOF, filesystem +durability and full destination byte verification remain required. Version 4 +flushes each affected filesystem through error cursors held before writing; +earlier formats retain per-file fsync. Public bundle inventories read every payload byte with at most four readers in flight. Each reader opens without following a final symlink and verifies the file identity and metadata before and after hashing. The inventory checks those diff --git a/scripts/release/restore.py b/scripts/release/restore.py index 73cbd4c..ecca58b 100644 --- a/scripts/release/restore.py +++ b/scripts/release/restore.py @@ -156,26 +156,96 @@ 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, ...]: +def _readback( + path: Path, + item: dict, + uid: int, + gid: int, + mode: int, + deadline: float, + *, + reference: bytes | None = None, + retain: bool = False, +) -> tuple[tuple[int, ...], bytes | None]: + if reference is not None and len(reference) != item["bytes"]: + raise DeliveryError("verified readback reference size differs") 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: raise DeliveryError("recovery file identity changed before readback") - total, checked = 0, hashlib.sha256() + total = 0 + checked = hashlib.sha256() if reference is None else None + retained = bytearray() if retain else None while block := stream.read(1024 * 1024): + start = total total += len(block) if total > item["bytes"] or time.monotonic() >= deadline: raise DeliveryError("recovery readback size/deadline changed") - checked.update(block) + if reference is not None and block != reference[start:total]: + raise DeliveryError("restored data byte verification failed") + if checked is not None: + checked.update(block) + if retained is not None: + retained.extend(block) after = file_witness(os.fstat(stream.fileno())) - if total != item["bytes"] or checked.hexdigest() != item["sha256"]: + if total != item["bytes"] or (checked is not None and checked.hexdigest() != item["sha256"]): raise DeliveryError("restored data byte verification failed") if before != after or after != file_witness(path.lstat()) or time.monotonic() >= deadline: raise DeliveryError("recovery file changed during readback") - return after + return after, bytes(retained) if retained is not None else None + + +def readback( + path: Path, item: dict, uid: int, gid: int, mode: int, deadline: float +) -> tuple[int, ...]: + return _readback(path, item, uid, gid, mode, deadline)[0] + + +def readback_group(group: list[tuple], deadline: float) -> list[tuple[int, ...]]: + """Read every byte; compare repeated files to an actually SHA256-checked buffer. + + The first target of a repeated digest/size is independently hashed after the + filesystem flush. A local immutable buffer holds at most 8 MiB; subsequent + targets are fully read and compared byte for byte, with the same ownership, + mode, size, descriptor/path identity and before/after 9-stat checks. Nothing + is certified from metadata or a source-copy assumption. Large or unique + files retain their independent streaming SHA256 readback. No buffer crosses + this joined half-batch, operator invocation or synchronization boundary. + """ + restore_io.remaining(deadline) + if len(group) > restore_io.BATCH_FILES or ( + len(group) > 1 and sum(row[0]["bytes"] for row in group) > restore_io.BATCH_BYTES + ): + raise DeliveryError("recovery readback group exceeds fixed batch bounds") + counts: dict[tuple[str, int], int] = {} + for item, *_ in group: + key = item["sha256"], item["bytes"] + counts[key] = counts.get(key, 0) + 1 + previous = None + reference = None + verified = [] + for item, path, uid, gid, mode, _ in group: + restore_io.remaining(deadline) + key = item["sha256"], item["bytes"] + if key != previous: + reference = None + previous = key + repeated_small = counts[key] > 1 and item["bytes"] <= 8 * 1024**2 + witness, retained = _readback( + path, + item, + uid, + gid, + mode, + deadline, + reference=reference if repeated_small else None, + retain=repeated_small and reference is None, + ) + if retained is not None: + reference = retained + verified.append(witness) + return verified def unchanged_file(path: Path, witness: tuple[int, ...], uid: int, gid: int, mode: int) -> None: @@ -331,6 +401,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: readback, filesystems, deadline, + group_readback=readback_group, ) else: witnesses = file_copy.bounded_copies(value["files"], copy, deadline=deadline) diff --git a/scripts/release/restore_io.py b/scripts/release/restore_io.py index 33d7daa..3de3cff 100644 --- a/scripts/release/restore_io.py +++ b/scripts/release/restore_io.py @@ -171,6 +171,8 @@ def materialize( readback: Callable[[Path, dict, int, int, int, float], tuple[int, ...]], filesystems: Filesystems, deadline: float, + *, + group_readback: Callable[[list[tuple], float], list[tuple[int, ...]]] | None = None, ) -> dict[str, tuple[int, ...]]: """Copy, join, sync and independently read each complete destination. @@ -180,8 +182,10 @@ def materialize( 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. + Every returned witness follows an actual filesystem flush and complete + destination byte verification. The optional grouped reader can reuse an + actually hashed immutable reference only after full byte equality readback + of each remaining target. Exceptions join outstanding work and retain partial files. """ remaining(deadline) if os.name != "posix" or threading.active_count() != 1: @@ -218,7 +222,12 @@ def materialize( 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] + calls = [ + executor.submit(group_readback, group, deadline) + if group_readback is not None + else 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: