diff --git a/src/powercontext/builtin/code/cache.py b/src/powercontext/builtin/code/cache.py index fb179ccfe0..6b3da96b8e 100644 --- a/src/powercontext/builtin/code/cache.py +++ b/src/powercontext/builtin/code/cache.py @@ -125,6 +125,11 @@ def source(self, path: str, deadline: float) -> bytes: class GenerationCache: + #: How many publication records a binding normally carries. The bound is + #: enforced by recalling the commits it retires, never by forgetting a handle, + #: so a binding that keeps many generations pinned may stand above it. + PUBLICATION_WINDOW = 64 + def __init__(self, directory: Path, store: GraphStore | None = None) -> None: self.directory = directory self.store = store if store is not None else SQLiteGraphStore() @@ -198,7 +203,116 @@ def staging(self) -> Path: (path / child).mkdir(mode=0o700) return path - def publish(self, staging: Path, manifest: dict[str, Any], deadline: float, max_bytes: int) -> Generation: + def publications(self) -> dict[str, dict[str, Any]]: + path = self.directory / "publication.json" + if not path.exists(): + return {} + records = read_json(path, maximum=64 * 1024).get("attempts") + if not isinstance(records, dict): + raise CodeError("index_integrity_failed") + return {str(attempt): record for attempt, record in records.items() if isinstance(record, dict)} + + def begin_publication(self, staging: Path, deadline: float) -> str: + """Name the attempt before the store writes any row under it. + + The identifier has to come from here rather than the store: a commit that + has to be recalled later must be nameable even when the directory that + carried its name is gone. + """ + attempt = uuid.uuid4().hex + records = self.publications() + records[attempt] = {"directory": staging.name, "state": "pending", "started_at": time.time()} + self._write_publications(records, deadline) + return attempt + + def recover(self, deadline: float) -> int: + """Reclaim commits whose generation directory no longer exists. + + ``collect()`` starts from directory names, so it cannot see a commit whose + directory was removed out of band — a wiped cache, a pruned staging area. + The records are the only place that still knows the identifier. + + A missing directory is not proof that the commit is unwanted. The pointer + can still reach a generation whose manifest names the identifier these rows + were committed under, and a reader may be holding it. Reclaiming on + directory absence alone deletes live data, so that name is checked first. + """ + records = self.publications() + retained = self._retained_identifiers() + abandoned = [ + attempt + for attempt, record in records.items() + if attempt not in retained and not (self.directory / self._recorded_directory(record)).exists() + ] + for attempt in abandoned: + self.store.remove_generation(attempt, self.directory.name, deadline) + records.pop(attempt) + if abandoned or len(records) > self.PUBLICATION_WINDOW: + self._write_publications(records, deadline) + return len(abandoned) + + def _retained_identifiers(self) -> set[str]: + """Identifiers the pointer still reaches, read from the generations themselves. + + A row is addressed by its identifier, not by the directory that carries it, + so the manifest inside each reachable generation is the only place that can + say which rows a live generation owns. A record whose directory has gone can + still point at one of them — that is the crash this guard exists for. + """ + try: + pointer = self.pointer() + except CodeError: + return set() + retained: set[str] = set() + for reference in (pointer or {}).values(): + if not isinstance(reference, dict): + continue + directory = reference.get("directory") + if not isinstance(directory, str): + continue + try: + manifest = read_json(self.directory / directory / "manifest.json") + except (CodeError, OSError): + continue + graph = manifest.get("graph") + identifier = graph.get("id") if isinstance(graph, dict) else None + if isinstance(identifier, str): + retained.add(identifier) + return retained + + @staticmethod + def _recorded_directory(record: dict[str, Any]) -> str: + return str(record.get("directory", "")) + + def _write_publications(self, records: dict[str, dict[str, Any]], deadline: float) -> None: + excess = len(records) - self.PUBLICATION_WINDOW + if excess > 0: + # A record is the only handle to rows that outlive their directory, so the + # window recalls those commits instead of forgetting them. A record whose + # directory still exists is a generation a reader may hold; it is never + # dropped, and the excess is allowed to stand. + stale = sorted( + ( + (attempt, record) + for attempt, record in records.items() + if not (self.directory / self._recorded_directory(record)).exists() + ), + key=lambda item: item[1].get("started_at", 0), + ) + for attempt, _ in stale[:excess]: + self.store.remove_generation(attempt, self.directory.name, deadline) + records.pop(attempt) + self.replace_json("publication.json", {"attempts": records}) + + def publish( + self, + staging: Path, + manifest: dict[str, Any], + deadline: float, + max_bytes: int, + *, + attempt: str | None = None, + ) -> Generation: manifest_content = json_bytes(manifest) write_private(staging / "manifest.json", manifest_content) write_private(staging / "reader.lock", b"") @@ -221,6 +335,21 @@ def publish(self, staging: Path, manifest: dict[str, Any], deadline: float, max_ if size > max_bytes: raise CodeError("code_cache_limit_exceeded") directory = self.directory / ("generation-" + uuid.uuid4().hex) + if attempt is not None: + # Name the generation before it exists. Recording the final name only + # after the rename leaves a window where the record still points at a + # staging directory that has already gone: recovery would then read + # "directory missing" and delete the rows of a live generation. + # Written first, the record can only be wrong in the harmless + # direction — it names a directory that was never created, and the + # rows it recalls are the ones this attempt committed. + records = self.publications() + records[attempt] = { + "directory": directory.name, + "state": "published", + "started_at": records.get(attempt, {}).get("started_at", time.time()), + } + self._write_publications(records, deadline) os.replace(staging, directory) reference = {"directory": directory.name, "manifest_sha256": digest_bytes(manifest_content)} self.replace_json( @@ -269,6 +398,10 @@ def clear(self, deadline: float) -> dict[str, Any]: finally: os.close(descriptor) self.collect(None, deadline) + # A directory that was removed out of band leaves nothing for collect() to + # start from; clearing is the last moment the binding is still named, so + # the records have to be settled here or their rows are charged forever. + self.recover(deadline) retained = sum(path.is_dir() and not path.is_symlink() for path in self.directory.glob("generation-*")) return {"status": "cleared", "retained_reader_generations": retained} diff --git a/src/powercontext/builtin/code/graph.py b/src/powercontext/builtin/code/graph.py index f0de02b6de..db3b33e03c 100644 --- a/src/powercontext/builtin/code/graph.py +++ b/src/powercontext/builtin/code/graph.py @@ -37,8 +37,16 @@ class GraphStore(Protocol): identity: str def create( - self, directory: Path, nodes: list[dict[str, Any]], edges: list[dict[str, Any]], deadline: float - ) -> dict[str, Any]: ... + self, + directory: Path, + nodes: list[dict[str, Any]], + edges: list[dict[str, Any]], + deadline: float, + *, + generation_id: str | None = None, + ) -> dict[str, Any]: + """Write one generation; the caller may name it so a recall can address it later.""" + ... def verify(self, directory: Path, manifest: dict[str, Any], deadline: float) -> None: ... @@ -48,4 +56,13 @@ def open( def remove(self, directory: Path, deadline: float) -> None: ... + def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None: + """Drop a commit by identifier, for directories that no longer exist. + + ``remove()`` takes a directory and reads the identifier out of it. When the + directory is already gone, this is the only entry point that can still + reach the rows. + """ + ... + def size(self, directory: Path, deadline: float) -> int: ... diff --git a/src/powercontext/builtin/code/seekdb.py b/src/powercontext/builtin/code/seekdb.py index 298e031e29..c46b391011 100644 --- a/src/powercontext/builtin/code/seekdb.py +++ b/src/powercontext/builtin/code/seekdb.py @@ -241,10 +241,24 @@ def _reference(self, directory: Path) -> dict[str, Any]: return reference def create( - self, directory: Path, nodes: list[dict[str, Any]], edges: list[dict[str, Any]], deadline: float + self, + directory: Path, + nodes: list[dict[str, Any]], + edges: list[dict[str, Any]], + deadline: float, + *, + generation_id: str | None = None, ) -> dict[str, Any]: self.initialize(deadline) - reference = {"backend": self.identity, "binding": directory.parent.name, "id": uuid.uuid4().hex} + if generation_id is not None and ( + len(generation_id) != 32 or any(character not in "0123456789abcdef" for character in generation_id) + ): + raise CodeError("index_integrity_failed") + reference = { + "backend": self.identity, + "binding": directory.parent.name, + "id": generation_id or uuid.uuid4().hex, + } # Persist the cleanup handle before any database commit can leave orphan rows. # A killed writer may leave the temporary file incomplete, but remove() must # only see a complete descriptor or no committed database work at all. @@ -404,6 +418,25 @@ def remove(self, directory: Path, deadline: float) -> None: ) connection.commit() + def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None: + """Delete a commit whose directory is gone; the identifier is the only handle left.""" + with self.connection(deadline) as connection, connection.cursor() as cursor: + try: + for table in ("pc_code_edges", "pc_code_nodes"): + check_deadline(deadline) + cursor.execute(f"DELETE FROM {table} WHERE generation_id = %s", (generation_id,)) # noqa: S608 - constant tables. + cursor.execute( + "DELETE FROM pc_code_generations WHERE id = %s AND binding = %s", (generation_id, binding) + ) + connection.commit() + except pymysql.MySQLError as error: + # A binding whose first build was interrupted before initialize() has no + # tables yet, so nothing can be committed under this identifier. Handled + # inside the context because connection() turns MySQL errors into + # CodeError on exit, which would fail the retry this call exists to allow. + if error.args[0] != 1146: + raise + def _incomplete_staging(self, directory: Path, deadline: float) -> bool: # Older builders wrote graph.json in place before opening the transaction. # Recover their interrupted writes without accepting corrupt published graphs diff --git a/src/powercontext/builtin/code/service.py b/src/powercontext/builtin/code/service.py index ac9bb6ea72..afa9f8997e 100644 --- a/src/powercontext/builtin/code/service.py +++ b/src/powercontext/builtin/code/service.py @@ -90,6 +90,9 @@ def index(self, scope_id: str, *, full: bool = False) -> dict[str, Any]: if abandoned.is_dir() and not abandoned.is_symlink(): self.store.remove(abandoned, deadline) shutil.rmtree(abandoned) + # A staged build is not the only way to lose a directory: reclaim what the + # records still name, so a wiped generation does not keep paying for rows. + cache.recover(deadline) try: result = self._build(repository, cache, binding, deadline, full=full) result.update(status="ready", elapsed_seconds=time.monotonic() - started, finished_at=_now()) @@ -153,7 +156,8 @@ def _build( attributes.update(edge_count=len(edges), diagnostic_count=sum(map(len, diagnostics.values()))) nodes = [node for fact in facts for node in fact["nodes"]] with stage("store"): - storage = self.store.create(staging, nodes, edges, deadline) + publication = cache.begin_publication(staging, deadline) + storage = self.store.create(staging, nodes, edges, deadline, generation_id=publication) for path, issues in diagnostics.items(): key = entries[path]["extraction_key"] content = json_bytes({"items": issues}) @@ -169,7 +173,9 @@ def _build( manifest.update(storage) manifest["previous_fingerprint"] = previous.fingerprint if previous else None with stage("publish") as attributes: - generation = cache.publish(staging, manifest, deadline, self.config.limits.max_cache_bytes) + generation = cache.publish( + staging, manifest, deadline, self.config.limits.max_cache_bytes, attempt=publication + ) attributes.update(manifest["coverage"]) attributes.update(extracted_files=extracted, reused_files=len(entries) - extracted) return { diff --git a/src/powercontext/builtin/code/store.py b/src/powercontext/builtin/code/store.py index 4a2079356d..8da270b62d 100644 --- a/src/powercontext/builtin/code/store.py +++ b/src/powercontext/builtin/code/store.py @@ -200,14 +200,25 @@ class SQLiteGraphStore: identity = "sqlite" def create( - self, directory: Path, nodes: list[dict[str, Any]], edges: list[dict[str, Any]], deadline: float + self, + directory: Path, + nodes: list[dict[str, Any]], + edges: list[dict[str, Any]], + deadline: float, + *, + generation_id: str | None = None, ) -> dict[str, Any]: from powercontext.builtin.code.cache import hash_file + del generation_id # Rows live inside the generation directory, so no name is needed. path = directory / "graph.sqlite" create_graph(path, nodes, edges, deadline) return {"database_sha256": hash_file(path, deadline)} + def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None: + # A directory that no longer exists takes its SQLite rows with it. + del generation_id, binding, deadline + def verify(self, directory: Path, manifest: dict[str, Any], deadline: float) -> None: from powercontext.builtin.code.cache import hash_file diff --git a/tests/builtin/test_code_publication.py b/tests/builtin/test_code_publication.py new file mode 100644 index 0000000000..fa34e3773c --- /dev/null +++ b/tests/builtin/test_code_publication.py @@ -0,0 +1,312 @@ +# Copyright (c) 2026 OceanBase. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software distributed +# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR +# CONDITIONS OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. + +"""Publication records: the commit that outlives the directory which named it.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import time +from pathlib import Path + +import pytest + +from powercontext.builtin.code import CodeConfig, CodeRepositoryConfig, CodeService +from powercontext.builtin.code.cache import GenerationCache +from powercontext.builtin.code.store import SQLiteGraphStore + +pytest.importorskip("tree_sitter") +pytest.importorskip("tree_sitter_python") + +pytestmark = pytest.mark.skipif( + os.name != "posix", reason="native code indexing requires POSIX file locks and resource limits" +) + + +class RecordingStore(SQLiteGraphStore): + """A store that remembers which identifiers were committed and which were recalled. + + Both lists matter: recalling a name only reclaims the rows if they were + committed under that same name. A test that checks `reclaimed` alone passes + even when the store ignores the name it was given. + """ + + def __init__(self) -> None: + self.created: list[str | None] = [] + self.reclaimed: list[str] = [] + + def create( + self, + directory: Path, + nodes: list[dict[str, object]], + edges: list[dict[str, object]], + deadline: float, + *, + generation_id: str | None = None, + ) -> dict[str, object]: + self.created.append(generation_id) + return super().create(directory, nodes, edges, deadline, generation_id=generation_id) + + def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None: + self.reclaimed.append(generation_id) + + +class NamedStore(RecordingStore): + """Behaves like the SeekDB backend where it matters: the manifest carries a name. + + SQLite rows live inside the generation directory, so losing the directory loses + the rows and there is nothing to protect. The invariant this file is about only + has consequences when the rows outlive the directory, which is the shape here. + """ + + def create( + self, + directory: Path, + nodes: list[dict[str, object]], + edges: list[dict[str, object]], + deadline: float, + *, + generation_id: str | None = None, + ) -> dict[str, object]: + reference = super().create(directory, nodes, edges, deadline, generation_id=generation_id) + return { + "graph": { + "backend": "named", + "binding": directory.parent.name, + "id": generation_id, + **{key: value for key, value in reference.items() if key == "database_sha256"}, + } + } + + +def git(root: Path, *arguments: str) -> None: + executable = shutil.which("git") + assert executable + subprocess.run([executable, "-C", str(root), *arguments], check=True, capture_output=True) + + +def cache_of(service: CodeService) -> GenerationCache: + directories = [path for path in (service.config.cache_dir).iterdir() if path.is_dir()] + assert len(directories) == 1 + return GenerationCache(directories[0], service.store) + + +@pytest.fixture +def repository(tmp_path, request): + root = tmp_path / "repository" + (root / "src").mkdir(parents=True) + (root / "src/app.py").write_text("def run():\n return 1\n") + git(root, "init") + git(root, "add", ".") + git(root, "-c", "user.name=Code Test", "-c", "user.email=test@example.invalid", "commit", "-m", "fixture") + store = getattr(request, "param", RecordingStore)() + service = CodeService( + CodeConfig( + enabled=True, + repositories={"scope": CodeRepositoryConfig(root=root)}, + cache_dir=tmp_path / "cache", + ), + store=store, + ) + service.index("scope", full=True) + return service, root + + +def test_a_published_generation_is_recorded_under_its_final_directory(repository): + service, _ = repository + cache = cache_of(service) + + records = cache.publications() + pointer = cache.pointer() + assert pointer is not None + assert len(records) == 1 + _, record = records.popitem() + assert record["state"] == "published" + assert record["directory"] == pointer["current"]["directory"] + + +def test_reindexing_keeps_a_live_generation_out_of_the_sweep(repository): + service, root = repository + (root / "src/app.py").write_text("def run():\n return 2\n") + git(root, "add", ".") + + service.index("scope") + + assert service.store.reclaimed == [] + assert [record["state"] for record in cache_of(service).publications().values()] == ["published", "published"] + + +def test_a_generation_directory_removed_out_of_band_is_reclaimed_by_name(repository): + service, root = repository + cache = cache_of(service) + attempt = next(iter(cache.publications())) + pointer = cache.pointer() + assert pointer is not None + shutil.rmtree(cache.directory / pointer["current"]["directory"]) + (root / "src/app.py").write_text("def run():\n return 3\n") + git(root, "add", ".") + + service.index("scope") + + assert service.store.reclaimed == [attempt] + assert attempt not in cache.publications() + + +def test_a_pending_attempt_without_its_staging_directory_is_reclaimed(repository): + service, _ = repository + cache = cache_of(service) + attempt = "0" * 32 + journal = json.loads((cache.directory / "publication.json").read_text()) + journal["attempts"][attempt] = {"directory": "staging-nowhere", "state": "pending", "started_at": 0} + (cache.directory / "publication.json").write_text(json.dumps(journal)) + + reclaimed = cache.recover(time.monotonic() + 30) + + assert reclaimed == 1 + assert service.store.reclaimed == [attempt] + assert attempt not in cache.publications() + + +@pytest.mark.parametrize("repository", [NamedStore], indirect=True) +def test_the_recorded_name_is_the_identifier_the_store_committed(repository): + """The name in the record has to be the name the rows were written under. + + Recalling a name the store never used deletes nothing and raises nothing, so + the leak this change exists to fix would survive every other assertion here. + """ + + service, _ = repository + cache = cache_of(service) + attempt = next(iter(cache.publications())) + pointer = cache.pointer() + assert pointer is not None + manifest = json.loads( + (cache.directory / pointer["current"]["directory"] / "manifest.json").read_text(encoding="utf-8") + ) + + assert service.store.created == [attempt] + assert manifest["graph"]["id"] == attempt + + +@pytest.mark.parametrize("repository", [NamedStore], indirect=True) +def test_recovery_cannot_reclaim_rows_a_reachable_generation_still_owns(repository): + """A record naming a moved-away staging directory must not decide a live commit. + + This is the state a crash between the rename and the journal update used to + leave behind: `current.json` reaches a generation whose manifest owns the rows, + while the record still names the staging directory that no longer exists. + """ + + service, _ = repository + cache = cache_of(service) + attempt = next(iter(cache.publications())) + pointer = cache.pointer() + assert pointer is not None + journal = json.loads((cache.directory / "publication.json").read_text()) + journal["attempts"][attempt] = {"directory": "staging-renamed-away", "state": "pending", "started_at": 0} + (cache.directory / "publication.json").write_text(json.dumps(journal)) + + reclaimed = cache.recover(time.monotonic() + 30) + + assert reclaimed == 0 + assert service.store.reclaimed == [] + assert attempt in cache.publications() + + +def test_clear_settles_a_record_whose_directory_went_missing(repository): + """Clearing used to forget the handle and leave the rows charged to the binding. + + Directory absence is not evidence that collection reclaimed the commit: the + record is the only thing that still names the rows, so clearing has to settle + it rather than drop it. + """ + + service, _ = repository + cache = cache_of(service) + attempt = next(iter(cache.publications())) + pointer = cache.pointer() + assert pointer is not None + shutil.rmtree(cache.directory / pointer["current"]["directory"]) + + service.clear("scope") + + assert service.store.reclaimed == [attempt] + assert cache.publications() == {} + + +def test_the_record_window_never_drops_a_generation_that_still_exists(repository): + """The bound may not discard the only handle to rows that can still be recalled. + + A generation pinned by a reader keeps its directory, so `collect()` cannot + retire it. If the window evicted its record anyway, the rows would be + unreachable the moment the directory went away. + """ + + service, _ = repository + cache = cache_of(service) + kept = [f"{index:032x}" for index in range(70)] + abandoned = [f"{index:032x}" for index in range(1000, 1006)] + for name in kept: + (cache.directory / ("generation-" + name)).mkdir() + attempts = { + **{name: {"directory": "generation-" + name, "state": "published", "started_at": 1} for name in kept}, + **{name: {"directory": "generation-" + name, "state": "published", "started_at": 0} for name in abandoned}, + } + (cache.directory / "publication.json").write_text(json.dumps({"attempts": attempts})) + + cache.begin_publication(cache.staging(), time.monotonic() + 30) + + records = cache.publications() + assert sorted(service.store.reclaimed) == sorted(abandoned) + assert all(name in records for name in kept) + + +def test_a_commit_can_be_recalled_before_its_tables_exist(tmp_path, monkeypatch): + """The first build can name its commit and die before initialize() creates tables. + + ``connection()`` converts every MySQL error into ``CodeError`` on exit, so the + outer handler never saw 1146 and every retry failed during recovery — before + ``create()`` could create the tables it was waiting for. A missing table means + there is nothing to recall, not that the recall failed. + """ + + import pymysql + + from powercontext.builtin.code.seekdb import SeekDBGraphStore + + class Cursor: + def __enter__(self) -> Cursor: + return self + + def __exit__(self, *arguments: object) -> bool: + return False + + def execute(self, *arguments: object) -> None: + raise pymysql.err.ProgrammingError(1146, "Table 'test.pc_code_edges' doesn't exist") + + class Connection: + def __enter__(self) -> Connection: + return self + + def __exit__(self, *arguments: object) -> bool: + return False + + def begin(self) -> None: ... + def rollback(self) -> None: ... + def commit(self) -> None: ... + def cursor(self) -> Cursor: + return Cursor() + + monkeypatch.setattr(pymysql, "connect", lambda **options: Connection()) + store = SeekDBGraphStore(tmp_path / "seekdb", {}) + + store.remove_generation("a" * 32, "binding", time.monotonic() + 30) diff --git a/tests/e2e/test_native_code_seekdb.py b/tests/e2e/test_native_code_seekdb.py index 4aa874f491..b61d442575 100644 --- a/tests/e2e/test_native_code_seekdb.py +++ b/tests/e2e/test_native_code_seekdb.py @@ -294,8 +294,8 @@ def interrupted_write(path, content): seekdb.write_private = interrupted_write else: create = seekdb.SeekDBGraphStore.create - def interrupted_create(self, *args): - result = create(self, *args) + def interrupted_create(self, *args, **kwargs): + result = create(self, *args, **kwargs) pause() return result seekdb.SeekDBGraphStore.create = interrupted_create @@ -339,6 +339,24 @@ def interrupted_create(self, *args): assert cursor.fetchone()[0] == 0 +def test_a_record_names_the_row_the_commit_wrote(repository): + """The journal key has to be the identifier the rows were committed under. + + Recall deletes by name. If the store ignored the name it was given and minted + its own, the delete would match no rows and raise nothing, so the leak would be + back with every other test here still passing. + """ + + _, service = repository + directory = next(service.config.cache_dir.iterdir()) + cache = GenerationCache(directory, service.store) + pointer = cache.pointer() + assert pointer is not None + manifest = json.loads((directory / pointer["current"]["directory"] / "manifest.json").read_text()) + + assert manifest["graph"]["id"] in cache.publications() + + @pytest.mark.parametrize("content", (b"", b'{"backend":')) @pytest.mark.parametrize("recovery", ("sync", "index", "clear")) def test_legacy_incomplete_staging_descriptor_recovers(repository, content, recovery):