diff --git a/backend/tests/test_restore_native_transport.py b/backend/tests/test_restore_native_transport.py index f77e0ca..fa3da19 100644 --- a/backend/tests/test_restore_native_transport.py +++ b/backend/tests/test_restore_native_transport.py @@ -1,12 +1,15 @@ """Ordinary pipe fixtures only; no Root, filesystem-copy or VM acceptance.""" import json +import os +import stat import subprocess import sys import time +import weakref import pytest -from scripts.release import restore_native +from scripts.release import restore_io, restore_native from scripts.release.policy import DeliveryError @@ -113,3 +116,118 @@ def test_transport_deadline_reaps_its_owned_waiting_child(transport): with pytest.raises(DeliveryError): restore_native.Session(payload, "bk-" + "1" * 32, "rr-" + "2" * 32, 1, 0, then + 0.2) assert time.monotonic() - then < 5 + + +@pytest.mark.parametrize("owned", [False, True]) +def test_owned_planner_spool_preserves_transport_counts_and_join(transport, owned): + payload, start, children = transport + start(""" +import json,sys +count=total=batch=0 +for line in sys.stdin: + value=json.loads(line) + kind=value['kind'] + stage='opened' + if kind=='restore.io.batch': + count+=len(value['files']) + total+=sum(r['bytes'] for r in value['files']) + batch=value['batch_id'] + stage='verified' + if kind=='restore.io.finish':stage='finished' + reply={'schema_version':1,'kind':'restore.io.reply','stage':stage,'batch_id':batch,'verified_files':count,'verified_logical_bytes':total,'execution_verified':False,'error':'none'} + print(json.dumps(reply),flush=True) + if stage=='finished':break +""") + rows = [{"path": f"artifacts/{n:04}", "bytes": n, "sha256": "1" * 64} for n in range(513)] + session = restore_native.Session( + payload, + "bk-" + "1" * 32, + "rr-" + "2" * 32, + len(rows), + sum(r["bytes"] for r in rows), + time.monotonic() + 10, + ) + try: + if owned: + session.materialize_owned(rows, payload) + assert rows == [] + else: + session.materialize(rows) + assert len(rows) == 513 + session.finish() + assert session.finished and children[0].returncode == 0 + assert session.files == 513 and session.batch_id == 3 + finally: + session.close() + + +def test_owned_planner_spool_releases_rows_and_preserves_exact_sorted_batches(tmp_path): + class Row(dict): + pass + + rows = [Row(path=f"artifacts/{n:04}", bytes=n, sha256=f"{n % 7:064x}") for n in range(600)] + expected = [ + json.dumps(b, separators=(",", ":"), ensure_ascii=True).encode() + b"\n" + for b in restore_io.batches(rows) + ] + references = [weakref.ref(row) for row in rows] + with restore_native.BatchSpool(rows, tmp_path, time.monotonic() + 10) as spool: + assert not rows and all(ref() is None for ref in references) + assert spool.stream is not None + info = os.fstat(spool.stream.fileno()) + assert stat.S_ISREG(info.st_mode) and stat.S_IMODE(info.st_mode) == 0o600 + assert info.st_nlink == 0 and not os.get_inheritable(spool.stream.fileno()) + assert list(tmp_path.iterdir()) == [] + actual = [ + json.dumps(b, separators=(",", ":"), ensure_ascii=True).encode() + b"\n" for b in spool + ] + assert actual == expected and spool.complete + with pytest.raises(DeliveryError, match="already consumed"): + list(spool) + assert spool.stream.closed and list(tmp_path.iterdir()) == [] + + +def test_owned_planner_spool_partial_consumer_fails_and_closes(tmp_path): + spool = restore_native.BatchSpool( + [{"path": "artifacts/a", "bytes": 1, "sha256": "1" * 64}], + tmp_path, + time.monotonic() + 10, + ) + with pytest.raises(DeliveryError, match="input incomplete"), spool: + next(iter(spool)) + assert spool.stream.closed and list(tmp_path.iterdir()) == [] + + +def test_owned_planner_spool_preserves_consumer_error_and_closes(tmp_path): + spool = restore_native.BatchSpool([], tmp_path, time.monotonic() + 10) + with pytest.raises(RuntimeError, match="fixture failure"), spool: + raise RuntimeError("fixture failure") + assert spool.stream.closed and list(tmp_path.iterdir()) == [] + + +def test_owned_planner_spool_bounds_creation_and_releases_descriptor(tmp_path, monkeypatch): + original = restore_native.tempfile.TemporaryFile + opened = [] + + def temporary_file(**options): + stream = original(**options) + opened.append(stream) + return stream + + monkeypatch.setattr(restore_native.tempfile, "TemporaryFile", temporary_file) + monkeypatch.setattr(restore_native, "MAX_SPOOL_BYTES", 1) + with pytest.raises(DeliveryError, match="input exceeds bound"): + restore_native.BatchSpool( + [{"path": "artifacts/a", "bytes": 1, "sha256": "1" * 64}], + tmp_path, + time.monotonic() + 10, + ) + assert len(opened) == 1 and opened[0].closed and list(tmp_path.iterdir()) == [] + + +def test_owned_planner_spool_retains_original_deadline(tmp_path): + spool = restore_native.BatchSpool([], tmp_path, time.monotonic() + 10) + spool.deadline = time.monotonic() - 1 + with pytest.raises(DeliveryError, match="deadline exceeded"), spool: + list(spool) + assert spool.stream.closed and list(tmp_path.iterdir()) == [] diff --git a/scripts/release/README.md b/scripts/release/README.md index a1b0e83..b30c568 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -430,6 +430,16 @@ limits, the read-only backup volume and stopped writers. It starts no programs or services. Python retains its filesystem error cursors from before directory creation; the native child holds its own cursors before file writes. Both cursor sets are synchronized before completion. The original deadlines remain. +After the complete private manifest and its objects have been verified, Python +writes the same sorted batches to a bounded, unlinked 0600 temporary descriptor +in the private recovery directory. It checks the complete spool digest and +stable descriptor identity before sending any batch, then releases the decoded +file inventory and selected bundle manifest. Each streamed batch remains bounded; +its descriptor identity, complete stream digest, count and logical byte total +are checked through EOF. The descriptor is closed on success or failure. This +reduces retained planner objects during copying without changing the native +copy, synchronization, full target readback or final witness checks. The spool +adds no recovery memory, CPU, process or time allowance. Each batch copies, joins both workers, flushes the affected filesystems, then reads every target. Final directory sync and stopped/selected-identity checks precede the native inventory's complete metadata recheck and thread joins. diff --git a/scripts/release/restore.py b/scripts/release/restore.py index fc78efc..180f2c3 100644 --- a/scripts/release/restore.py +++ b/scripts/release/restore.py @@ -348,6 +348,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: deadline = time.monotonic() + 1700 filesystems = None native = None + file_count = len(value["files"]) try: fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) stopped() @@ -405,11 +406,15 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: payload, backup_id, recovery_id, - len(value["files"]), + file_count, backup_manifest.logical_bytes(value), deadline, ) - native.materialize(value["files"]) + # Compatibility and every original private object were verified + # above. Release decoded file rows before copying the full logical + # inventory; the private input spool is completely verified too. + del data + native.materialize_owned(value.pop("files"), recovery) else: witnesses = file_copy.bounded_copies(value["files"], copy, deadline=deadline) verified = { @@ -459,7 +464,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: record.update( data_restored=True, restored_payload_bytes=backup_manifest.logical_bytes(value), - restored_files=len(value["files"]), + restored_files=file_count, restored_directories=len(value["directories"]), historical_configuration_active=False, factory_replay_history_retained=True, diff --git a/scripts/release/restore_native.py b/scripts/release/restore_native.py index b5a7da6..116344e 100644 --- a/scripts/release/restore_native.py +++ b/scripts/release/restore_native.py @@ -1,22 +1,141 @@ """One bounded native data-copy child in the existing exclusive restore unit.""" +import hashlib import json import os import selectors import signal import stat import subprocess +import tempfile import time +from collections.abc import Iterator from pathlib import Path from kf_oj.contracts.restore_io import RestoreIOReply -from . import file_copy, preflight, restore_io +from . import backup_manifest, file_copy, preflight, restore_io from .policy import DeliveryError, invalid_constant, unique_object MAX_REQUEST_BYTES = 256 * 1024 MAX_REPLY_BYTES = 64 * 1024 MAX_STDERR_BYTES = 16 * 1024 +# Only additional list/newline framing beyond the already bounded source metadata. +MAX_SPOOL_BYTES = backup_manifest.MAX_METADATA_BYTES + 4 * backup_manifest.MAX_LOGICAL_ENTRIES + + +class BatchSpool: + """Consume validated rows into private, completely verified descriptor-only input.""" + + def __init__(self, rows: list[dict], directory: Path, deadline: float): + self.deadline = deadline + self.files = len(rows) + self.logical_bytes = sum(row["bytes"] for row in rows) + self.started = self.complete = False + self.size = 0 + self.stream = None + restore_io.remaining(deadline) + if self.files > backup_manifest.MAX_LOGICAL_ENTRIES: + raise DeliveryError("recovery planner inventory exceeds bound") + try: + self.stream = tempfile.TemporaryFile(mode="w+b", dir=directory) # noqa: SIM115 + # Ownership spans construction, consumption and every failure path. + info = os.fstat(self.stream.fileno()) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_nlink != 0 + or (info.st_uid, info.st_gid) != (os.geteuid(), os.getegid()) + or stat.S_IMODE(info.st_mode) != 0o600 + or os.get_inheritable(self.stream.fileno()) + ): + raise DeliveryError("private recovery planner descriptor required") + expected = hashlib.sha256() + batch = None + for batch in restore_io.batches(rows): + restore_io.remaining(deadline) + raw = json.dumps(batch, separators=(",", ":"), ensure_ascii=True).encode() + b"\n" + self.size += len(raw) + if len(raw) > MAX_REQUEST_BYTES or self.size > MAX_SPOOL_BYTES: + raise DeliveryError("recovery planner input exceeds bound") + if self.stream.write(raw) != len(raw): + raise DeliveryError("recovery planner write incomplete") + expected.update(raw) + # No row dictionary or duplicate decoded string remains here while + # the native child materializes the full logical data inventory. + batch = None + rows.clear() + self.stream.flush() + os.fsync(self.stream.fileno()) + self.witness = file_copy._identity(os.fstat(self.stream.fileno())) + self.digest = expected.hexdigest() + self.stream.seek(0) + observed = hashlib.file_digest(self.stream, "sha256").hexdigest() + self.unchanged() + if observed != self.digest: + raise DeliveryError("recovery planner complete input verification failed") + self.stream.seek(0) + restore_io.remaining(deadline) + except BaseException: + self.close() + raise + + def unchanged(self) -> None: + if ( + self.stream is None + or self.stream.closed + or file_copy._identity(os.fstat(self.stream.fileno())) != self.witness + ): + raise DeliveryError("recovery planner descriptor changed") + + def __iter__(self) -> Iterator[list[dict]]: + if self.started: + raise DeliveryError("recovery planner input already consumed") + self.started = True + files = logical_bytes = 0 + observed = hashlib.sha256() + assert self.stream is not None + while True: + restore_io.remaining(self.deadline) + self.unchanged() + raw = self.stream.readline(MAX_REQUEST_BYTES + 1) + if not raw: + break + if len(raw) > MAX_REQUEST_BYTES or not raw.endswith(b"\n"): + raise DeliveryError("recovery planner input framing differs") + observed.update(raw) + batch = json.loads( + raw, object_pairs_hook=unique_object, parse_constant=invalid_constant + ) + if not isinstance(batch, list) or not 1 <= len(batch) <= restore_io.BATCH_FILES: + raise DeliveryError("recovery planner batch shape differs") + files += len(batch) + logical_bytes += sum(row["bytes"] for row in batch) + if files > self.files or logical_bytes > self.logical_bytes: + raise DeliveryError("recovery planner inventory differs") + yield batch + self.unchanged() + if (files, logical_bytes, observed.hexdigest()) != ( + self.files, + self.logical_bytes, + self.digest, + ): + raise DeliveryError("recovery planner complete input differs") + restore_io.remaining(self.deadline) + self.complete = True + + def close(self) -> None: + if self.stream is not None: + self.stream.close() + + def __enter__(self): + return self + + def __exit__(self, kind, _value, _traceback): + try: + if kind is None and not self.complete: + raise DeliveryError("recovery planner input incomplete") + finally: + self.close() class Session: @@ -67,7 +186,11 @@ class Session: if file_copy._identity(helper.lstat()) != witness: raise DeliveryError("selected native recovery helper changed") assert self.process.stdin and self.process.stdout and self.process.stderr - for handle in (self.process.stdin, self.process.stdout, self.process.stderr): + for handle in ( + self.process.stdin, + self.process.stdout, + self.process.stderr, + ): os.set_blocking(handle.fileno(), False) self.selector.register(self.process.stdout, selectors.EVENT_READ, "stdout") self.selector.register(self.process.stderr, selectors.EVENT_READ, "stderr") @@ -145,7 +268,11 @@ class Session: if ( reply.stage != stage or reply.error != "none" - or (reply.batch_id, reply.verified_files, reply.verified_logical_bytes) + or ( + reply.batch_id, + reply.verified_files, + reply.verified_logical_bytes, + ) != (self.batch_id, self.files, self.bytes) ): raise DeliveryError("native recovery file verification refused") @@ -158,7 +285,14 @@ class Session: self.selector.unregister(self.process.stdin) def materialize(self, rows: list[dict]) -> None: - for batch in restore_io.batches(rows): + self._materialize(restore_io.batches(rows)) + + def materialize_owned(self, rows: list[dict], directory: Path) -> None: + with BatchSpool(rows, directory, self.deadline) as batches: + self._materialize(iter(batches)) + + def _materialize(self, batches: Iterator[list[dict]]) -> None: + for batch in batches: self.batch_id += 1 self.files += len(batch) self.bytes += sum(item["bytes"] for item in batch)