diff --git a/.kiro/specs/pre-push-safety-net/implementation-group-6.md b/.kiro/specs/pre-push-safety-net/implementation-group-6.md index 8aeb0fb..ab10ca0 100644 --- a/.kiro/specs/pre-push-safety-net/implementation-group-6.md +++ b/.kiro/specs/pre-push-safety-net/implementation-group-6.md @@ -37,3 +37,35 @@ graphs cannot qualify. Query locations normalize to repository-relative paths. - CLAIM: Task 6.1 identity and lease behavior, not persistent runner reuse or rollout readiness. - EVIDENCE: tests/test_graph_reuse.py, 105 focused passing tests, full suite and real Joern control above; measurement JSON retained in benchmarks/measurements/2026-09-13-graph-lease-control.json. - GAPS: Warm push feasibility and independent capability eligibility remain unproven. + +## Task 6.2 brief and review + +Requirements 4.1, 4.3, 4.4 and 7.3; design: Cache and Engine Boundary. Private +0700 cache roots, 0600 files, fixed lock stripes plus a publication lock, streamed payload +digests, manifest receipts, atomic completed directory publication and size accounting +(payload plus manifests) implement local storage. Incomplete data is never published as +complete. A lookup holds a shared lease; eviction requires an exclusive lock and skips +leased entries. Abandoned private partials count toward the limit and can be reclaimed. +All lock waits and copying check the caller's absolute deadline; cleanup uses its allowance. +No graph node mutation, executable deserialization or repository writes are introduced. + +## Status Report +- STATUS: READY_FOR_REVIEW +- TASK: 6.2 +- RED_PHASE_OUTPUT: `/tmp/group6-2-red.log`: seven tests failed with the temporary cache switch OFF. ON passed; switch removed; nine final cache tests passed. +- TESTS_RUN: 13 cache/audit tests passed, full suite 1256 passed and 9 skipped (104.11s); ruff and mypy passed. Final receipt-validation repair was followed by the 13-test focused rerun. +- EVIDENCE: concurrent subprocess writers, corrupted payload/manifest, interrupted atomic replace, abandoned partial files, bounded lock contention, leased eviction, unsafe paths and preservation of consumer exceptions. + +## Review Verdict +- VERDICT: APPROVED +- TASK: 6.2 +- MECHANICAL_RESULTS: canonical suite and final focused regression passed; static checks and diff whitespace passed; no new placeholders or secrets. +- FINDINGS: review added metadata receipt validation and fixed contextmanager exception ownership; lock inodes are retained in 64 bounded stripes to avoid unlink/recreate races. Storage size means bytes of payloads/manifests, not filesystem allocation overhead. +- SUMMARY: Complete publication or explicit miss is preserved through tested concurrency, corruption and interruption paths. + +## Verification Result +- STATUS: VERIFIED +- CLAIM_TYPE: TASK +- CLAIM: Task 6.2 bounded local artifact publication and eviction. +- EVIDENCE: nine cache contract tests and full suite above; production cache implementation, not a mocked storage path. +- GAPS: Semantic keys and live replay integration follow in 6.3 and 6.4; no warm-latency claim. diff --git a/.kiro/specs/pre-push-safety-net/tasks.md b/.kiro/specs/pre-push-safety-net/tasks.md index 123b78d..842cb99 100644 --- a/.kiro/specs/pre-push-safety-net/tasks.md +++ b/.kiro/specs/pre-push-safety-net/tasks.md @@ -167,7 +167,7 @@ hook latency is admitted. - Done when identical valid inputs can reuse a graph while any changed graph input or failed census rejects the lease; original cached artifacts survive query loading unchanged. - _Boundary: Cache and Engine Boundary, generic driver integration_ - _Requirements: 4.2, 4.3, 4.4, 7.3_ -- [ ] 6.2 Publish and evict local artifacts safely +- [x] 6.2 Publish and evict local artifacts safely - Add private temporary publication, per-key bounded locking, atomic completed manifests, cache size enforcement and unleased eviction. - Retain partial records only as diagnostics; never interpret a directory or failed write as a complete result. Do not rebuild graphs incrementally by mutating old node identities. - Done when concurrent writers, corrupt entries and interrupted publication yield either a validated complete artifact or an explicit miss without exceeding the remaining lock budget. diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index 74065d9..a60b336 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,10 +1,10 @@ { "counts": { - "load_bearing": 102, + "load_bearing": 103, "standalone_capability": 9 }, - "modules": 111, - "note": "Refreshed for task 5.1: explicit local replay integrates the existing ranker and comparison policy.", + "modules": 112, + "note": "Refreshed for task 6.2: bounded local artifact storage, integrated by task 6.4.", "orphaned": [], "rows": [ { @@ -128,6 +128,7 @@ "classification": "load_bearing", "importers": [ "cli", + "cpg.artifact", "cpg.backend", "model.config_value", "model.dominance", @@ -136,6 +137,7 @@ "model.pipeline", "model.scan", "model.taint", + "push.cache", "push.policy", "push.runner" ], @@ -144,7 +146,10 @@ }, { "classification": "load_bearing", - "importers": [], + "importers": [ + "cpg.backend", + "push.cache" + ], "module": "cpg.artifact", "reason": "pre-push-safety-net Req 4.3/7.3: backend graph identity and live lease contract; validation integration pending" }, @@ -152,6 +157,7 @@ "classification": "load_bearing", "importers": [ "cli", + "cpg.artifact", "model.config_value", "model.dominance", "model.judge", @@ -159,6 +165,7 @@ "model.pipeline", "model.scan", "model.taint", + "push.cache", "push.policy", "push.runner" ], @@ -363,6 +370,7 @@ "model.specs", "model.taint", "preprocess", + "push.cache", "push.contracts", "push.policy", "push.report", @@ -409,6 +417,7 @@ "model.partitions", "model.regions", "model.scan", + "push.cache", "push.contracts", "push.policy", "push.report", @@ -687,6 +696,12 @@ "module": "push", "reason": "pre-push-safety-net Req 5.5/7.3: experimental push contract package; no hook installed" }, + { + "classification": "load_bearing", + "importers": [], + "module": "push.cache", + "reason": "pre-push-safety-net task 6.2: bounded artifact publication and leased eviction; runner integration in 6.4" + }, { "classification": "load_bearing", "importers": [ diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index 9dd5648..7544f0e 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -28,6 +28,7 @@ ROOTS = ("gate", "map_gate", "pair_gate", "cli", "__main__") # Modules that stand alone: their own CLI subcommand or external protocol, deliberately off the scan path. SPEC_OWNED = { + "push.cache": "pre-push-safety-net task 6.2: bounded artifact publication and leased eviction; runner integration in 6.4", "push.report": "pre-push-safety-net Req 5.1/5.2/5.5/7.3: compact output and complete artifacts; runner integration pending", "push.policy": "pre-push-safety-net Req 2.4/3.1/3.4/4.1: targeted same-semantics base comparison; replay integration pending", "push.snapshot": "pre-push-safety-net Req 1.1/1.3/1.4/7.1: local immutable ref resolution; runner integration pending", diff --git a/src/openultrasast/push/cache.py b/src/openultrasast/push/cache.py new file mode 100644 index 0000000..cf3296f --- /dev/null +++ b/src/openultrasast/push/cache.py @@ -0,0 +1,217 @@ +"""Private local artifacts with bounded locks, complete publication and leased eviction.""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import re +import stat +import tempfile +import time +from collections.abc import Iterator +from contextlib import ExitStack, contextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from openultrasast.cpg.artifact import digest_value, graph_bytes, read_bytes +from openultrasast.cpg.backend import _DeadlineExpired, _remove_owned_tree +from openultrasast.model.contracts import ExecutionBudget + +_KEY = re.compile(r"[0-9a-f]{64}") + + +@dataclass(frozen=True) +class CacheEntry: + path: Path + metadata: dict[str, Any] + + +class ArtifactCache: + def __init__(self, root: Path, *, max_bytes: int) -> None: + if max_bytes <= 0: + raise ValueError("cache_size_must_be_positive") + if root.is_symlink(): + raise ValueError("cache_root_symlink") + root.mkdir(parents=True, exist_ok=True, mode=0o700) + info = root.stat() + if info.st_uid != os.getuid() or info.st_mode & 0o077: + raise ValueError("cache_root_must_be_private_and_owned") + self.root = root.resolve() + self.max_bytes = max_bytes + self.last_reason = "" + self.hits = 0 + self.misses = 0 + + @contextmanager + def _lock(self, name: str, budget: ExecutionBudget, *, shared: bool = False, wait: bool = True) -> Iterator[bool]: + fd = None + acquired = False + try: + fd = os.open(self.root / name, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600) + if not stat.S_ISREG(os.fstat(fd).st_mode): + raise ValueError("cache_lock_not_regular") + while time.monotonic() < budget.deadline_monotonic: + try: + fcntl.flock(fd, (fcntl.LOCK_SH if shared else fcntl.LOCK_EX) | fcntl.LOCK_NB) + acquired = True + break + except BlockingIOError: + if not wait: + break + time.sleep(min(0.01, max(0, budget.deadline_monotonic - time.monotonic()))) + yield acquired + finally: + if fd is not None: + if acquired: + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + + @staticmethod + def _bucket(key: str) -> str: + # Fixed lock stripes bound lock-file growth; never unlink a live lock inode. + return f".lock-{int(key[:2], 16) % 64:02d}" + + def _entry(self, key: str, budget: ExecutionBudget) -> CacheEntry | None: + directory = self.root / key + if directory.is_symlink() or not directory.is_dir(): + return None + manifest_path = directory / "manifest.json" + if manifest_path.stat().st_size > 1024 * 1024: + return None + manifest = json.loads(read_bytes(manifest_path, budget.deadline_monotonic)) + if ( + not isinstance(manifest, dict) + or manifest.get("version") != 1 + or manifest.get("key") != key + or manifest.get("complete") is not True + ): + return None + receipt = manifest.pop("receipt", None) + if receipt != digest_value(manifest): + return None + if not isinstance(manifest.get("metadata"), dict): + return None + payload = directory / "payload" + if payload.stat().st_size != manifest.get("size") or payload.stat().st_size > self.max_bytes: + return None + if graph_bytes(payload, budget.deadline_monotonic) != manifest.get("sha256"): + return None + return CacheEntry(payload, manifest["metadata"]) + + @contextmanager + def lookup(self, key: str, budget: ExecutionBudget) -> Iterator[CacheEntry | None]: + entry = None + if not _KEY.fullmatch(key): + self.misses += 1 + self.last_reason = "invalid_cache_key" + yield None + return + with ExitStack() as stack: + try: + locked = stack.enter_context(self._lock(self._bucket(key), budget, shared=True)) + if locked: + entry = self._entry(key, budget) + except (OSError, ValueError, TimeoutError): + entry = None + if entry is None: + self.misses += 1 + self.last_reason = "cache_miss_or_incompatible" + else: + self.hits += 1 + yield entry + + def _remove(self, root: Path, budget: ExecutionBudget) -> None: + _remove_owned_tree(root, budget.deadline_monotonic) + + def _make_room(self, incoming: int, key: str, budget: ExecutionBudget) -> bool: + entries = [] + used = 0 + for path in self.root.iterdir(): + if time.monotonic() >= budget.deadline_monotonic: + return False + if path.is_symlink(): + continue + if path.is_dir() and (_KEY.fullmatch(path.name) or path.name.startswith(".pending-")): + size = 0 + for child in path.iterdir(): + if time.monotonic() >= budget.deadline_monotonic: + return False + size += child.lstat().st_size + used += size + entries.append((path.stat().st_mtime, path, size)) + for _, path, size in sorted(entries): + if used + incoming <= self.max_bytes: + return True + if path.name == key or path.name.startswith(".pending-"): + self._remove(path, budget) + used -= size + else: + with self._lock(self._bucket(path.name), budget, wait=False) as locked: + if locked: + self._remove(path, budget) + used -= size + return used + incoming <= self.max_bytes + + def publish(self, key: str, payload: bytes | Path, metadata: dict[str, Any], budget: ExecutionBudget, *, complete: bool = True) -> bool: + if not complete or not _KEY.fullmatch(key): + self.last_reason = "partial_or_invalid_cache_publication" + return False + pending = None + try: + with self._lock(self._bucket(key), budget) as locked: + if not locked: + return False + with self._lock(".publication-lock", budget) as publishing: + if not publishing: + return False + size = len(payload) if isinstance(payload, bytes) else payload.stat().st_size + if not size or size > self.max_bytes: + return False + sha = ( + hashlib.sha256(payload).hexdigest() + if isinstance(payload, bytes) + else graph_bytes(payload, budget.deadline_monotonic) + ) + record = {"version": 1, "key": key, "complete": True, "metadata": metadata, "size": size, "sha256": sha} + record["receipt"] = digest_value(record) + manifest = json.dumps(record, sort_keys=True).encode() + if len(manifest) > 1024 * 1024 or not self._make_room(size + len(manifest), key, budget): + return False + pending = Path(tempfile.mkdtemp(prefix=".pending-", dir=self.root)) + target = pending / "payload" + if isinstance(payload, bytes): + with target.open("xb") as stream: + for offset in range(0, len(payload), 1024 * 1024): + if time.monotonic() >= budget.deadline_monotonic: + raise TimeoutError("deadline_exhausted") + stream.write(payload[offset : offset + 1024 * 1024]) + elif graph_bytes(payload, budget.deadline_monotonic, target) != sha: + raise ValueError("publication_source_changed") + target.chmod(0o600) + with target.open("rb") as stream: + os.fsync(stream.fileno()) + with (pending / "manifest.json").open("xb") as stream: + stream.write(manifest) + stream.flush() + os.fsync(stream.fileno()) + (pending / "manifest.json").chmod(0o600) + if time.monotonic() >= budget.deadline_monotonic: + raise TimeoutError("deadline_exhausted") + existing = self.root / key + if existing.exists() or existing.is_symlink(): + self._remove(existing, budget) + os.replace(pending, existing) + pending = None + return True + except (OSError, ValueError, TimeoutError, _DeadlineExpired) as error: + self.last_reason = type(error).__name__ + return False + finally: + if pending is not None: + try: + _remove_owned_tree(pending, budget.deadline_monotonic + budget.cancellation_allowance_seconds) + except (OSError, _DeadlineExpired): + self.last_reason = "cache_partial_cleanup_incomplete" diff --git a/tests/test_push_cache.py b/tests/test_push_cache.py new file mode 100644 index 0000000..2f33908 --- /dev/null +++ b/tests/test_push_cache.py @@ -0,0 +1,148 @@ +"""Publication, locking, corruption and eviction are observable cache contracts.""" + +import time + +from openultrasast.cpg.artifact import digest_value +from openultrasast.model.contracts import ExecutionBudget +from openultrasast.push.cache import ArtifactCache + + +def budget(seconds=3): + return ExecutionBudget(time.monotonic() + seconds, 0.2) + + +def test_completed_publication_is_readable_and_private(tmp_path): + cache = ArtifactCache(tmp_path / "cache", max_bytes=10000) + key = digest_value("one") + assert cache.publish(key, b"real bytes", {"kind": "test"}, budget()) + with cache.lookup(key, budget()) as entry: + assert entry is not None + assert entry.path.read_bytes() == b"real bytes" + assert entry.metadata == {"kind": "test"} + assert entry.path.stat().st_mode & 0o077 == 0 + assert cache.root.stat().st_mode & 0o077 == 0 + + +def test_partial_and_corrupt_entries_are_misses(tmp_path): + cache = ArtifactCache(tmp_path / "cache", max_bytes=10000) + key = digest_value("one") + assert not cache.publish(key, b"partial", {}, budget(), complete=False) + with cache.lookup(key, budget()) as entry: + assert entry is None + assert cache.publish(key, b"good", {}, budget()) + (cache.root / key / "payload").write_bytes(b"bad") + with cache.lookup(key, budget()) as entry: + assert entry is None + assert cache.publish(key, b"repaired", {}, budget()) + (cache.root / key / "manifest.json").write_text("{}") + with cache.lookup(key, budget()) as entry: + assert entry is None + + +def test_eviction_skips_leases_and_enforces_limit(tmp_path): + cache = ArtifactCache(tmp_path / "cache", max_bytes=1700) + one, two = digest_value("one"), digest_value("two") + assert cache.publish(one, b"a" * 1000, {}, budget()) + with cache.lookup(one, budget()) as entry: + assert entry is not None + assert not cache.publish(two, b"b" * 1000, {}, budget(0.05)) + assert entry.path.read_bytes() == b"a" * 1000 + assert cache.publish(two, b"b" * 1000, {}, budget()) + with cache.lookup(one, budget()) as entry: + assert entry is None + with cache.lookup(two, budget()) as entry: + assert entry is not None + + +def test_lock_wait_cannot_exceed_transaction_budget(tmp_path): + cache = ArtifactCache(tmp_path / "cache", max_bytes=10000) + key = digest_value("one") + assert cache.publish(key, b"x", {}, budget()) + with cache.lookup(key, budget()) as entry: + assert entry is not None + started = time.monotonic() + assert not cache.publish(key, b"y", {}, budget(0.05)) + assert time.monotonic() - started < 0.25 + with cache.lookup(key, budget()) as entry: + assert entry.path.read_bytes() == b"x" + + +def test_interrupted_publication_never_creates_complete_entry(tmp_path, monkeypatch): + import os + + cache = ArtifactCache(tmp_path / "cache", max_bytes=10000) + key = digest_value("one") + + def fail(*args): + raise OSError("interrupted before atomic publication") + + monkeypatch.setattr(os, "replace", fail) + assert not cache.publish(key, b"x", {}, budget()) + with cache.lookup(key, budget()) as entry: + assert entry is None + assert not list(cache.root.glob(".pending-*")) + + +def test_concurrent_writers_publish_only_complete_payloads(tmp_path): + import multiprocessing + + cache = ArtifactCache(tmp_path / "cache", max_bytes=10000) + key = digest_value("one") + + def write(value): + assert cache.publish(key, value * 2000, {"writer": value.decode()}, budget()) + + processes = [multiprocessing.get_context("fork").Process(target=write, args=(v,)) for v in (b"a", b"b")] + for process in processes: + process.start() + for process in processes: + process.join(5) + assert process.exitcode == 0 + with cache.lookup(key, budget()) as entry: + assert entry.path.read_bytes() == entry.metadata["writer"].encode() * 2000 + + +def test_symlink_and_invalid_key_are_rejected(tmp_path): + import pytest + + target = tmp_path / "target" + target.mkdir() + link = tmp_path / "link" + link.symlink_to(target) + with pytest.raises(ValueError): + ArtifactCache(link, max_bytes=10000) + cache = ArtifactCache(tmp_path / "cache", max_bytes=10000) + with cache.lookup("../escape", budget()) as entry: + assert entry is None + assert not cache.publish("../escape", b"x", {}, budget()) + key = digest_value("link") + (cache.root / key).symlink_to(target) + with cache.lookup(key, budget()) as entry: + assert entry is None + + +def test_lookup_failure_is_miss_and_consumer_exception_is_preserved(tmp_path): + import pytest + + cache = ArtifactCache(tmp_path / "cache", max_bytes=10000) + key = digest_value("one") + assert cache.publish(key, b"bytes", {}, budget()) + with pytest.raises(OSError, match="consumer"), cache.lookup(key, budget()) as entry: + assert entry is not None + raise OSError("consumer") + lock = cache.root / cache._bucket(key) + lock.unlink() + lock.symlink_to(tmp_path / "absent") + with cache.lookup(key, budget()) as entry: + assert entry is None + + +def test_abandoned_partial_files_count_toward_size_and_are_evicted(tmp_path): + cache = ArtifactCache(tmp_path / "cache", max_bytes=1700) + abandoned = cache.root / ".pending-dead-writer" + abandoned.mkdir(mode=0o700) + (abandoned / "payload").write_bytes(b"a" * 1000) + key = digest_value("one") + assert cache.publish(key, b"b" * 1000, {}, budget()) + assert not abandoned.exists() + assert sum(p.stat().st_size for p in cache.root.rglob("*") if p.is_file()) <= cache.max_bytes