diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c315d2..f0d4940 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [1.2.2] - 2026-08-04 + ### Added - **`GET /health` now carries a `cascade` readiness block** — `healthy`, @@ -96,6 +98,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **Maintenance deadlines now cover the whole call, not just the critical + section.** Resolving a table handle sat outside the timeout, so a hang there + never returned — and because the scheduler runs one maintenance task per kind + and skips a kind whose task is in flight, that table stopped being maintained + permanently and silently (a soak run caught one table 13 minutes without a + reclaim, retained versions climbing, while its siblings reclaimed normally and + nothing was logged because nothing failed). Handle resolution moved inside the + deadline for all seven locked operations, the lock-free compaction beat got + its own deadline, and the scheduler adds a last-resort 180s bound on the whole + call. The per-kind staleness alert added in this release is what surfaced it. - **AGENTIC search crashed on agent memory** (`agent_case` / `agent_skill`) — candidate metadata now satisfies the everalgo `_format_docs` contract, removing a `TypeError` in the sufficiency / multi-query steps. diff --git a/docs/openapi.json b/docs/openapi.json index 8af418a..63653c6 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -3,7 +3,7 @@ "info": { "title": "everos", "description": "md-first memory extraction framework", - "version": "1.2.1" + "version": "1.2.2" }, "paths": { "/health": { diff --git a/pyproject.toml b/pyproject.toml index 10b7e58..a413d2d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "everos" -version = "1.2.1" +version = "1.2.2" description = "EverOS — local-first markdown memory framework for AI agents and user chats; lightweight, dev-friendly, small-team" license = {text = "Apache-2.0"} readme = "README.md" diff --git a/src/everos/core/persistence/lancedb/repository.py b/src/everos/core/persistence/lancedb/repository.py index 102008f..53a8025 100644 --- a/src/everos/core/persistence/lancedb/repository.py +++ b/src/everos/core/persistence/lancedb/repository.py @@ -67,6 +67,13 @@ minutes covers a multi-million-row table with wide headroom.""" # leaves a real write window before the next attempt (review P2 / N1). _PRUNE_TIMEOUT_SECONDS = 60.0 +_COMPACT_TIMEOUT_SECONDS = 60.0 +"""Deadline on the lock-free compaction beat. It takes no lock, so it cannot +block writers — but it still must not hang: the scheduler runs one maintenance +task per kind and skips a kind whose task is in flight, so a compaction that +never returns parks that kind's maintenance permanently. Measured at ~460ms on +a table with 77 retained versions.""" + _SLOW_HOLD_LOG_SECONDS = 1.0 """Log a completed critical section that held the write lock at least this long. Normal writes are 2-25ms and a normal prune ~40ms, so anything past a @@ -191,6 +198,29 @@ class LanceRepoBase[T: BaseLanceTable]: """ return cls._table_locks.setdefault(table_name, asyncio.Lock()) + @asynccontextmanager + async def _deadline(self, budget: float, op: str) -> AsyncIterator[None]: + """Bound an operation that does **not** take the write lock. + + Same last-resort guarantee as :meth:`_locked` minus the lock: the + maintenance scheduler runs one task per kind and skips a kind whose + task has not finished, so any await in that path which can hang must + have a deadline or that kind stops being maintained for good. + """ + try: + async with asyncio.timeout(budget): + yield + except TimeoutError as exc: + logger.warning( + "lancedb_operation_deadline_exceeded", + table=self.table_name, + op=op, + budget_seconds=budget, + ) + raise VectorStoreBusyError( + f"{op} on table {self.table_name!r} exceeded its {budget:g}s deadline" + ) from exc + @asynccontextmanager async def _locked(self, budget: float, op: str) -> AsyncIterator[None]: """Hold the table write lock for at most ``budget`` seconds. @@ -209,6 +239,13 @@ class LanceRepoBase[T: BaseLanceTable]: is re-raised as :class:`VectorStoreBusyError` so the cascade worker treats it as transient and retries instead of marking the row permanently failed. + + Callers resolve the table handle **inside** this block, not before it. + Resolving it outside leaves an unbounded await ahead of the deadline, + and a maintenance task that hangs there never returns — which silently + parks that kind forever, because the scheduler skips a kind whose task + is still in flight (observed in a soak run: one table stopped pruning + for 13 minutes with zero failure logs while its siblings pruned fine). """ started = time.monotonic() acquired_at: float | None = None @@ -285,8 +322,8 @@ class LanceRepoBase[T: BaseLanceTable]: async def add(self, records: Sequence[T]) -> None: """Insert one or more records.""" - table = await self._table() async with self._locked(_WRITE_TIMEOUT_SECONDS, "add"): + table = await self._table() await table.add(list(records)) # ── Upsert ───────────────────────────────────────────────────────────── @@ -307,8 +344,8 @@ class LanceRepoBase[T: BaseLanceTable]: for the first time inserts; an entry that was edited in md updates its existing row. """ - table = await self._table() async with self._locked(_WRITE_TIMEOUT_SECONDS, "upsert"): + table = await self._table() await ( table.merge_insert(by) .when_matched_update_all() @@ -361,8 +398,9 @@ class LanceRepoBase[T: BaseLanceTable]: writer is benign here (compaction is not urgent — the next scheduled beat retries), so it must not stall writers. """ - table = await self._table() - await table.optimize() + async with self._deadline(_COMPACT_TIMEOUT_SECONDS, "optimize"): + table = await self._table() + await table.optimize() async def prune(self, older_than: dt.timedelta) -> None: """Physically reclaim files from versions older than ``older_than``. @@ -406,8 +444,8 @@ class LanceRepoBase[T: BaseLanceTable]: stall is rare. Does *not* shrink **active** index internals (FTS ``part_N`` / index UUID count) — that is ``rebuild_indexes``'s job. """ - table = await self._table() async with self._locked(_PRUNE_TIMEOUT_SECONDS, "prune"): + table = await self._table() await table.optimize(cleanup_older_than=older_than, delete_unverified=False) table_uri = await table.uri() removed = await asyncio.to_thread(_remove_empty_index_dirs, table_uri) @@ -470,8 +508,8 @@ class LanceRepoBase[T: BaseLanceTable]: in lance v7.0.0) - https://docs.rs/lancedb/latest/lancedb/table/struct.OptimizeOptions.html """ - table = await self._table() async with self._locked(_REBUILD_TIMEOUT_SECONDS, "rebuild_indexes"): + table = await self._table() for idx in await table.list_indices(): await table.drop_index(idx.name) await self.schema.ensure_fts_indexes(table) @@ -651,16 +689,16 @@ class LanceRepoBase[T: BaseLanceTable]: updates: Column-name to new-value mapping. where: SQL-like predicate scoping the update. """ - table = await self._table() async with self._locked(_WRITE_TIMEOUT_SECONDS, "update"): + table = await self._table() await table.update(updates, where=where) # ── Delete ───────────────────────────────────────────────────────────── async def delete(self, predicate: str) -> None: """Delete rows matching a SQL-like predicate.""" - table = await self._table() async with self._locked(_WRITE_TIMEOUT_SECONDS, "delete"): + table = await self._table() await table.delete(predicate) async def delete_by_md_path(self, md_path: str) -> int: @@ -670,8 +708,8 @@ class LanceRepoBase[T: BaseLanceTable]: (or when reverse-reconcile discovers an orphaned LanceDB row). Single quotes in ``md_path`` are doubled defensively. """ - table = await self._table() async with self._locked(_WRITE_TIMEOUT_SECONDS, "delete_by_md_path"): + table = await self._table() result = await table.delete(f"md_path = '{_q(md_path)}'") return int(result.num_deleted_rows) diff --git a/src/everos/memory/cascade/worker.py b/src/everos/memory/cascade/worker.py index 8926d73..89c6fb5 100644 --- a/src/everos/memory/cascade/worker.py +++ b/src/everos/memory/cascade/worker.py @@ -109,6 +109,19 @@ Once exhausted, ``mark_failed(retryable=False)`` so the reconciler stops re-enqueuing. Recover via ``cascade fix --apply`` (resets retry_count) or editing the md (mtime change resets retry_count).""" +_MAINTENANCE_TASK_TIMEOUT_SECONDS = 180.0 +"""Last-resort deadline on one whole maintenance call (compact or prune). + +The repo bounds its own critical sections, but this scheduler is the thing that +breaks if a call never returns at all: it runs one task per kind and skips a +kind whose task is still in flight, so a single non-returning await parks that +kind's maintenance forever — silently, since nothing failed. A soak run hit +exactly that (one table stopped pruning for 13 minutes with zero failure logs +while its siblings pruned normally), through an await that sat outside the +repo's deadline. Generous enough never to fire on a healthy beat (prune's own +budget is 60s), tight enough that a hang costs one cadence, not forever. +""" + DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS = 12 * 60 * 60.0 """How often (per kind) to do a full ``drop_index + create_index`` rebuild. @@ -775,13 +788,17 @@ class CascadeWorker: # only after the call returns. if state is not None: state.last_prune_attempt_at = now - await repo.prune(dt.timedelta(seconds=self._optimize_prune_retention)) + async with asyncio.timeout(_MAINTENANCE_TASK_TIMEOUT_SECONDS): + await repo.prune( + dt.timedelta(seconds=self._optimize_prune_retention) + ) if state is not None: state.last_prune_at = now else: # Light beat: lock-free compaction. A commit conflict here # is benign — handled below. - await repo.optimize() + async with asyncio.timeout(_MAINTENANCE_TASK_TIMEOUT_SECONDS): + await repo.optimize() if state is not None: state.optimize_failures = 0 logger.debug( diff --git a/tests/unit/test_core/test_persistence/test_lancedb/test_repository.py b/tests/unit/test_core/test_persistence/test_lancedb/test_repository.py index 06f7ade..36fb025 100644 --- a/tests/unit/test_core/test_persistence/test_lancedb/test_repository.py +++ b/tests/unit/test_core/test_persistence/test_lancedb/test_repository.py @@ -821,6 +821,42 @@ async def test_waiting_for_a_stuck_holder_also_times_out( await repo.add([_row(owner="u1", entry="n2")]) +async def test_a_hanging_table_handle_still_hits_the_deadline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Resolving the table handle must happen **inside** the deadline. + + With it outside, a hang there never returns, and the maintenance scheduler + runs one task per kind and skips a kind whose task is still in flight — so + that kind silently stops being maintained. A soak run hit exactly this: one + table went 13 minutes without a prune, zero failure logs, while its two + siblings pruned on schedule. + """ + from everos.core.errors import VectorStoreBusyError + from everos.core.persistence.lancedb import repository as repo_mod + + monkeypatch.setattr(repo_mod, "_PRUNE_TIMEOUT_SECONDS", 0.05) + monkeypatch.setattr(repo_mod, "_COMPACT_TIMEOUT_SECONDS", 0.05) + monkeypatch.setattr(repo_mod, "_WRITE_TIMEOUT_SECONDS", 0.05) + + class _HangingLookupRepo(_NoteRepo): + async def _table_lookup(self): # type: ignore[no-untyped-def] + await asyncio.sleep(30) # never resolves within the deadline + + repo = _HangingLookupRepo() + + # Every maintenance/write entry point must give up rather than park. + with pytest.raises(VectorStoreBusyError): + await repo.prune(dt.timedelta(seconds=60)) + with pytest.raises(VectorStoreBusyError): + await repo.optimize() + with pytest.raises(VectorStoreBusyError): + await repo.add([_row(owner="u1", entry="e1")]) + + # And the lock was never left held. + assert not repo._write_lock(repo.table_name).locked() + + def test_write_budgets_are_sized_from_measurements_not_guesses() -> None: """Write budgets must stay in the tens of seconds, not hundreds. diff --git a/uv.lock b/uv.lock index c52c2eb..4f40871 100644 --- a/uv.lock +++ b/uv.lock @@ -562,7 +562,7 @@ wheels = [ [[package]] name = "everos" -version = "1.2.1" +version = "1.2.2" source = { editable = "." } dependencies = [ { name = "aiosqlite" },