Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 134 additions & 1 deletion src/powercontext/builtin/code/cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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)}
Comment on lines +206 to +213

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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve reader-pinned generations after they leave current/previous

This set only covers the two generations in current.json, while reader leases can keep older generations alive. On 1072176e with real embedded SeekDB, I reconstructed the old-writer crash state by pointing the current identifier's journal entry at a vanished staging name. Initial recovery correctly skips it. I then paused a public symbols query after it pinned that generation and ran three index(full=True) calls without changing the source. On the third recovery, the identifier is no longer in current/previous, so its database rows are deleted even though the original directory and reader lock still exist. The pending query returns status="ok", the same fingerprint, and items=[] instead of run.

This occurs below the 64-entry window. The control with the journal naming its final directory keeps the rows and returns run. Recovery must reconcile and protect all generations still held by readers before deleting an identifier from an older publication record.

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", ""))
Comment on lines +284 to +285

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),
)
Comment on lines +294 to +301
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"")
Expand All @@ -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(
Expand Down Expand Up @@ -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}

Expand Down
21 changes: 19 additions & 2 deletions src/powercontext/builtin/code/graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -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: ...

Expand All @@ -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: ...
37 changes: 35 additions & 2 deletions src/powercontext/builtin/code/seekdb.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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)
)
Comment on lines +421 to +430
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
Expand Down
10 changes: 8 additions & 2 deletions src/powercontext/builtin/code/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down Expand Up @@ -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)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 5d26c4c9, before this round of review. interrupted_create in tests/e2e/test_native_code_seekdb.py now takes *args, **kwargs and forwards both, so the generation_id keyword reaches create() and the subprocess still stops at its pause marker. code-seekdb (3.11) and (3.14) both pass on ccec4880.

for path, issues in diagnostics.items():
key = entries[path]["extraction_key"]
content = json_bytes({"items": issues})
Expand All @@ -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 {
Expand Down
13 changes: 12 additions & 1 deletion src/powercontext/builtin/code/store.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading
Loading