diff --git a/docs/artifact-admission.md b/docs/artifact-admission.md index cf396433..4ee9c0e5 100644 --- a/docs/artifact-admission.md +++ b/docs/artifact-admission.md @@ -59,3 +59,40 @@ idempotent recovery, revision fencing, policy binding, conservative charges and private/exclusive file setup. Process-crash tests do not establish power-loss or multi-machine durability. Full repository/installed-package validation and independent review are required before publication or integration with hosting. + +## Read-only restart reconciliation + +`artifact_inventory.reconcile_inventory(ledger, selections, max_read_bytes=...)` +rechecks every admitted request against a caller-selected absolute file path and +manifest. `selections` is a tuple of `InventorySelection(request_id, manifest, +path)`, with exactly one entry per admitted request, at most 192. The trusted +ledger supplies each pinned manifest root, project, fragment index, length and +content/transfer binding. All selections and the aggregate logical read budget +(up to 64 MiB) are checked before any fragment is opened. The full charged amount +must fit that budget even when some files are absent. Actual reads use the +existing bounded 32 KiB verifier; the budget is not a total CPU/RSS or physical +I/O claim. Work is synchronous and sequential; admission SQLite waits retain the +existing five-second busy bound and are not asynchronously cancellable. + +A report identifies each request as `verified`, `missing` or `unverified`. +Missing includes disappearance of a selected file or parent during inspection; +unverified includes unsafe file type/path, observable mutation, bad bytes and +filesystem errors. It is not an automatic repair or deletion decision. Cancellation +callbacks may raise any exception: they abort the pass rather than being mistaken +for a missing/corrupt file. No partial report is returned. A changed ledger +snapshot also rejects the complete report, including changes committed by another +process during the pass. Corrupt or unreadable ledger state fails closed. + +Prior `verified_observation` metadata is never accepted as evidence of current +bytes. Reconciliation changes no records, revisions, charges or files, including +stale observations. Callers must use the fresh report for their recovery decision; +this API does not persist availability or grant a serving capability. Observations +are made at different times and are not an atomic filesystem snapshot or a +future retention promise. Reverify before serving or recovery use. Publisher +approval remains separate. Trusted stable physical directories remain required; +hostile same-user writers are outside the guarantee. + +Unknown/orphan files are not enumerated or charged by this API. Establishing an +owned retained store, durable path ownership, writer fencing, crash-safe ingestion, +orphan reconciliation, safe reclaim and bounded repair remains future work. Never +release charges or execute downloaded bytes because this audit found a file missing. diff --git a/scripts/peer_install_smoke.py b/scripts/peer_install_smoke.py index dd28c1ea..0bcaa1e8 100644 --- a/scripts/peer_install_smoke.py +++ b/scripts/peer_install_smoke.py @@ -698,6 +698,10 @@ def admission_smoke(cwd: Path) -> None: from aethermesh_core.artifact_admission import AdmissionLimits, ArtifactAdmission from aethermesh_core.artifact_codec import ArtifactLimits, package_artifact from aethermesh_core.artifact_files import verify_fragment_file + from aethermesh_core.artifact_inventory import ( + InventorySelection, + reconcile_inventory, + ) package = package_artifact( b"synthetic catalog bytes", @@ -734,6 +738,15 @@ def admission_smoke(cwd: Path) -> None: invalidated = reopened.invalidate("8" * 64, expected_revision=saved.revision) assert invalidated.state == "reserved" assert reopened.snapshot().charged_bytes == len(package.fragments[0]) + selection = (InventorySelection("8" * 64, package.manifest, fragment),) + report = reconcile_inventory(reopened, selection, max_read_bytes=1024) + assert report.observations[0].status == "verified" + assert not report.retention_confirmed and not report.publisher_authorized + fragment.unlink() + missing = reconcile_inventory(reopened, selection, max_read_bytes=1024) + assert missing.observations[0].status == "missing" + assert missing.charged_bytes == report.charged_bytes + assert reopened.snapshot().records == (invalidated,) finally: reopened.close() print( diff --git a/src/aethermesh_core/artifact_inventory.py b/src/aethermesh_core/artifact_inventory.py new file mode 100644 index 00000000..4ad22caa --- /dev/null +++ b/src/aethermesh_core/artifact_inventory.py @@ -0,0 +1,133 @@ +"""Bounded read-only reconciliation of admitted, explicitly selected local files. + +Observations are sequential and ephemeral, not retention leases, publisher +approval, a filesystem scan, recovery writes or permission to reclaim charges. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Literal + +from .artifact_admission import AdmissionConflict, ArtifactAdmission, FragmentAdmission +from .artifact_codec import ArtifactManifest, CheckCancel, _cancel, _integer, _trusted +from .artifact_files import verify_fragment_file + + +@dataclass(frozen=True) +class InventorySelection: + request_id: str + manifest: ArtifactManifest + path: Path + + +@dataclass(frozen=True) +class InventoryObservation: + admission: FragmentAdmission + status: Literal["verified", "missing", "unverified"] + + +@dataclass(frozen=True) +class InventoryReport: + ledger_revision: int + charged_bytes: int + read_budget_bytes: int + observations: tuple[InventoryObservation, ...] + publisher_authorized: Literal[False] = False + retention_confirmed: Literal[False] = False + + +def _preflight( + records: tuple[FragmentAdmission, ...], selections: tuple[InventorySelection, ...] +) -> dict[str, InventorySelection]: + if not isinstance(selections, tuple) or len(selections) > 192: + raise ValueError("inventory selections must be a bounded tuple") + chosen = {item.request_id: item for item in selections} + if len(chosen) != len(selections) or set(chosen) != { + record.request_id for record in records + }: + raise ValueError("select every admitted request exactly once") + for record in records: + item = chosen[record.request_id] + _trusted(item.manifest, record.manifest_digest) + if ( + item.manifest.project != record.project + or item.manifest.fragment_size(record.fragment_index) != record.size + or item.manifest.fragments[record.fragment_index] + != (record.sha256, record.transfer_digest) + ): + raise ValueError("inventory manifest and admission binding mismatch") + if not isinstance(item.path, Path) or not item.path.is_absolute(): + raise ValueError("inventory requires explicitly selected absolute paths") + return chosen + + +def _observe( + record: FragmentAdmission, + item: InventorySelection, + check_cancel: CheckCancel | None, +) -> InventoryObservation: + cancelled = False + + def check() -> None: + nonlocal cancelled + try: + _cancel(check_cancel) + except BaseException: + cancelled = True + raise + + try: + verify_fragment_file( + item.manifest, + record.fragment_index, + item.path, + expected_digest=record.manifest_digest, + max_read_bytes=record.size, + check_cancel=check, + ) + except (OSError, ValueError) as error: + if cancelled: + raise + status: Literal["missing", "unverified"] = ( + "missing" if isinstance(error, FileNotFoundError) else "unverified" + ) + return InventoryObservation(record, status) + return InventoryObservation(record, "verified") + + +def reconcile_inventory( + ledger: ArtifactAdmission, + selections: tuple[InventorySelection, ...], + *, + max_read_bytes: int, + check_cancel: CheckCancel | None = None, +) -> InventoryReport: + """Recheck every admitted request without changing files or ledger metadata. + + Require a complete caller-selected mapping and aggregate logical read budget + before any fragment open. Existing verified_observation state is never used + as evidence of current bytes. Missing/unsafe/corrupt/unreadable files remain + charged. Callback exceptions always abort; no partial report is returned. + + Fail if the ledger changes during the pass. This fences report metadata, not + filesystem writers: file observations occur at different times and must be + refreshed before serving. Unselected/orphan files are outside this report. + """ + _integer(max_read_bytes, 0, 64 * 1024 * 1024) + _cancel(check_cancel) + snapshot = ledger.snapshot() + chosen = _preflight(snapshot.records, selections) + if snapshot.charged_bytes > max_read_bytes: + raise ValueError("aggregate inventory read budget exceeded") + observations = tuple( + _observe(record, chosen[record.request_id], check_cancel) + for record in snapshot.records + ) + _cancel(check_cancel) + if ledger.snapshot() != snapshot: + raise AdmissionConflict("inventory changed during reconciliation") + return InventoryReport( + snapshot.revision, snapshot.charged_bytes, snapshot.charged_bytes, observations + ) diff --git a/tests/test_artifact_inventory.py b/tests/test_artifact_inventory.py new file mode 100644 index 00000000..a976193d --- /dev/null +++ b/tests/test_artifact_inventory.py @@ -0,0 +1,288 @@ +"""Restart reconciliation, aggregate bounds and conservative failure behavior.""" + +import dataclasses +import errno +import os +import subprocess +import sys +import tempfile +import unittest +from contextlib import closing +from pathlib import Path +from unittest.mock import patch + +from aethermesh_core.artifact_admission import ( + AdmissionConflict, + AdmissionLimits, + ArtifactAdmission, +) +from aethermesh_core.artifact_files import verify_fragment_file +from aethermesh_core.artifact_inventory import InventorySelection, reconcile_inventory +from tests.test_artifact_codec import package + + +class InventoryTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name).resolve() + self.limits = AdmissionLimits(max_bytes=1024, max_objects=6) + self.directory = self.root / "ledger" + self.ledger = ArtifactAdmission.create( + self.directory, store_id="a" * 64, limits=self.limits + ) + self.addCleanup(self.ledger.close) + self.pkg = package() + self.items = [] + for index in range(3): + request = f"{index:064x}" + self.ledger.reserve( + request, + self.pkg.manifest, + index, + expected_digest=self.pkg.manifest.digest, + ) + path = self.root / f"fragment-{index}" + path.write_bytes(self.pkg.fragments[index]) + self.items.append(InventorySelection(request, self.pkg.manifest, path)) + self.before = self.ledger.snapshot() + + def audit(self, **kwargs): + options = {"max_read_bytes": self.before.charged_bytes} + return reconcile_inventory(self.ledger, tuple(self.items), **(options | kwargs)) + + def test_reopen_rechecks_files_and_keeps_all_charges_and_metadata(self): + for record, item in zip(self.before.records, self.items, strict=True): + observation = verify_fragment_file( + item.manifest, + record.fragment_index, + item.path, + expected_digest=record.manifest_digest, + max_read_bytes=record.size, + ) + self.ledger.record_verification( + record.request_id, + expected_revision=record.revision, + observation=observation, + ) + saved = self.ledger.snapshot() + self.items[1].path.unlink() + self.items[2].path.write_bytes(b"bad!") + with closing( + ArtifactAdmission(self.directory, store_id="a" * 64, limits=self.limits) + ) as reopened: + report = reconcile_inventory( + reopened, + tuple(reversed(self.items)), + max_read_bytes=saved.charged_bytes, + ) + self.assertEqual( + [entry.status for entry in report.observations], + ["verified", "missing", "unverified"], + ) + self.assertEqual(reopened.snapshot(), saved) + self.assertEqual(report.ledger_revision, saved.revision) + self.assertEqual(report.charged_bytes, saved.charged_bytes) + self.assertEqual(report.read_budget_bytes, saved.charged_bytes) + self.assertFalse(report.retention_confirmed) + self.assertFalse(report.publisher_authorized) + self.assertFalse(self.items[1].path.exists()) + self.assertEqual(self.items[2].path.read_bytes(), b"bad!") + + def test_complete_selection_and_all_bindings_preflight_before_file_io(self): + other = package(b"other bytes") + choices = ( + (), + tuple(self.items[:-1]), + tuple(self.items + [self.items[0]]), + tuple( + self.items + [dataclasses.replace(self.items[0], request_id="f" * 64)] + ), + tuple( + self.items[:2] + + [dataclasses.replace(self.items[2], manifest=other.manifest)] + ), + tuple( + self.items[:2] + + [dataclasses.replace(self.items[2], path=Path("relative"))] + ), + tuple([self.items[0]] * 193), + list(self.items), + ) + for selections in choices: + with ( + self.subTest(selections=selections), + patch( + "aethermesh_core.artifact_inventory.verify_fragment_file" + ) as verify, + ): + with self.assertRaises(ValueError): + reconcile_inventory(self.ledger, selections, max_read_bytes=1024) + verify.assert_not_called() + for budget in ( + 0, + self.before.charged_bytes - 1, + True, + -1, + 64 * 1024 * 1024 + 1, + ): + with ( + self.subTest(budget=budget), + patch( + "aethermesh_core.artifact_inventory.verify_fragment_file" + ) as verify, + ): + with self.assertRaises(ValueError): + self.audit(max_read_bytes=budget) + verify.assert_not_called() + + def test_empty_inventory_requires_no_reads(self): + with ( + closing( + ArtifactAdmission.create( + self.root / "empty", store_id="c" * 64, limits=self.limits + ) + ) as ledger, + patch("aethermesh_core.artifact_inventory.verify_fragment_file") as verify, + ): + report = reconcile_inventory(ledger, (), max_read_bytes=0) + self.assertEqual(report.observations, ()) + self.assertEqual(report.read_budget_bytes, 0) + verify.assert_not_called() + + def test_unsafe_and_unreadable_files_are_not_available_or_deleted(self): + first = self.items[0].path + first.unlink() + first.symlink_to(self.items[1].path) + last = self.items[2].path + last.unlink() + os.mkfifo(last) + report = self.audit() + self.assertEqual( + [entry.status for entry in report.observations], + ["unverified", "verified", "unverified"], + ) + self.assertTrue(first.is_symlink()) + self.assertEqual(self.ledger.snapshot(), self.before) + with patch( + "aethermesh_core.artifact_files.os.read", + side_effect=OSError(errno.EIO, "read"), + ): + report = self.audit() + self.assertTrue( + all(entry.status == "unverified" for entry in report.observations) + ) + self.assertEqual(self.ledger.snapshot(), self.before) + + def test_cancellation_exceptions_are_never_reported_as_file_failures(self): + baseline_calls = [] + self.audit(check_cancel=lambda: baseline_calls.append(True)) + for error_type in ( + InterruptedError, + FileNotFoundError, + ValueError, + KeyboardInterrupt, + ): + for stop in (1, 3, len(baseline_calls)): + calls = [] + + def cancel( + selected_stop=stop, selected_error=error_type, selected_calls=calls + ): + selected_calls.append(True) + if len(selected_calls) == selected_stop: + raise selected_error("cancelled") + + with self.subTest(error=error_type, stop=stop): + with self.assertRaises(error_type): + self.audit(check_cancel=cancel) + self.assertEqual(len(calls), stop) + self.assertEqual(self.ledger.snapshot(), self.before) + + def test_ledger_change_from_another_connection_rejects_entire_report(self): + changed = [] + with closing( + ArtifactAdmission(self.directory, store_id="a" * 64, limits=self.limits) + ) as other: + + def change(): + if not changed: + record = self.before.records[0] + other.invalidate( + record.request_id, expected_revision=record.revision + ) + changed.append(True) + + with patch( + "aethermesh_core.artifact_inventory.verify_fragment_file", + wraps=verify_fragment_file, + ) as verify: + calls = [] + + def callback(): + calls.append(True) + if len(calls) == 3: + change() + + with self.assertRaises(AdmissionConflict): + self.audit(check_cancel=callback) + self.assertEqual(verify.call_count, 3) + self.assertEqual( + self.ledger.snapshot().charged_bytes, self.before.charged_bytes + ) + + def test_process_update_during_file_read_rejects_report(self): + original = os.read + writes = [] + child = """ +import sys +from pathlib import Path +from aethermesh_core.artifact_admission import ArtifactAdmission, AdmissionLimits +ledger = ArtifactAdmission(Path(sys.argv[1]), store_id="a"*64, + limits=AdmissionLimits(max_bytes=1024, max_objects=6)) +try: + ledger.invalidate("0"*64, expected_revision=1) +finally: + ledger.close() +""" + + def read(fd, size): + if not writes: + writes.append(True) + result = subprocess.run( + [sys.executable, "-c", child, str(self.directory)], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + return original(fd, size) + + with ( + patch("aethermesh_core.artifact_files.os.read", side_effect=read), + self.assertRaises(AdmissionConflict), + ): + self.audit() + self.assertEqual(writes, [True]) + self.assertEqual( + self.ledger.snapshot().charged_bytes, self.before.charged_bytes + ) + + def test_real_reads_stay_within_aggregate_budget(self): + reads = [] + original = os.read + + def read(fd, size): + value = original(fd, size) + reads.append(len(value)) + return value + + with patch("aethermesh_core.artifact_files.os.read", side_effect=read): + report = self.audit() + self.assertEqual(sum(reads), report.read_budget_bytes) + self.assertLessEqual(max(reads), 32768) + self.assertTrue( + all(entry.status == "verified" for entry in report.observations) + ) + self.assertEqual(self.ledger.snapshot(), self.before)