diff --git a/docs/operations/CORPUS_HANDOFF.md b/docs/operations/CORPUS_HANDOFF.md index 63e9176..dc1fb10 100644 --- a/docs/operations/CORPUS_HANDOFF.md +++ b/docs/operations/CORPUS_HANDOFF.md @@ -68,10 +68,12 @@ python3 infra/deploy/compare_corpus_inventories.py \ The checker streams every relative path, size, and SHA-256 from both SQLite inventories, and requires the same verifier/schema version and different -normalized roots. A zero exit status confirms a non-empty staging tree matches -the source. Keep both trees quiescent during their scans and comparison. This -is an equality check only; the owner still reviews the code `6` disposition -and approves the transfer destination. +normalized roots. It opens both existing private `.inventory.lock` files +read-only and holds shared nonblocking locks through comparison; a missing, +unsafe, or busy lock fails closed. A zero exit status confirms a non-empty +staging tree matches the source. Keep both trees quiescent during their scans +and comparison. This is an equality check only; the owner still reviews the +code `6` disposition and approves the transfer destination. ## Current data disposition diff --git a/docs/operations/CURRENT_STATE.md b/docs/operations/CURRENT_STATE.md index 373b961..b77ddcf 100644 --- a/docs/operations/CURRENT_STATE.md +++ b/docs/operations/CURRENT_STATE.md @@ -199,7 +199,8 @@ protected file metadata remain unverified. Adds a read-only, resumable full-file corpus inventory and normalized-data cross-check, a source/staging comparator requiring nonzero documents and -fragments, synthetic regression tests, and an operator handoff runbook. CI's +fragments plus held shared artifact locks, synthetic regression tests, and an +operator handoff runbook. CI's existing `infra/deploy` unittest discovery includes the new tests. No workflow, runner, VPS, production data, or Docker settings are changed. The first real corpus scan stopped on an integer-storage bug; full validation with the diff --git a/infra/deploy/compare_corpus_inventories.py b/infra/deploy/compare_corpus_inventories.py index 22d6c4b..93028b6 100644 --- a/infra/deploy/compare_corpus_inventories.py +++ b/infra/deploy/compare_corpus_inventories.py @@ -4,11 +4,14 @@ from __future__ import annotations import argparse +import fcntl import itertools import json +import os import sqlite3 import stat import sys +from contextlib import ExitStack, contextmanager from pathlib import Path from typing import Any @@ -17,6 +20,32 @@ class ComparisonError(ValueError): pass +@contextmanager +def shared_inventory_lock(result_path: Path, label: str): + lock_path = result_path.absolute().parent / ".inventory.lock" + if not hasattr(os, "O_NOFOLLOW"): + raise ComparisonError(f"{label}: safe inventory lock opens are unsupported") + try: + descriptor = os.open(lock_path, os.O_RDONLY | os.O_NOFOLLOW) + except OSError as error: + raise ComparisonError(f"{label}: cannot open inventory lock: {error}") from error + try: + metadata = os.fstat(descriptor) + if ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_nlink != 1 + or stat.S_IMODE(metadata.st_mode) & 0o077 + ): + raise ComparisonError(f"{label}: inventory lock must be a private regular file") + try: + fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB) + except BlockingIOError as error: + raise ComparisonError(f"{label}: inventory is active (lock busy)") from error + yield + finally: + os.close(descriptor) + + def read_audit(result_path: Path, label: str) -> tuple[sqlite3.Connection, dict[str, Any]]: result_path = result_path.absolute() try: @@ -87,36 +116,39 @@ def read_audit(result_path: Path, label: str) -> tuple[sqlite3.Connection, dict[ def compare_inventories(source_result: Path, staging_result: Path) -> dict[str, int]: - source_db, source = read_audit(source_result, "source") - try: - staging_db, staging = read_audit(staging_result, "staging") - except Exception: - source_db.close() - raise - try: - if source["root"] == staging["root"]: - raise ComparisonError("source and staging must be different normalized roots") - if source["verifier_sha256"] != staging["verifier_sha256"] or source["schema_version"] != staging["schema_version"]: - raise ComparisonError("source and staging audits must use the same verifier and schema") - if (source["documents"], source["fragments"]) != (staging["documents"], staging["fragments"]): - raise ComparisonError("source and staging document or fragment counts differ") + with ExitStack() as locks: + locks.enter_context(shared_inventory_lock(source_result, "source")) + locks.enter_context(shared_inventory_lock(staging_result, "staging")) + source_db, source = read_audit(source_result, "source") + try: + staging_db, staging = read_audit(staging_result, "staging") + except Exception: + source_db.close() + raise + try: + if source["root"] == staging["root"]: + raise ComparisonError("source and staging must be different normalized roots") + if source["verifier_sha256"] != staging["verifier_sha256"] or source["schema_version"] != staging["schema_version"]: + raise ComparisonError("source and staging audits must use the same verifier and schema") + if (source["documents"], source["fragments"]) != (staging["documents"], staging["fragments"]): + raise ComparisonError("source and staging document or fragment counts differ") - query = ( - "SELECT relative_path,size,sha256 FROM files " - "WHERE root_id='normalized' ORDER BY relative_path" - ) - left = iter(source_db.execute(query)) - right = iter(staging_db.execute(query)) - mismatch_count = 0 - for source_row, staging_row in itertools.zip_longest(left, right): - if source_row != staging_row: - mismatch_count += 1 - if mismatch_count: - raise ComparisonError(f"source and staging differ in {mismatch_count} inventory rows") - return {"documents": source["documents"], "fragments": source["fragments"], "files": source["files"], "bytes": source["bytes"]} - finally: - source_db.close() - staging_db.close() + query = ( + "SELECT relative_path,size,sha256 FROM files " + "WHERE root_id='normalized' ORDER BY relative_path" + ) + left = iter(source_db.execute(query)) + right = iter(staging_db.execute(query)) + mismatch_count = 0 + for source_row, staging_row in itertools.zip_longest(left, right): + if source_row != staging_row: + mismatch_count += 1 + if mismatch_count: + raise ComparisonError(f"source and staging differ in {mismatch_count} inventory rows") + return {"documents": source["documents"], "fragments": source["fragments"], "files": source["files"], "bytes": source["bytes"]} + finally: + source_db.close() + staging_db.close() def main() -> int: diff --git a/infra/deploy/test_compare_corpus_inventories.py b/infra/deploy/test_compare_corpus_inventories.py index d77a731..a346249 100644 --- a/infra/deploy/test_compare_corpus_inventories.py +++ b/infra/deploy/test_compare_corpus_inventories.py @@ -1,4 +1,6 @@ +import fcntl import json +import os import sqlite3 import tempfile import unittest @@ -20,6 +22,7 @@ class CompareCorpusInventoryTests(unittest.TestCase): def _artifact(self, name, root, documents=1, fragments=2, rows=None): directory = self.base / name directory.mkdir(mode=0o700) + (directory / ".inventory.lock").touch(mode=0o600) database = directory / "inventory.sqlite3" connection = sqlite3.connect(database) connection.execute( @@ -89,6 +92,41 @@ class CompareCorpusInventoryTests(unittest.TestCase): with self.assertRaisesRegex(ComparisonError, "differ in 1 inventory rows"): compare_inventories(self.source, changed) + def test_busy_source_lock_is_rejected(self): + self._assert_busy_lock(self.source, "source") + + def test_busy_staging_lock_is_rejected(self): + self._assert_busy_lock(self.staging, "staging") + + def _assert_busy_lock(self, result_path, label): + descriptor = os.open(result_path.parent / ".inventory.lock", os.O_RDWR) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + with self.assertRaisesRegex(ComparisonError, rf"{label}: inventory is active"): + compare_inventories(self.source, self.staging) + finally: + os.close(descriptor) + + def test_missing_lock_is_rejected_without_creating_it(self): + lock = self.source.parent / ".inventory.lock" + lock.unlink() + with self.assertRaisesRegex(ComparisonError, "source: cannot open inventory lock"): + compare_inventories(self.source, self.staging) + self.assertFalse(lock.exists()) + + def test_symlink_lock_is_rejected(self): + lock = self.source.parent / ".inventory.lock" + lock.unlink() + target = self.base / "external-lock" + target.touch(mode=0o600) + lock.symlink_to(target) + with self.assertRaisesRegex(ComparisonError, "source: cannot open inventory lock"): + compare_inventories(self.source, self.staging) + + def test_read_only_lock_can_be_shared(self): + (self.source.parent / ".inventory.lock").chmod(0o400) + self.assertEqual(compare_inventories(self.source, self.staging)["documents"], 1) + if __name__ == "__main__": unittest.main()