Lock corpus artifacts during comparison

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
 

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

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:

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()