From 5e7c2ca692b5bee79048463b8f5cf4c48040600f Mon Sep 17 00:00:00 2001 From: colombod Date: Wed, 26 Aug 2026 13:23:46 +0000 Subject: [PATCH] feat(storage): lease_store backend, storage-boundary guard, and session_keys MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Storage-protocol hygiene, stacked on the schema-version work. Carries only the two storage-agnosticism concerns from the maintenance re-seat -- no maintenance mode, auto-repair, or blob-reclaim (those land later). Writer-lease persistence moves behind a new backend-neutral lease_store package (protocol + filesystem + factory), the fourth storage backend alongside blob_store, queue_manager, and identity_store. The writer-lease detector keeps all policy -- staleness, conflict, the bounded single-thread I/O executor -- and reaches the lease only through the store, so the same detector runs unchanged against any backend. A standing AST tripwire (test_storage_boundary_guard) asserts no module outside the four storage backend packages performs a storage-artifact file operation or reads a storage root path; it is proven red on a planted leak. It also catches a raw queues_dir glob or path-join. QueueManager gains session_keys() -- a backend-neutral way to enumerate every persisted session key. Boot reclaim sweeps through it instead of globbing the queue directory. queues_dir is removed from the QueueManager Protocol: the two main.py consumers now go through session_keys()/the session key, and the single sanctioned exception (registry.queues_dir_path, for the WriterLease boot detector) resolves the directory straight from settings. The Batch docstring now states its offsets are opaque queue-produced cursors, matching Record. Version 7.3.0. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- CHANGELOG.md | 28 ++++ .../lease_store/__init__.py | 8 + .../lease_store/factory.py | 19 +++ .../lease_store/filesystem.py | 89 ++++++++++ .../lease_store/protocol.py | 66 ++++++++ context_intelligence_server/main.py | 9 +- .../queue_manager/filesystem.py | 15 ++ .../queue_manager/protocol.py | 22 +-- context_intelligence_server/registry.py | 16 +- context_intelligence_server/writer_lease.py | 154 +++++------------- pyproject.toml | 2 +- tests/test_lease_store.py | 99 +++++++++++ tests/test_queue_manager.py | 14 ++ tests/test_storage_boundary_guard.py | 137 ++++++++++++++++ tests/test_writer_lease.py | 34 ++-- uv.lock | 2 +- 16 files changed, 559 insertions(+), 155 deletions(-) create mode 100644 context_intelligence_server/lease_store/__init__.py create mode 100644 context_intelligence_server/lease_store/factory.py create mode 100644 context_intelligence_server/lease_store/filesystem.py create mode 100644 context_intelligence_server/lease_store/protocol.py create mode 100644 tests/test_lease_store.py create mode 100644 tests/test_storage_boundary_guard.py diff --git a/CHANGELOG.md b/CHANGELOG.md index efff75b3..daf3532a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,34 @@ All notable changes to the Context Intelligence Server are recorded here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [7.3.0] + +### Added + +- **`lease_store` package.** Writer-lease persistence now lives behind a + backend-neutral `LeaseStore` protocol (`protocol` + `filesystem` + `factory`), + the fourth storage backend alongside `blob_store`, `queue_manager`, and + `identity_store`. The writer-lease detector keeps all policy (staleness, + conflict, the bounded single-thread I/O executor) and reaches the lease only + through the store, so the same detector runs unchanged against any backend. +- **`QueueManager.session_keys()`.** A backend-neutral way to enumerate every + persisted session key. Boot reclaim sweeps the queue through this method + instead of globbing the queue directory, so the sweep works unchanged against + any queue backend. +- **Storage-boundary guard test.** A standing AST tripwire asserts no module + outside the four storage backend packages performs a storage-artifact file + operation (glob/unlink/scandir) or reads a storage root path; it now also + catches a raw `queues_dir` glob/path-join. + +### Changed + +- **`queues_dir` removed from the `QueueManager` Protocol.** A caller enumerates + sessions via `session_keys()` and never learns the on-disk layout. The single + sanctioned exception (`registry.queues_dir_path`, used by the WriterLease boot + detector) resolves the directory straight from settings. The `Batch` + docstring now states its offsets are opaque queue-produced cursors, matching + `Record`'s contract. + ## [7.2.0] ### Added diff --git a/context_intelligence_server/lease_store/__init__.py b/context_intelligence_server/lease_store/__init__.py new file mode 100644 index 00000000..ca5fe090 --- /dev/null +++ b/context_intelligence_server/lease_store/__init__.py @@ -0,0 +1,8 @@ +"""Writer-lease persistence behind a backend-neutral Protocol.""" + +from __future__ import annotations + +from context_intelligence_server.lease_store.factory import create_lease_store +from context_intelligence_server.lease_store.protocol import LeaseRecord, LeaseStore + +__all__ = ["LeaseRecord", "LeaseStore", "create_lease_store"] diff --git a/context_intelligence_server/lease_store/factory.py b/context_intelligence_server/lease_store/factory.py new file mode 100644 index 00000000..0f55411d --- /dev/null +++ b/context_intelligence_server/lease_store/factory.py @@ -0,0 +1,19 @@ +"""Backend selection for the writer-lease store. + +One backend today (filesystem). A future backend is added here and nowhere +else -- the detector never learns which one it got. +""" + +from __future__ import annotations + +from collections.abc import Callable +from pathlib import Path + +from context_intelligence_server.lease_store.filesystem import FileSystemLeaseStore +from context_intelligence_server.lease_store.protocol import LeaseStore + + +def create_lease_store(dir_source: Callable[[], Path]) -> LeaseStore: + """Build the lease store. *dir_source* is resolved lazily per operation, so + nothing is constructed and no path is read at build time.""" + return FileSystemLeaseStore(dir_source) diff --git a/context_intelligence_server/lease_store/filesystem.py b/context_intelligence_server/lease_store/filesystem.py new file mode 100644 index 00000000..efdd4001 --- /dev/null +++ b/context_intelligence_server/lease_store/filesystem.py @@ -0,0 +1,89 @@ +"""Filesystem-backed writer-lease store. + +The lease is one atomically-replaced ``.writer.lease`` file in a directory +resolved lazily via *dir_source* -- resolved per operation (a cheap attribute +read, zero syscalls) so the store, like the detector it serves, constructs +nothing at build time and reflects a directory the tests may re-point. +""" + +from __future__ import annotations + +import json +import os +from collections.abc import Callable +from pathlib import Path + +from context_intelligence_server.lease_store.protocol import LeaseRecord + +LEASE_FILENAME = ".writer.lease" +LEASE_TMP_FILENAME = ".writer.lease.tmp" +_LEASE_VERSION = 1 + + +class FileSystemLeaseStore: + """A ``LeaseStore`` backed by a single atomically-written file on disk.""" + + def __init__(self, dir_source: Callable[[], Path]) -> None: + self._dir_source = dir_source + + def _path(self) -> Path: + return self._dir_source() / LEASE_FILENAME + + def read(self) -> LeaseRecord | None: + try: + text = self._path().read_text(encoding="utf-8") + except FileNotFoundError: + # A missing lease means "free directory", not a share fault. + return None + try: + data = json.loads(text.strip()) + return LeaseRecord( + owner=str(data["owner"]), + host=str(data.get("host", "")), + pid=int(data.get("pid", 0)), + started_at=float(data.get("started_at", 0.0)), + heartbeat=float(data["heartbeat"]), + revision=data.get("revision"), + server_version=str(data.get("server_version", "")), + lease_version=int(data.get("lease_version", -1)), + ) + except (json.JSONDecodeError, KeyError, TypeError, ValueError): + # Torn/malformed lease is treated as fresh-and-foreign, same + # strength as a genuine live peer. + return LeaseRecord( + owner="", + host="", + pid=0, + started_at=0.0, + heartbeat=0.0, + revision=None, + server_version="", + lease_version=-1, + unreadable=True, + ) + + def write(self, record: LeaseRecord) -> None: + directory = self._dir_source() + payload = { + "lease_version": record.lease_version, + "owner": record.owner, + "host": record.host, + "pid": record.pid, + "started_at": record.started_at, + "heartbeat": record.heartbeat, + "revision": record.revision, + "server_version": record.server_version, + } + tmp = directory / LEASE_TMP_FILENAME + tmp.write_text( + json.dumps(payload, separators=(",", ":")) + "\n", encoding="utf-8" + ) + os.replace(tmp, directory / LEASE_FILENAME) + + def delete_if_owned(self, owner: str) -> None: + rec = self.read() + if rec is not None and not rec.unreadable and rec.owner == owner: + try: + self._path().unlink() + except FileNotFoundError: + pass diff --git a/context_intelligence_server/lease_store/protocol.py b/context_intelligence_server/lease_store/protocol.py new file mode 100644 index 00000000..8281a555 --- /dev/null +++ b/context_intelligence_server/lease_store/protocol.py @@ -0,0 +1,66 @@ +"""The writer-lease persistence boundary. + +A single named lease record (owner, heartbeat, identity) persisted somewhere +durable. The writer-lease DETECTOR (``writer_lease.py``) owns all policy -- +staleness, conflict, the bounded-thread I/O executor -- and reaches the lease +only through this Protocol, so the same detector runs unchanged against any +backend (a filesystem file today, a blob lease or a row tomorrow). + +The methods are synchronous by contract: the detector runs each one on its own +private single-thread executor to bound a hung mount to a single leaked thread, +which ``asyncio.to_thread`` (shared pool) cannot guarantee. A backend whose I/O +is natively async wraps itself to satisfy this sync surface. +""" + +from __future__ import annotations + +import dataclasses +from typing import Protocol + + +@dataclasses.dataclass +class LeaseRecord: + """Parsed view of one persisted lease record. + + ``unreadable=True`` marks a synthetic record standing in for a torn or + hand-mangled lease (decode error / missing key / wrong type / unknown + ``lease_version``) -- treated at fresh-foreign strength, never at face + value. + """ + + owner: str + host: str + pid: int + started_at: float + heartbeat: float + revision: str | None + server_version: str + lease_version: int + unreadable: bool = False + + +class LeaseStore(Protocol): + """Persistence for exactly one writer-lease record. + + All three operations may raise ``OSError`` (a share fault); the detector + absorbs that as "not armed", never as a conflict. + """ + + def read(self) -> LeaseRecord | None: + """Return the current lease record, or ``None`` when no lease exists + (a free directory). A torn/malformed record returns a ``LeaseRecord`` + with ``unreadable=True`` rather than ``None``.""" + ... + + def write(self, record: LeaseRecord) -> None: + """Persist *record* as the current lease, atomically (a reader never + observes a half-written record).""" + ... + + def delete_if_owned(self, owner: str) -> None: + """Delete the lease only if it is still owned by *owner*. + + Never deletes a foreign lease: if a peer took it over, removing theirs + would actively hand the directory to a third writer. Best-effort: a + lease already gone is not an error.""" + ... diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py index 978ae00a..492e20ff 100644 --- a/context_intelligence_server/main.py +++ b/context_intelligence_server/main.py @@ -429,9 +429,9 @@ async def _boot_reclaim() -> None: # sync with test monkeypatches bound to the same object. settings = _settings boot_state.reclaim_enabled = settings.reclaim_enabled - # Iterate the QueueManager's own directory, not settings.queues_path -- - # the two can differ (tests do this routinely). - keys = sorted(p.stem for p in qm.queues_dir.glob("*.log")) + # Enumerate the queue through the backend-neutral protocol method, not a + # raw directory glob, so the sweep works unchanged against any backend. + keys = await qm.session_keys() reclaimed = 0 reclaimed_bytes = 0 kept = 0 @@ -469,9 +469,8 @@ async def _boot_reclaim() -> None: # gated: they can act on a log whose offset was merely unreadable. if c.verdict.value != "drained" and not settings.reclaim_enabled: logger.warning( - "boot_reclaimed reason=%s path=%s session=%s bytes=%d action=dry_run", + "boot_reclaimed reason=%s session=%s bytes=%d action=dry_run", c.reason, - qm.queues_dir / f"{key}.log", key, c.size, ) diff --git a/context_intelligence_server/queue_manager/filesystem.py b/context_intelligence_server/queue_manager/filesystem.py index 75972abc..b01b082a 100644 --- a/context_intelligence_server/queue_manager/filesystem.py +++ b/context_intelligence_server/queue_manager/filesystem.py @@ -1708,6 +1708,21 @@ def _scan() -> list[str]: return await asyncio.to_thread(_scan) + async def session_keys(self) -> list[str]: + """Return every persisted session key (sorted). + + A session key is present whenever the backend holds a queue log for it, + regardless of drain state or live-worker status. Callers that need to + sweep the whole queue (e.g. boot reclaim) enumerate here rather than + reaching into the backend's on-disk layout, so the sweep works + unchanged against any backend. + """ + + def _scan() -> list[str]: + return sorted(log.stem for log in self._dir.glob("*.log")) + + return await asyncio.to_thread(_scan) + async def is_fully_drained(self, session_id: str) -> bool: """True iff the session has no undrained log data left. diff --git a/context_intelligence_server/queue_manager/protocol.py b/context_intelligence_server/queue_manager/protocol.py index a439267e..fea3483c 100644 --- a/context_intelligence_server/queue_manager/protocol.py +++ b/context_intelligence_server/queue_manager/protocol.py @@ -41,10 +41,12 @@ class Batch: records: Queue-produced ``Record``s -- each carries its own opaque ``start``/``end`` cursor. The queue produces these offsets; a caller (the registry) only ever hands them back via ``commit``. - start_offset: Byte position in the log where this batch begins. - end_offset: Byte position in the log AFTER the last returned record. - This is the value passed to ``commit``. When no complete records - are available, ``end_offset == start_offset``. + start_offset: Opaque queue-produced cursor where this batch begins. + end_offset: Opaque queue-produced cursor AFTER the last returned + record -- the value handed back to ``commit``. Like ``Record``'s + ``start``/``end``, callers MUST NOT compute it or assume it is a + byte position; that framing is the queue's private invariant. When + no complete records are available, ``end_offset == start_offset``. """ session_id: str @@ -66,14 +68,12 @@ def lines(self) -> list[bytes]: class QueueManager(Protocol): """Durable, per-session append-only queue. - The method set mirrors the on-disk backend's public surface. A backend - reports its own queue root via ``queues_dir``; every other on-disk detail - stays private to the implementation. + The method set mirrors the on-disk backend's public surface. No ``Path`` or + on-disk-layout detail appears here: a caller enumerates sessions via + ``session_keys`` and never learns where (or whether) they live on a disk, + so the same consumers run unchanged against any backend. """ - @property - def queues_dir(self) -> Any: ... - async def heal_torn_tails(self) -> dict[str, int]: ... async def append(self, session_id: str, raw: bytes) -> None: ... @@ -110,6 +110,8 @@ async def reclaim_orphans( async def active_sessions(self) -> list[str]: ... + async def session_keys(self) -> list[str]: ... + async def recover(self) -> list[str]: ... async def derive_all_stats(self) -> dict[str, Any]: ... diff --git a/context_intelligence_server/registry.py b/context_intelligence_server/registry.py index fe0ef21b..1ad77539 100644 --- a/context_intelligence_server/registry.py +++ b/context_intelligence_server/registry.py @@ -132,15 +132,15 @@ def queue_manager(self) -> QueueManager: @property def queues_dir_path(self) -> Path: - """Queue directory path, resolved without constructing a QueueManager. - - Unlike ``queue_manager``, never calls ``_ensure_infra`` -- avoids a - race where an observer builds a second QueueManager for the same - directory. Falls back to the same expression ``_ensure_infra`` uses, - so the two can never disagree. + """Queue directory path for the WriterLease boot detector. + + Resolved straight from settings -- the single sanctioned exception to + the storage-boundary rule -- so the lease can locate the queue + directory WITHOUT constructing a QueueManager (avoiding a race where an + observer builds a second one for the same directory). This is the same + expression ``_ensure_infra`` feeds the queue-manager factory, so the + detector and the constructed queue can never disagree. """ - if self._queue_manager is not None: - return self._queue_manager.queues_dir return Path(get_settings().queues_path) def _ensure_neo4j_driver(self) -> Any: diff --git a/context_intelligence_server/writer_lease.py b/context_intelligence_server/writer_lease.py index f6110897..b58d4e38 100644 --- a/context_intelligence_server/writer_lease.py +++ b/context_intelligence_server/writer_lease.py @@ -20,8 +20,6 @@ import asyncio import concurrent.futures -import dataclasses -import json import logging import os import socket @@ -31,6 +29,11 @@ from pathlib import Path from typing import Any, Literal, Protocol +from context_intelligence_server.lease_store import ( + LeaseRecord, + LeaseStore, + create_lease_store, +) from context_intelligence_server.status import SERVER_VERSION logger = logging.getLogger("context_intelligence_server") @@ -53,8 +56,6 @@ class WriterLeaseSettings(Protocol): writer_lease_force_acquire: bool -LEASE_FILENAME = ".writer.lease" -LEASE_TMP_FILENAME = ".writer.lease.tmp" _LEASE_VERSION = 1 # Private, single-thread executor: all lease I/O runs here, never on the @@ -88,26 +89,6 @@ class WriterLeaseBusy(RuntimeError): fault by every caller -- never a conflict, never wedges anything.""" -@dataclasses.dataclass -class LeaseRecord: - """Parsed view of one on-disk `.writer.lease` line. - - `unreadable=True` marks a synthetic record standing in for a torn or - hand-mangled lease (JSONDecodeError / missing key / wrong type / unknown - `lease_version`) -- treated at fresh-foreign strength, never at face - value.""" - - owner: str - host: str - pid: int - started_at: float - heartbeat: float - revision: str | None - server_version: str - lease_version: int - unreadable: bool = False - - def _now() -> float: return time.time() @@ -140,8 +121,7 @@ def __init__(self) -> None: self._acquire_timeout: float | None = None self._dir_source: Callable[[], Path] | None = None - self._dir: Path | None = None - self._path: Path | None = None + self._store: LeaseStore | None = None # Observable state. self.acquired: bool = False @@ -159,81 +139,19 @@ def __init__(self) -> None: # The one-slot in-flight gate. self._io_inflight: bool = False - @property - def path(self) -> Path: - assert self._path is not None - return self._path - - # ----------------------------------------------------------------- - # Sync I/O primitives -- run ONLY via `_io()`, on the private executor. - # ----------------------------------------------------------------- - - def _read(self) -> LeaseRecord | None: - assert self._path is not None - try: - text = self._path.read_text(encoding="utf-8") - except FileNotFoundError: - # A missing lease means "free directory", not a share fault. - return None - try: - data = json.loads(text.strip()) - return LeaseRecord( - owner=str(data["owner"]), - host=str(data.get("host", "")), - pid=int(data.get("pid", 0)), - started_at=float(data.get("started_at", 0.0)), - heartbeat=float(data["heartbeat"]), - revision=data.get("revision"), - server_version=str(data.get("server_version", "")), - lease_version=int(data.get("lease_version", -1)), - ) - except (json.JSONDecodeError, KeyError, TypeError, ValueError): - # Torn/malformed lease is treated as fresh-and-foreign, same - # strength as a genuine live peer. - return LeaseRecord( - owner="", - host="", - pid=0, - started_at=0.0, - heartbeat=0.0, - revision=None, - server_version="", - lease_version=-1, - unreadable=True, - ) - - def _write(self, heartbeat: float) -> None: - assert self._dir is not None - assert self._path is not None - record = { - "lease_version": _LEASE_VERSION, - "owner": self.owner, - "host": self.host, - "pid": self.pid, - "started_at": self.started_at, - "heartbeat": heartbeat, - "revision": os.environ.get("CONTAINER_APP_REVISION"), - "server_version": SERVER_VERSION, - } - tmp = self._dir / LEASE_TMP_FILENAME - tmp.write_text( - json.dumps(record, separators=(",", ":")) + "\n", encoding="utf-8" + def _build_record(self, heartbeat: float) -> LeaseRecord: + """The lease record this process would write at *heartbeat* -- pure + identity, no I/O. The store persists it; the detector owns it.""" + return LeaseRecord( + owner=self.owner, + host=self.host, + pid=self.pid, + started_at=self.started_at, + heartbeat=heartbeat, + revision=os.environ.get("CONTAINER_APP_REVISION"), + server_version=SERVER_VERSION, + lease_version=_LEASE_VERSION, ) - os.replace(tmp, self._path) - - def _unlink_if_owned(self) -> None: - """Best-effort, owner-gated unlink -- release()'s sync body. - - Never unlinks a foreign lease: if a peer stole it, deleting theirs - would actively hand the directory to a third process.""" - if self._path is None: - return - rec = self._read() - if rec is not None and not rec.unreadable and rec.owner == self.owner: - try: - self._path.unlink() - except FileNotFoundError: - pass # ----------------------------------------------------------------- # The dedicated single-thread executor + one-slot in-flight gate. @@ -280,7 +198,10 @@ async def acquire( """ # I/O-free prelude: attribute reads only. `_dir_source` assigned # first so a later prelude failure still leaves a real re-arm source. + # The store resolves the directory lazily per op, so this constructs + # nothing and reads no path here. self._dir_source = dir_source + self._store = create_lease_store(dir_source) self.mode = settings.writer_lease_mode self.heartbeat_seconds = settings.writer_lease_heartbeat_seconds self.staleness_seconds = ( @@ -341,12 +262,9 @@ async def _acquire_once_inner(self, *, refuse: bool, source: str) -> None: assert self._dir_source is not None assert self.staleness_seconds is not None assert self._confirm_delay is not None + assert self._store is not None - # Pure path read, zero syscalls -- this detector constructs nothing. - self._dir = self._dir_source() - self._path = self._dir / LEASE_FILENAME - - rec = await self._io(self._read) + rec = await self._io(self._store.read) if rec is not None and rec.owner != self.owner: age = 0.0 if rec.unreadable else (_now() - rec.heartbeat) if age < self.staleness_seconds: @@ -378,9 +296,10 @@ async def _acquire_once_inner(self, *, refuse: bool, source: str) -> None: ) heartbeat = _now() - await self._io(lambda: self._write(heartbeat)) + store = self._store + await self._io(lambda: store.write(self._build_record(heartbeat))) await asyncio.sleep(self._confirm_delay) - rec2 = await self._io(self._read) + rec2 = await self._io(self._store.read) if rec2 is None or rec2.owner != self.owner: if refuse: msg = f"lost the acquire race to owner={rec2.owner if rec2 else None}" @@ -408,11 +327,11 @@ def _latch_conflict(self, source: str, observed_owner: str | None) -> None: self.conflict_source = source def _refusal_message(self, rec: LeaseRecord, age: float) -> str: - assert self._dir is not None + assert self._dir_source is not None assert self.staleness_seconds is not None return ( "Refusing to boot: another writer holds the queue-directory lease.\n" - f" dir = {self._dir}\n" + f" dir = {self._dir_source()}\n" f" foreign owner = {rec.owner} (host={rec.host} pid={rec.pid} " f"revision={rec.revision} version={rec.server_version})\n" f" lease age = {age:.1f}s (stale after " @@ -465,7 +384,8 @@ async def _renew_once(self) -> None: logger.warning("writer_lease: tick failed (%s), will retry", exc) async def _renew_once_inner(self) -> None: - rec = await self._io(self._read) + assert self._store is not None + rec = await self._io(self._store.read) if rec is None or rec.owner != self.owner: self.conflict = True self.conflict_source = "runtime" # unconditional upgrade @@ -478,15 +398,17 @@ async def _renew_once_inner(self) -> None: ) return heartbeat = _now() - await self._io(lambda: self._write(heartbeat)) + store = self._store + await self._io(lambda: store.write(self._build_record(heartbeat))) self.last_renewed = heartbeat async def _observe_only(self) -> None: """Held-then-lost: read-only, best-effort, bounded. Never writes.""" assert self._acquire_timeout is not None + assert self._store is not None try: rec = await asyncio.wait_for( - self._io(self._read), timeout=self._acquire_timeout + self._io(self._store.read), timeout=self._acquire_timeout ) except (OSError, WriterLeaseBusy, TimeoutError): return @@ -530,11 +452,15 @@ async def release(self) -> None: failed shutdown -- the next boot just waits out the staleness window. Bounded by the acquire timeout so a hung mount can never block shutdown.""" - if self.mode is None or self.mode == "off" or self._path is None: + if self.mode is None or self.mode == "off" or self._store is None: return + store = self._store timeout = self._acquire_timeout if self._acquire_timeout is not None else 5.0 try: - await asyncio.wait_for(self._io(self._unlink_if_owned), timeout=timeout) + await asyncio.wait_for( + self._io(lambda: store.delete_if_owned(self.owner)), + timeout=timeout, + ) except TimeoutError: logger.warning("writer_lease: release timed out after %.1fs", timeout) except (OSError, WriterLeaseBusy) as exc: diff --git a/pyproject.toml b/pyproject.toml index 474c4b50..35f981f0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "context-intelligence-server" -version = "7.2.0" +version = "7.3.0" description = "Context Intelligence Server for Amplifier" requires-python = ">=3.11" dependencies = [ diff --git a/tests/test_lease_store.py b/tests/test_lease_store.py new file mode 100644 index 00000000..cf19fc1d --- /dev/null +++ b/tests/test_lease_store.py @@ -0,0 +1,99 @@ +"""Filesystem lease-store: the writer-lease persistence backend. + +The writer-lease detector (``writer_lease.py``) reaches the lease only through +this store, so these tests pin the persistence contract independently of the +detector's policy. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from context_intelligence_server.lease_store import LeaseRecord, create_lease_store +from context_intelligence_server.lease_store.filesystem import ( + LEASE_FILENAME, + FileSystemLeaseStore, +) + +pytestmark = pytest.mark.integration + + +def _store(directory: Path) -> FileSystemLeaseStore: + store = create_lease_store(lambda: directory) + assert isinstance(store, FileSystemLeaseStore) + return store + + +def _record(owner: str = "me", heartbeat: float = 100.0) -> LeaseRecord: + return LeaseRecord( + owner=owner, + host="h", + pid=7, + started_at=1.0, + heartbeat=heartbeat, + revision="rev", + server_version="6.7.3", + lease_version=1, + ) + + +def test_read_missing_is_none(tmp_path: Path) -> None: + """A missing lease reads as None (a free directory), never an error.""" + assert _store(tmp_path).read() is None + + +def test_write_then_read_roundtrips(tmp_path: Path) -> None: + store = _store(tmp_path) + store.write(_record(owner="alice", heartbeat=42.0)) + got = store.read() + assert got is not None + assert got.owner == "alice" + assert got.heartbeat == 42.0 + assert got.unreadable is False + + +def test_write_is_atomic_no_tmp_left(tmp_path: Path) -> None: + store = _store(tmp_path) + store.write(_record()) + assert (tmp_path / LEASE_FILENAME).exists() + assert list(tmp_path.glob("*.tmp")) == [] + + +def test_torn_lease_reads_unreadable(tmp_path: Path) -> None: + """A hand-mangled lease is a synthetic unreadable record (fresh-foreign + strength), not None and not a crash.""" + (tmp_path / LEASE_FILENAME).write_text("{not json", encoding="utf-8") + got = _store(tmp_path).read() + assert got is not None + assert got.unreadable is True + + +def test_delete_if_owned_only_deletes_own(tmp_path: Path) -> None: + store = _store(tmp_path) + store.write(_record(owner="mine")) + + # A foreign lease is never deleted -- deleting it would hand the directory + # to a third writer. + store.delete_if_owned("someone_else") + assert store.read() is not None + + store.delete_if_owned("mine") + assert store.read() is None + + +def test_delete_if_owned_absent_is_noop(tmp_path: Path) -> None: + """Best-effort: deleting an already-absent lease is not an error.""" + _store(tmp_path).delete_if_owned("mine") # no raise + + +def test_dir_source_resolved_lazily(tmp_path: Path) -> None: + """The store constructs nothing and reads no path at build time -- the + directory is resolved per operation, so a store built before its directory + exists still works once it does.""" + target = tmp_path / "queues" + store = create_lease_store(lambda: target) + target.mkdir() # created AFTER the store was built + store.write(_record(owner="late")) + got = store.read() + assert got is not None and got.owner == "late" diff --git a/tests/test_queue_manager.py b/tests/test_queue_manager.py index a488c4e7..606df8ca 100644 --- a/tests/test_queue_manager.py +++ b/tests/test_queue_manager.py @@ -320,6 +320,20 @@ async def test_active_sessions_excludes_fully_committed(qm): assert active == ["s_active"] +async def test_session_keys_empty_for_fresh_queue(qm): + assert await qm.session_keys() == [] + + +async def test_session_keys_lists_every_persisted_session_sorted(qm): + # session_keys enumerates every session that has a log, regardless of drain + # state -- both the undrained and the fully-committed one appear, sorted. + await qm.append("s_beta", b"x") # undrained + await qm.append("s_alpha", b"y") + done = await qm.read_batch("s_alpha", max_items=10) + await qm.commit("s_alpha", done.end_offset, None) # drained but still present + assert await qm.session_keys() == ["s_alpha", "s_beta"] + + async def test_is_fully_drained_true_for_unknown_session(qm): assert await qm.is_fully_drained("never_seen") is True diff --git a/tests/test_storage_boundary_guard.py b/tests/test_storage_boundary_guard.py new file mode 100644 index 00000000..d42f47dd --- /dev/null +++ b/tests/test_storage_boundary_guard.py @@ -0,0 +1,137 @@ +"""Best-effort AST tripwire for the storage-agnosticism boundary. + +Storage artifacts (blobs, durable queues, identity stores) are reached ONLY +through their backend Protocols. No module OUTSIDE the three backend packages +may enumerate, stat, or unlink a storage artifact, nor read a storage root +path from settings -- otherwise a second backend (e.g. Azure) could not be +dropped in without editing consumers. + +This guard walks the AST of every non-storage module and fails on the file +operations and settings reads that would reach around a Protocol. It is a +TRIPWIRE, not a proof: a determined caller can defeat any static check +(dynamic import, getattr, os.system, a C-extension). The real guarantee is +that each consumer is positively verified protocol-only by reading. This test +exists to catch the accidental reintroduction of a KNOWN leak shape, and to +fail loudly the moment one lands. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +# The three storage backend packages -- the ONLY place a raw file operation on +# a storage artifact is allowed to live. +_STORAGE_PACKAGES = {"blob_store", "queue_manager", "identity_store", "lease_store"} + +# Attribute calls that mutate or enumerate a storage artifact on disk. +_BANNED_CALLS = { + ("os", "unlink"), + ("os", "remove"), + ("os", "removedirs"), + ("os", "rmdir"), + ("os", "scandir"), + ("os", "listdir"), + ("os", "walk"), +} +# Attribute names that are banned regardless of the receiver (glob on any Path, +# any shutil operation, Path.unlink()). +_BANNED_METHODS = {"glob", "rglob", "iterdir"} +_BANNED_MODULE_PREFIXES = {"shutil"} + +# settings.*_path reads that leak a storage root into a consumer. The factory +# and config own these; nobody else reads them. +_BANNED_SETTINGS_ATTRS = {"blob_path", "queues_path"} + +# Queue-backend storage-root attributes that leak an on-disk layout into a +# consumer. ``queues_dir`` is off the QueueManager Protocol -- a consumer that +# reads it (to ``.glob`` or path-join a ``.log``) is reaching around the +# backend-neutral seam and must go through ``session_keys`` instead. +_BANNED_QUEUE_ATTRS = {"queues_dir"} + +# The single human-approved exception (workspace AGENTS.md): the WriterLease +# boot detector resolves the queue directory WITHOUT constructing a +# QueueManager, so registry.queues_dir_path reads settings.queues_path. +_APPROVED_EXCEPTIONS = { + ("context_intelligence_server/registry.py", "queues_path"), +} + +_SERVER_ROOT = Path(__file__).resolve().parent.parent / "context_intelligence_server" +_SCRIPTS_ROOT = Path(__file__).resolve().parent.parent / "scripts" + + +def _iter_guarded_files() -> list[Path]: + files: list[Path] = [] + for root in (_SERVER_ROOT, _SCRIPTS_ROOT): + if not root.exists(): + continue + for path in root.rglob("*.py"): + parts = set(path.relative_to(root.parent).parts) + if parts & _STORAGE_PACKAGES: + continue # backend packages are the sanctioned home + files.append(path) + return files + + +def _rel(path: Path) -> str: + try: + return str(path.relative_to(_SERVER_ROOT.parent)) + except ValueError: + # A path outside the repo (e.g. the planted-leak self-test's tmp file). + return str(path) + + +def _violations(path: Path) -> list[str]: + tree = ast.parse(path.read_text("utf-8"), filename=str(path)) + rel = _rel(path) + found: list[str] = [] + + for node in ast.walk(tree): + # os.unlink(...) / os.walk(...) / shutil.rmtree(...) etc. + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + attr = node.func.attr + recv = node.func.value + if isinstance(recv, ast.Name): + if (recv.id, attr) in _BANNED_CALLS: + found.append(f"{rel}:{node.lineno} {recv.id}.{attr}(...)") + if recv.id in _BANNED_MODULE_PREFIXES: + found.append(f"{rel}:{node.lineno} {recv.id}.{attr}(...)") + if attr in _BANNED_METHODS: + found.append(f"{rel}:{node.lineno} .{attr}(...)") + if attr == "unlink": # Path(...).unlink() + found.append(f"{rel}:{node.lineno} .unlink(...)") + + # settings.blob_path / settings.queues_path reads + if isinstance(node, ast.Attribute) and node.attr in _BANNED_SETTINGS_ATTRS: + if (rel, node.attr) in _APPROVED_EXCEPTIONS: + continue + found.append(f"{rel}:{node.lineno} settings.{node.attr}") + + # qm.queues_dir reads (leaked queue storage root -- .glob / path-join) + if isinstance(node, ast.Attribute) and node.attr in _BANNED_QUEUE_ATTRS: + if (rel, node.attr) in _APPROVED_EXCEPTIONS: + continue + found.append(f"{rel}:{node.lineno} .{node.attr}") + + return found + + +def test_no_storage_file_ops_outside_backend_packages() -> None: + """No consumer reaches around a storage Protocol with a raw file op.""" + violations: list[str] = [] + for path in _iter_guarded_files(): + violations.extend(_violations(path)) + + assert not violations, ( + "storage-agnosticism boundary breached -- storage artifacts must be " + "reached only through their backend Protocol. Offending sites:\n " + + "\n ".join(sorted(violations)) + ) + + +def test_guard_actually_detects_a_planted_leak(tmp_path: Path) -> None: + """The tripwire is armed: a planted os.unlink is caught (red-on-violation).""" + leak = tmp_path / "context_intelligence_server" / "routers" / "leaky.py" + leak.parent.mkdir(parents=True) + leak.write_text("import os\n\n\ndef f(p):\n os.unlink(p)\n", "utf-8") + assert _violations(leak), "guard failed to detect a planted os.unlink leak" diff --git a/tests/test_writer_lease.py b/tests/test_writer_lease.py index 3f11ee2f..f1cf07af 100644 --- a/tests/test_writer_lease.py +++ b/tests/test_writer_lease.py @@ -24,6 +24,7 @@ import httpx import pytest from context_intelligence_server.config import Settings +from context_intelligence_server.lease_store.filesystem import FileSystemLeaseStore from context_intelligence_server.main import lifespan from context_intelligence_server.queue_manager import FileSystemQueueManager, QueueManager from context_intelligence_server.status import boot_state @@ -594,10 +595,10 @@ async def test_share_fault_at_boot_continues_in_every_mode( ) -> None: lease = WriterLease() - def _boom() -> None: + def _boom(_self: object) -> None: raise OSError(errno.ESTALE, "stale file handle") - monkeypatch.setattr(lease, "_read", _boom) + monkeypatch.setattr(FileSystemLeaseStore, "read", _boom) await lease.acquire(_settings(writer_lease_mode=mode), lambda: tmp_path) @@ -644,17 +645,15 @@ async def test_hung_mount_acquire_times_out(tmp_path: Path) -> None: lease = WriterLease() blocker = threading.Event() - def _hang() -> None: + def _hang(_self: object) -> None: blocker.wait(timeout=5.0) - monkeypatch_target = lease - monkeypatch_target._read = _hang # type: ignore[method-assign] - - start = time.monotonic() - await lease.acquire( - _settings(writer_lease_acquire_timeout_seconds=0.1), lambda: tmp_path - ) - elapsed = time.monotonic() - start + with patch.object(FileSystemLeaseStore, "read", _hang): + start = time.monotonic() + await lease.acquire( + _settings(writer_lease_acquire_timeout_seconds=0.1), lambda: tmp_path + ) + elapsed = time.monotonic() - start assert lease.acquired is False assert lease.error is not None @@ -663,16 +662,18 @@ def _hang() -> None: blocker.set() -async def test_hung_mount_does_not_starve_shared_pool(tmp_path: Path) -> None: +async def test_hung_mount_does_not_starve_shared_pool( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: """F2's bound: a stalled detector can never starve the append/commit path, which run on the SHARED default executor.""" lease = WriterLease() blocker = threading.Event() - def _hang() -> None: + def _hang(_self: object) -> None: blocker.wait(timeout=3.0) - lease._read = _hang # type: ignore[method-assign] + monkeypatch.setattr(FileSystemLeaseStore, "read", _hang) await lease.acquire( _settings( @@ -721,7 +722,8 @@ def _boom_once(*args: object, **kwargs: object) -> object: patch.object(wl_module._LEASE_IO, "submit", side_effect=_boom_once), pytest.raises(WriterLeaseBusy), ): - await lease._io(lease._read) + assert lease._store is not None + await lease._io(lease._store.read) assert lease._io_inflight is False # Re-arm: unpatched, a subsequent op succeeds. @@ -836,7 +838,7 @@ async def test_status_writer_lease_present_during_boot_zero_disk_reads( lease = main_module.writer_lease await lease.acquire(_settings(), lambda: tmp_path) - lease_path = lease.path + lease_path = tmp_path / ".writer.lease" read_calls = {"n": 0} real_read_text = Path.read_text diff --git a/uv.lock b/uv.lock index 7e613eab..9cd64dbd 100644 --- a/uv.lock +++ b/uv.lock @@ -233,7 +233,7 @@ wheels = [ [[package]] name = "context-intelligence-server" -version = "7.2.0" +version = "7.3.0" source = { editable = "." } dependencies = [ { name = "aiofiles" },