diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b2c275..3c315d2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,117 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **`GET /health` now carries a `cascade` readiness block** — `healthy`, + human-readable `reasons`, and the counters behind them (`pending`, + `failed_permanent`, `failed_retryable`, `drain_consecutive_failures`, + `unrecoverable_total`, `optimize_failure_streak`, `prune_stale_seconds`). + `null` when the app runs without the cascade lifespan. **Alert on + `cascade.healthy`**: it flips false only on operational faults — drain loop + failing (≥3 in a row), index maintenance wedged (≥5), or version cleanup + stalled on some table (≥3 missed 300s beats, and `reasons` names the table). + `failed_permanent` is a data-quality backlog awaiting `cascade fix` and + deliberately does **not** flip `healthy`, otherwise the signal sits red until + a human edits markdown. The HTTP status stays 200 even when the block says + unhealthy — it is a liveness signal, and a degraded projection must not + trigger a container restart. If the probe itself fails (locked / full + SQLite), the block returns `healthy=false` with a `cascade health probe + failed: …` reason and zeroed counters — read zeros next to that reason as + "unknown", not "clean". +- **`everos cascade rebuild` CLI command** — drops every business LanceDB table, + clears the cascade queue, and re-indexes all markdown from scratch. The + supported recovery from a drifted or corrupt index: unlike deleting the index + directory, it re-enqueues every file (a bare `rm -rf` leaves the queue marked + `done`, so nothing re-indexes and the index comes back empty), and unlike + deleting `.index/` it preserves SQLite state that markdown cannot rebuild — + notably `unprocessed_buffer`. **Requires the server to be stopped**: it + refuses to start (exit code `3`) while a server holds the OME lock, because a + live daemon keeps writing through cached table handles to the dropped + dataset. `--yes/-y` for non-interactive use; `Ctrl-C` exits `130` and the + re-index resumes on the next run or server start. +- **Startup schema verification now detects column *type* drift**, not just + missing / extra columns. Catches the class of corruption behind #337 — an + `episode.subject_vector` left as `string` by an older build while the schema + declares a 1024-d `fixed_size_list` — which a name-only check waved through + and which then failed deep inside `merge_insert` with an opaque + `LanceError(IO)`. The error now points at `everos cascade rebuild`. + +### Changed + +- **LanceDB maintenance is split into compaction and reclamation.** + `optimize()` is lock-free compaction; the new `prune()` runs + `cleanup_older_than` under the per-table write lock. Fixes unbounded index + growth: the previous bundled call issued a Rewrite that concurrent writes + kept preempting, so version cleanup lost the race indefinitely (a soak run + measured 16 successes against 547 conflicts over 21h, with the index + directory growing to the disk guardrail). Reclamation now completes on every + beat, at the cost of a brief same-table write stall (measured ~40ms). + Retention is decoupled from cadence: files older than 60s are eligible, + reclaimed on a 300s beat. +- **Every write-lock critical section is now bounded.** All seven operations + (`add` / `upsert` / `update` / `delete` / `delete_by_md_path` / `prune` / + `rebuild_indexes`) run under a deadline that covers **lock acquisition as + well as the body**, so no code path can wait for the lock — or hold it — + indefinitely. Budgets are sized from measured durations (row writes are + 2–25ms, worst observed 63ms → 15s; index rebuild → 300s; prune → 60s). + Expiry raises the retryable `VectorStoreBusyError`, so the cascade worker + retries the row instead of marking it permanently failed. Without this, one + operation stuck outside the old narrow timeout wedged a table permanently: + every writer blocked on acquire, and the maintenance scheduler skipped a kind + whose task never finished, so that table stopped reclaiming versions + altogether (observed: 150 versions retained, disk 11x live size, with nothing + logged because nothing failed). +- **Benign LanceDB commit conflicts no longer count as failures.** A lost + optimistic-concurrency race logs at `debug` on either maintenance beat. The + heavy beat needs this too: its write lock is in-process only, so a second + process (`cascade sync`, `cascade backfill`) can preempt its commit. Counting + those triggered spurious fallback index rebuilds, which drop every index + before recreating them — and if the rebuild also lost the race, the table sat + without an FTS index and every search on that kind returned 500 until the + next 12h sweep. +- **A query vector whose width disagrees with the embedding provider's declared + `dim` now fails immediately** with `CONFIGURATION_ERROR` instead of reaching + LanceDB. It previously surfaced as an opaque `ValueError` after the query was + built — 13–14s per request, as an unhandled 500. +- **Exception logging no longer renders frame locals.** structlog's default + traceback formatter (`show_locals=True`, up to 100 frames) rendered one + unhandled exception on an async stack into 6423 log lines — 85MB of logs + across 11 exceptions in one soak run — at ~290ms of synchronous CPU each, and + risked printing request payloads into logs. Now locals-off and capped at 15 + frames: the same traceback is 103 lines. +- **`cascade backfill` reclaims through the daemon's retention window** rather + than at zero age, so it cannot delete files out from under an in-flight + `/search` in the server process. +- **`lancedb` pinned to `>=0.34.0,<0.35.0`.** 0.35 embeds lance-rust v9; 0.34.0 + is the version validated under sustained churn. Environments installing from + `uv.lock` are unaffected (already 0.34.0). Never widen the floor below 0.34 — + older lance cannot read v8 data. + +### Fixed + +- **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. +- **Per-kind version-cleanup staleness is no longer masked.** The health signal + reported time since the newest successful prune across all kinds, so on a + multi-kind deployment one kind whose cleanup died was hidden by the others + pruning on schedule. It now reports the worst kind and names it. +- **`cascade backfill` silently skipped compaction and reclamation** — it still + called the removed `optimize(cleanup_older_than=…)` signature, and the + resulting `TypeError` was swallowed by a best-effort `except`, so the disk + growth this release fixes came back after every backfill. +- **Empty `_indices//` husks are removed after cleanup** (a soak run + accumulated 13061 directories, 98% of them empty), which bloated inode usage + and slowed directory scans. + +### Docs + +- Rewrote the cascade runbook's recovery paths: the `/health` cascade block and + its alert thresholds, `cascade rebuild` (including the stop-the-server + requirement), why `rm -rf .index/lancedb` yields an empty index, and why + `rm -rf .index` loses un-extracted buffered messages. + ## [1.2.1] - 2026-07-29 ### Added diff --git a/docs/cascade_runbook.md b/docs/cascade_runbook.md index aaea10d..3e0dcd9 100644 --- a/docs/cascade_runbook.md +++ b/docs/cascade_runbook.md @@ -49,6 +49,42 @@ lsn: never auto-clear these — they represent malformed md the user must edit. +### Machine-readable: the `cascade` block on `GET /health` + +`GET /health` carries a `cascade` block while the daemon runs (`null` +for an app built without the cascade lifespan). **Alert on +`cascade.healthy`** — it is the operational readiness verdict, and +`reasons` explains a `false` in plain text: + +```json +{"status": "ok", "cascade": { + "healthy": false, + "reasons": ["version cleanup stalled for kind 'episode' (1200s since its last prune — that table's index dir may grow)"], + "pending": 3, "failed_permanent": 1, "failed_retryable": 0, + "drain_consecutive_failures": 0, "unrecoverable_total": 4, + "optimize_failure_streak": 0, "prune_stale_seconds": 1200.0}} +``` + +What flips `healthy` false — and nothing else does: + +| Symptom | Signal | Threshold | +|---|---|---| +| Writes accepted but not projected to LanceDB | `drain_consecutive_failures` | ≥ 3 in a row | +| Index maintenance wedged | `optimize_failure_streak` | ≥ 5 in a row (lost commit races excluded — they are expected under churn) | +| Version cleanup stopped, that table's disk will grow | `prune_stale_seconds` (worst kind, named in `reasons`) | ≥ 900s (3 missed 300s beats) | + +`failed_permanent` is **informational only** — it is a data-quality +backlog awaiting `cascade fix`, so it never flips `healthy` (otherwise +the signal sits red until a human edits md). Watch it separately. + +The HTTP code is a *liveness* signal and stays 200 even when the block +says `healthy: false` — a degraded projection must not trigger a +container restart, which fixes neither a bad md file nor disk bloat. If +the probe itself fails (locked / full SQLite), the block comes back +`healthy: false` with a `cascade health probe failed: …` reason and the +counters zeroed — treat zeros alongside that reason as "unknown", not +as "clean". + ## Recovering from failures: `everos cascade fix` `cascade fix` (no flag) lists every failed row. With `--apply`: @@ -134,8 +170,11 @@ declare (or vice versa), the boot fails with: ``` LanceDB table 'episode' schema drift: missing=[...], extra=[...], -type_drift=[...]. The index is rebuildable from md — recover with -`everos cascade rebuild`. +type_drift=[...]. Recover with `everos cascade rebuild` (stop the server +first): it drops and re-indexes from md, preserving un-extracted buffered +messages. Restarting will not clear this — the startup migrations only +alter column nullability, never a column's name or type, so a name/type +drift never resolves on its own. ``` `verify_business_schemas` compares both the column **names** and their diff --git a/pyproject.toml b/pyproject.toml index 034ef70..10b7e58 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,8 +29,10 @@ dependencies = [ # Upper bound: 0.35.0 embeds lance-rust v9 (large storage/encoding jump, not # yet stable-released). 0.32-0.34 carry a compaction offset-overflow regression # (lance-format/lance#7653); we run 0.34.0 safely via the with_position=False - # FTS workaround. Do not float past 0.34.x until 0.35 is validated by the soak - # harness. Never widen this floor back below 0.34 (older lance can't read v8 data). + # FTS workaround. Do not float past 0.34.x until 0.35 has been validated under + # sustained churn (bounded + reclaimable index dir, no compaction crash) — see + # docs/cascade_runbook.md. Never widen this floor back below 0.34 (older lance + # cannot read v8 data). "lancedb>=0.34.0,<0.35.0", # Vector + BM25 + scalar filter (Arrow-based) "aiosqlite>=0.20.0", # Async SQLite driver (used by SA async engine) "sqlmodel>=0.0.22", # ORM (Pydantic + SQLAlchemy 2.0 async) diff --git a/src/everos/core/errors.py b/src/everos/core/errors.py index 4d24aed..bd6687b 100644 --- a/src/everos/core/errors.py +++ b/src/everos/core/errors.py @@ -129,6 +129,18 @@ class ExternalServiceError(InfrastructureError): """An external service (LLM, embedding, rerank) returned an error or timed out.""" +class VectorStoreBusyError(ExternalServiceError): + """A LanceDB table's write lock could not be taken, or the critical + section it guards overran its deadline. + + Deliberately under :class:`ExternalServiceError` rather than + :class:`VectorStoreError`: the cascade worker retries that branch, and a + contended or slow table is exactly the transient condition retrying is + for. Under :class:`VectorStoreError` the row would be marked permanently + failed and need a manual ``cascade fix``. + """ + + class LLMServiceError(ExternalServiceError): """The configured LLM provider returned an error or timed out.""" diff --git a/src/everos/core/observability/logging/factory.py b/src/everos/core/observability/logging/factory.py index 14ed276..1437758 100644 --- a/src/everos/core/observability/logging/factory.py +++ b/src/everos/core/observability/logging/factory.py @@ -84,7 +84,21 @@ def configure_logging(level: str = "INFO") -> None: foreign_pre_chain=shared_processors, processors=[ structlog.stdlib.ProcessorFormatter.remove_processors_meta, - structlog.dev.ConsoleRenderer(), + structlog.dev.ConsoleRenderer( + # structlog's default exception formatter is + # ``RichTracebackFormatter(show_locals=True, max_frames=100, + # extra_lines=3)``. On an async stack that is brutal: one + # unhandled exception in a soak run rendered 82 frames into + # **6423 log lines** (85MB of server.log across 11 such + # exceptions), and each render costs ~290ms of synchronous CPU + # inside the event loop. Locals also risk printing request + # payloads into logs. Keep the frames, drop the locals. + exception_formatter=structlog.dev.RichTracebackFormatter( + show_locals=False, + max_frames=15, + extra_lines=1, + ), + ), ], ) diff --git a/src/everos/core/persistence/lancedb/repository.py b/src/everos/core/persistence/lancedb/repository.py index 99057f5..102008f 100644 --- a/src/everos/core/persistence/lancedb/repository.py +++ b/src/everos/core/persistence/lancedb/repository.py @@ -12,18 +12,49 @@ from __future__ import annotations import asyncio import datetime as dt -from collections.abc import Sequence +import time +from collections.abc import AsyncIterator, Sequence +from contextlib import asynccontextmanager from pathlib import Path from typing import Any, ClassVar from lancedb import AsyncTable +from everos.core.errors import VectorStoreBusyError from everos.core.observability.logging import get_logger from .base import BaseLanceTable logger = get_logger(__name__) +# Write-lock deadlines, per operation class. Every critical section on a +# table's write lock runs under one of these (see ``LanceRepoBase._locked``): +# acquisition included, so neither waiting for the lock nor holding it can be +# unbounded. They are hang-catchers, not throughput limits — but they are sized +# from measured durations, not guessed, because the budget is also how long a +# wedged table stays invisible. +_WRITE_TIMEOUT_SECONDS = 15.0 +"""Row writes: add / upsert / update / delete. + +Measured on a local SSD across table sizes and batch sizes (10k–100k rows, +50–500 rows per call): median 2–25ms, worst observed 63ms — flat in both +dimensions, because these are append-and-commit operations, not scans. +``merge_insert`` (upsert) is the read-modify-write one and still lands at +5–25ms. + +15s is ~240x the worst observation, which covers a slow/contended disk and +several operations queued ahead on the same lock (the deadline includes +acquisition, and ``asyncio.Lock`` is FIFO so no waiter starves). Reaching it +means the table is not merely busy — it is stuck, and failing fast into the +worker's retry is better than blocking writers for minutes. Deliberately not +sized in the hundreds of seconds: the budget doubles as the detection latency +for a wedged table.""" + +_REBUILD_TIMEOUT_SECONDS = 300.0 +"""Index rebuild (drop + recreate every index) — the one genuinely slow +critical section, measured at ~0.3s per 50k rows per indexed column, so 5 +minutes covers a multi-million-row table with wide headroom.""" + # Safety cap on a single prune's ``optimize(cleanup_older_than=…)`` call — a # pure hang-catcher, not a bound on normal runtime. A real cleanup is # milliseconds even on a heavily churned table (measured ~40ms at 320k writes / @@ -36,6 +67,14 @@ logger = get_logger(__name__) # leaves a real write window before the next attempt (review P2 / N1). _PRUNE_TIMEOUT_SECONDS = 60.0 +_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 +second means writers were queued behind it — without this, a section that is +slow but under its deadline is invisible (the maintenance beat only logs at +``debug``), and a soak run left no way to tell whether a 16s stall was a slow +prune or a deep write queue.""" + def _remove_empty_index_dirs(table_uri: str) -> int: """Delete empty ``_indices//`` husks under a table dir; return count. @@ -152,6 +191,62 @@ class LanceRepoBase[T: BaseLanceTable]: """ return cls._table_locks.setdefault(table_name, asyncio.Lock()) + @asynccontextmanager + async def _locked(self, budget: float, op: str) -> AsyncIterator[None]: + """Hold the table write lock for at most ``budget`` seconds. + + **The deadline covers acquisition *and* the body.** That is the whole + point: every critical section on this lock is bounded, so no code path + can wait for it — or hold it — indefinitely. A single stuck operation + would otherwise wedge the table forever, because a stuck holder blocks + every writer *and* the maintenance scheduler skips a kind whose task + never finishes (observed in a soak run: one table stopped reclaiming + versions permanently, 150 versions retained, disk 11x live size, with + no error logged anywhere because nothing failed — it simply never + returned). + + On expiry the body is cancelled, the lock is released, and the timeout + is re-raised as :class:`VectorStoreBusyError` so the cascade worker + treats it as transient and retries instead of marking the row + permanently failed. + """ + started = time.monotonic() + acquired_at: float | None = None + try: + async with asyncio.timeout(budget): + async with self._write_lock(self.table_name): + acquired_at = time.monotonic() + yield + except TimeoutError as exc: + # ``acquired`` is the load-bearing field: it separates "never got + # the lock" (a holder is slow or stuck) from "got it and overran" + # (this operation itself is slow), which is exactly what a soak + # investigation cannot otherwise tell apart. + now = time.monotonic() + logger.warning( + "lancedb_write_lock_deadline_exceeded", + table=self.table_name, + op=op, + budget_seconds=budget, + acquired=acquired_at is not None, + waited_seconds=round((acquired_at or now) - started, 3), + held_seconds=round(now - acquired_at, 3) if acquired_at else 0.0, + ) + raise VectorStoreBusyError( + f"{op} on table {self.table_name!r} exceeded its " + f"{budget:g}s write-lock deadline" + ) from exc + else: + held = time.monotonic() - (acquired_at or started) + if held >= _SLOW_HOLD_LOG_SECONDS: + logger.info( + "lancedb_write_lock_slow_hold", + table=self.table_name, + op=op, + held_seconds=round(held, 3), + waited_seconds=round((acquired_at or started) - started, 3), + ) + @classmethod def _reset_locks_for_tests(cls) -> None: """Test-only: drop the write-lock pool. @@ -191,7 +286,7 @@ class LanceRepoBase[T: BaseLanceTable]: async def add(self, records: Sequence[T]) -> None: """Insert one or more records.""" table = await self._table() - async with self._write_lock(self.table_name): + async with self._locked(_WRITE_TIMEOUT_SECONDS, "add"): await table.add(list(records)) # ── Upsert ───────────────────────────────────────────────────────────── @@ -213,7 +308,7 @@ class LanceRepoBase[T: BaseLanceTable]: updates its existing row. """ table = await self._table() - async with self._write_lock(self.table_name): + async with self._locked(_WRITE_TIMEOUT_SECONDS, "upsert"): await ( table.merge_insert(by) .when_matched_update_all() @@ -312,11 +407,8 @@ class LanceRepoBase[T: BaseLanceTable]: ``part_N`` / index UUID count) — that is ``rebuild_indexes``'s job. """ table = await self._table() - async with self._write_lock(self.table_name): - async with asyncio.timeout(_PRUNE_TIMEOUT_SECONDS): - await table.optimize( - cleanup_older_than=older_than, delete_unverified=False - ) + async with self._locked(_PRUNE_TIMEOUT_SECONDS, "prune"): + 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) if removed: @@ -379,7 +471,7 @@ class LanceRepoBase[T: BaseLanceTable]: - https://docs.rs/lancedb/latest/lancedb/table/struct.OptimizeOptions.html """ table = await self._table() - async with self._write_lock(self.table_name): + async with self._locked(_REBUILD_TIMEOUT_SECONDS, "rebuild_indexes"): for idx in await table.list_indices(): await table.drop_index(idx.name) await self.schema.ensure_fts_indexes(table) @@ -560,7 +652,7 @@ class LanceRepoBase[T: BaseLanceTable]: where: SQL-like predicate scoping the update. """ table = await self._table() - async with self._write_lock(self.table_name): + async with self._locked(_WRITE_TIMEOUT_SECONDS, "update"): await table.update(updates, where=where) # ── Delete ───────────────────────────────────────────────────────────── @@ -568,7 +660,7 @@ class LanceRepoBase[T: BaseLanceTable]: async def delete(self, predicate: str) -> None: """Delete rows matching a SQL-like predicate.""" table = await self._table() - async with self._write_lock(self.table_name): + async with self._locked(_WRITE_TIMEOUT_SECONDS, "delete"): await table.delete(predicate) async def delete_by_md_path(self, md_path: str) -> int: @@ -579,7 +671,7 @@ class LanceRepoBase[T: BaseLanceTable]: Single quotes in ``md_path`` are doubled defensively. """ table = await self._table() - async with self._write_lock(self.table_name): + async with self._locked(_WRITE_TIMEOUT_SECONDS, "delete_by_md_path"): result = await table.delete(f"md_path = '{_q(md_path)}'") return int(result.num_deleted_rows) diff --git a/src/everos/entrypoints/cli/commands/cascade.py b/src/everos/entrypoints/cli/commands/cascade.py index c5aa827..8710f94 100644 --- a/src/everos/entrypoints/cli/commands/cascade.py +++ b/src/everos/entrypoints/cli/commands/cascade.py @@ -58,7 +58,11 @@ from everos.infra.persistence.sqlite import ( get_engine, md_change_state_repo, ) -from everos.memory.cascade import CascadeOrchestrator, match_kind +from everos.memory.cascade import ( + CascadeOrchestrator, + match_kind, + ome_lock_is_free, +) logger = get_logger(__name__) @@ -131,17 +135,30 @@ _VERBOSE_OPTION_HELP = ( @asynccontextmanager -async def _runtime(*, verify: bool = True): # type: ignore[no-untyped-def] +async def _runtime( # type: ignore[no-untyped-def] + *, verify: bool = True, ensure: bool = True +): """Stand up sqlite + lancedb the same way the API lifespan would. - The CLI piggybacks on the same singletons as the running daemon - (lazy + process-wide), so if a server happens to be running on - the same memory root, both share state correctly. + The CLI uses the same lazy, process-wide singletons the API lifespan + does. They are **per-process**: a running daemon has its own + connection and table-handle cache, so read/write traffic interleaves + safely, but a change to the table *set* made here (drop / recreate) + is invisible to the daemon's cached handles — which is why + ``rebuild`` refuses to run while a server holds the OME lock. ``verify=False`` skips :func:`verify_business_schemas` — required by ``cascade rebuild``, whose whole purpose is to recover from a table whose schema *has* drifted; running the guard there would abort startup before the rebuild could fix it (chicken-and-egg). + + ``ensure=False`` additionally skips :func:`ensure_business_indexes`. + That call runs the schema / FTS migrations against the **existing** + tables, and on the corruption classes rebuild exists to repair (a + missing column, an un-alterable type) it raises before the drop can + happen — the recovery path dying on the damage it was invoked to fix. + Rebuild recreates the tables and their indexes itself after dropping, + so skipping the pre-drop pass loses nothing. """ engine = get_engine() async with engine.begin() as conn: @@ -149,7 +166,8 @@ async def _runtime(*, verify: bool = True): # type: ignore[no-untyped-def] await get_connection() if verify: await verify_business_schemas() - await ensure_business_indexes() + if ensure: + await ensure_business_indexes() try: yield finally: @@ -435,9 +453,15 @@ def rebuild( ) -> None: """Rebuild the LanceDB index from markdown (recover from schema drift). + **Stop the ``everos server`` first** — this is the one cascade command + that is not safe alongside a live daemon. It drops and recreates the + tables, and the daemon's cached table handles would keep writing to + the dropped dataset; the command refuses to start while a server holds + the OME lock. + Drops every business LanceDB table and re-indexes all md from scratch. Markdown is the source of truth, so no memory content is - lost, and this is the **safe** recovery from a drifted / corrupt + lost, and this is the safe recovery from a drifted / corrupt index (e.g. the ``verify_business_schemas`` startup failure): - unlike ``rm -rf ~/.everos/.index/lancedb``, it re-populates @@ -448,16 +472,39 @@ def rebuild( is NOT rebuildable from md — notably ``unprocessed_buffer`` (messages received but not yet extracted). """ + if not ome_lock_is_free(): + typer.echo( + "error: a server (or another exclusive CLI phase) is running on " + "this memory root.\n" + " cascade rebuild drops and recreates the LanceDB tables; a live " + "daemon holds cached\n" + " table handles and would keep writing to the dropped dataset. " + "Stop `everos server`\n" + " first, then re-run.", + err=True, + ) + raise typer.Exit(code=3) if not yes: typer.confirm( - "Drop all LanceDB business tables and re-index from markdown?", + "Drop all LanceDB business tables and re-index from markdown? " + "(requires the server to be stopped)", abort=True, ) async def _run() -> None: # verify=False: the on-disk schema may be exactly what we're here # to fix; the startup guard would abort before we could rebuild. - async with _runtime(verify=False): + # ensure=False: the pre-drop migration pass would raise on exactly + # the damage we are here to repair (see _runtime). + async with _runtime(verify=False, ensure=False): + # Reset the queue FIRST so every crash window converges on + # "queue pending → next scan re-indexes". Doing it after the + # drop leaves a window where a crash yields empty tables with + # a fully-`done` queue: nothing re-indexes, the schema guard + # passes, and the deployment comes up silently empty — the + # exact state this command exists to avoid. + cleared = await md_change_state_repo.reset_all() + typer.echo(f"reset {cleared} cascade queue row(s)") dropped = await drop_business_tables() typer.echo( f"dropped {len(dropped)} LanceDB table(s): " @@ -465,15 +512,21 @@ def rebuild( ) # Recreate the tables (current schema) + FTS indexes. await ensure_business_indexes() - # Clear the work queue so every md file re-enqueues as `added`. - cleared = await md_change_state_repo.reset_all() - typer.echo(f"reset {cleared} cascade queue row(s)") # Re-scan + drain: re-embed and re-insert every md entry. orchestrator = _build_orchestrator() processed = await orchestrator.sync_once() typer.echo(f"rebuild complete — re-indexed {processed} md file(s)") - asyncio.run(_run()) + try: + asyncio.run(_run()) + except KeyboardInterrupt: + typer.echo( + "\ninterrupted — the cascade queue is reset, so re-running " + "`everos cascade rebuild` (or starting the server) resumes the " + "re-index from where it stopped.", + err=True, + ) + raise typer.Exit(code=130) from None # ── helpers ────────────────────────────────────────────────────────────── diff --git a/src/everos/infra/persistence/lancedb/__init__.py b/src/everos/infra/persistence/lancedb/__init__.py index 6d23bb1..78ea873 100644 --- a/src/everos/infra/persistence/lancedb/__init__.py +++ b/src/everos/infra/persistence/lancedb/__init__.py @@ -209,8 +209,10 @@ async def migrate_table_schemas() -> None: continuing in a half-migrated state — cascade would otherwise write ``vector=None`` (soft-dependency embedding) into a still NOT-NULL column and every row would silently fail. Recovery escalates from - a plain restart (transient hiccup) to wiping the LanceDB index - directory (rebuildable from md, the SoT). + a plain restart (transient hiccup) to ``everos cascade rebuild``, + which re-indexes from md (the SoT) *and* re-enqueues every file — + unlike deleting the index dir, which leaves the queue ``done`` and + yields an empty index. """ logger = get_logger(__name__) memory_root = MemoryRoot.resolve() @@ -254,9 +256,12 @@ async def migrate_table_schemas() -> None: f"corrupted index. Recovery, in order of least- to " f"most-destructive: (1) restart the process; a transient " f"filesystem or LanceDB-side hiccup may resolve. (2) If " - f"the error persists, wipe the index directory " - f"`{memory_root.lancedb_dir}` and restart — cascade will " - f"re-index from source markdown." + f"the error persists, run `everos cascade rebuild` (with " + f"the server stopped) — it re-indexes from source markdown " + f"and preserves un-extracted buffered messages. Do NOT " + f"just delete `{memory_root.lancedb_dir}`: that leaves the " + f"cascade queue marked done, so nothing re-indexes and the " + f"index comes back empty." ) marker.parent.mkdir(parents=True, exist_ok=True) @@ -341,13 +346,12 @@ async def verify_business_schemas() -> None: f"LanceDB table {schema.TABLE_NAME!r} schema drift: " f"missing={sorted(missing)}, extra={sorted(extra)}, " f"type_drift={type_drift}.\n" - "Recovery, escalating:\n" - " 1. Restart the server — an in-flight migration may " - "still be finishing (harmless if this is your first " - "restart after upgrading EverOS).\n" - " 2. If restart doesn't clear it, recover with " - "`everos cascade rebuild` (drops + re-indexes from md, " - "preserving un-extracted buffered messages)." + "Recover with `everos cascade rebuild` (stop the server " + "first): it drops and re-indexes from md, preserving " + "un-extracted buffered messages. Restarting will not " + "clear this — the startup migrations only alter column " + "nullability, never a column's name or type, so a " + "name/type drift never resolves on its own." ) diff --git a/src/everos/memory/cascade/__init__.py b/src/everos/memory/cascade/__init__.py index e99bf97..7ff6b1b 100644 --- a/src/everos/memory/cascade/__init__.py +++ b/src/everos/memory/cascade/__init__.py @@ -21,6 +21,7 @@ Public surface — what lifespan providers / CLI commands import: from ._backfill import BackfillPhase as BackfillPhase from ._backfill import BackfillPresenter as BackfillPresenter from ._backfill import NullBackfillPresenter as NullBackfillPresenter +from ._backfill import ome_lock_is_free as ome_lock_is_free from .orchestrator import CascadeConfig as CascadeConfig from .orchestrator import CascadeHealth as CascadeHealth from .orchestrator import CascadeOrchestrator as CascadeOrchestrator @@ -38,4 +39,5 @@ __all__ = [ "KindSpec", "NullBackfillPresenter", "match_kind", + "ome_lock_is_free", ] diff --git a/src/everos/memory/cascade/_backfill.py b/src/everos/memory/cascade/_backfill.py index 1283207..767ee6d 100644 --- a/src/everos/memory/cascade/_backfill.py +++ b/src/everos/memory/cascade/_backfill.py @@ -64,6 +64,9 @@ from everos.infra.persistence.lancedb import ( ) from everos.infra.persistence.markdown import AgentSkillFrontmatter from everos.infra.persistence.sqlite import cluster_repo, get_engine +from everos.memory.cascade.worker import ( + DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS, +) from everos.memory.events import ( AgentCaseExtracted, EpisodeExtracted, @@ -747,11 +750,18 @@ async def _backfill_table( # optimize() compacts the per-row-update fragments; prune() # physically reclaims the superseded manifest versions. Split # after the repo API separated them (compact is lock-free, prune - # runs under the write lock). prune(0) reclaims everything now; - # it is cross-process safe (delete_unverified=False), so running - # ``backfill`` alongside a live daemon cannot corrupt the table. + # runs under the write lock) and cross-process safe because prune + # passes delete_unverified=False. + # + # Keep the daemon's retention window rather than reclaiming at + # zero age: the window's job is to outlive an in-flight read (a + # /search holding a version reference), and this runs in a + # separate process where the write lock cannot fence one. Files + # younger than the window are reclaimed by the next daemon prune. await backlog.spec.repo.optimize() - await backlog.spec.repo.prune(dt.timedelta(0)) + await backlog.spec.repo.prune( + dt.timedelta(seconds=DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS) + ) logger.info( "cascade_backfill_table_optimized", table=backlog.table_name, @@ -1035,6 +1045,22 @@ async def _emit_synthetic_events( return emitted +def ome_lock_is_free() -> bool: + """Whether no other process holds the OME jobstore lock. + + ``False`` means a live ``everos server`` (or another exclusive CLI + phase) is running against this memory root. Public entry point for + commands that must not run concurrently with the daemon — notably + ``cascade rebuild``, which drops and recreates the LanceDB tables + under any cached handles a running daemon still holds. + + Same best-effort caveat as :func:`_probe_ome_lock_available`: the lock + can be taken between this probe and the destructive step, so it is a + guard against the common mistake, not a mutual-exclusion primitive. + """ + return _probe_ome_lock_available() + + def _probe_ome_lock_available() -> bool: """Probe whether the OME jobstore file lock is free. diff --git a/src/everos/memory/cascade/worker.py b/src/everos/memory/cascade/worker.py index b10e7db..8926d73 100644 --- a/src/everos/memory/cascade/worker.py +++ b/src/everos/memory/cascade/worker.py @@ -78,10 +78,16 @@ _OPTIMIZE_FAILURE_ALERT_THRESHOLD = 5 """Consecutive **non-benign** ``optimize()`` failures (per kind) before the log escalates ``warning``→``error``, a fallback rebuild is triggered (:meth:`_run_rebuild_once`), and :meth:`CascadeWorker.health` reports the -kind degraded. A benign light-beat commit conflict (lost concurrency -race) is expected under churn, logged at ``debug``, and does **not** +kind degraded. A benign commit conflict (lost concurrency race, either +beat) is expected under churn, logged at ``debug``, and does **not** count — otherwise the streak pins high on a busy table and drowns the -real signal (the disk-bloat failure mode behind lance-format/lance#7653).""" +real signal (the disk-bloat failure mode behind lance-format/lance#7653). + +Prune that stops succeeding is **not** detected here: a lost race is +excluded by design, and an intervening light-beat success resets the +streak anyway. That failure mode belongs to the per-kind prune-staleness +signal (:data:`_PRUNE_STALE_FACTOR`), which fires on the symptom (nothing +reclaimed for 3 cadences) rather than on a particular exception.""" _DRAIN_FAILURE_ALERT_THRESHOLD = 3 """Consecutive :meth:`drain_once` exceptions at or above which :meth:`CascadeWorker.health` reports degraded — the md → LanceDB @@ -243,9 +249,20 @@ class CascadeWorkerHealth: kinds; benign light-beat commit conflicts do not count.""" prune_stale_seconds: float - """Seconds since the most recent successful prune (version cleanup) - across all active kinds, measured from worker start if none has run - yet. ``0`` when there has been no write activity to prune.""" + """Staleness of the **worst** kind — seconds since that kind's last + successful prune (version cleanup), measured from worker start if it + has never pruned. ``0`` when no kind has an optimizer state yet (no + write activity to prune). + + Deliberately the worst kind, not the newest prune across kinds: a + per-kind cleanup that dies (hung lance cleanup, lost commit races) + grows *that table's* index dir unbounded, and every other kind + pruning normally must not mask it.""" + + prune_stale_kind: str | None = None + """The kind :attr:`prune_stale_seconds` belongs to; ``None`` when no + kind has state yet. Named in :meth:`reasons` so an operator knows + which table to look at.""" def reasons(self) -> list[str]: """Operational degradation reasons; empty when healthy. @@ -262,9 +279,11 @@ class CascadeWorkerHealth: if self.optimize_failure_streak >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD: out.append(f"optimize stuck ({self.optimize_failure_streak} in a row)") if self.prune_stale_seconds >= _PRUNE_STALE_SECONDS_ALERT: + kind = self.prune_stale_kind or "unknown" out.append( - f"version cleanup stalled ({int(self.prune_stale_seconds)}s " - "since last prune — disk may grow)" + f"version cleanup stalled for kind '{kind}' " + f"({int(self.prune_stale_seconds)}s since its last prune " + "— that table's index dir may grow)" ) return out @@ -274,9 +293,11 @@ def _is_benign_commit_conflict(exc: BaseException) -> bool: LanceDB surfaces the Rust ``Retryable commit conflict`` as a plain exception whose message carries the phrase — there is no dedicated - class to catch. On the lock-free light beat this is expected under - concurrent writes and benign (the next beat retries), so it is - logged at ``debug`` and does not count toward the failure streak. + class to catch. Either beat can lose the race: the light beat is + lock-free, and the heavy beat's write lock is in-process only, so a + second process (CLI ``cascade sync`` / ``backfill``) can preempt it. + Both are expected under churn and benign (the next beat retries), so + they log at ``debug`` and do not count toward the failure streak. Match only the specific ``commit conflict`` phrase (a substring of the Rust message), not a bare ``retryable`` — the latter appears in the @@ -451,29 +472,42 @@ class CascadeWorker: """ states = self._optimizer_states.values() optimize_failure_streak = max((s.optimize_failures for s in states), default=0) + stale_seconds, stale_kind = self._prune_staleness() return CascadeWorkerHealth( drain_consecutive_failures=self._drain_consecutive_failures, unrecoverable_total=self._unrecoverable_total, optimize_failure_streak=optimize_failure_streak, - prune_stale_seconds=self._prune_stale_seconds(), + prune_stale_seconds=stale_seconds, + prune_stale_kind=stale_kind, ) - def _prune_stale_seconds(self) -> float: - """Seconds since the most recent successful prune across kinds, - measured from worker start when nothing has pruned yet. + def _prune_staleness(self) -> tuple[float, str | None]: + """Staleness of the **worst** kind: ``(seconds, kind)``. - Returns ``0`` before the worker has started (``_started_at == 0``) - or before any kind has registered an optimizer state — no prune - beat has run yet, so there is nothing to be stale about. Once a - beat registers state, staleness is the time since the newest - ``last_prune_at`` (or since start, whichever is later). + Per kind, staleness is the time since its own last successful + prune — or since worker start if it has never pruned — and the + worst (largest) one is reported. Taking the worst rather than the + newest prune across kinds is what makes the signal work on a + multi-kind deployment: one kind whose cleanup dies grows that + table's index dir unbounded, and the ~5 healthy kinds pruning on + schedule must not hide it (that masking was the pre-fix bug). + + Returns ``(0.0, None)`` before the worker has started + (``_started_at == 0``) or before any kind has registered an + optimizer state — no prune beat has run, so nothing is stale yet. """ - states = list(self._optimizer_states.values()) + states = list(self._optimizer_states.items()) if not states or self._started_at == 0.0: - return 0.0 - latest_prune = max(s.last_prune_at for s in states) - baseline = max(latest_prune, self._started_at) - return max(0.0, time.monotonic() - baseline) + return 0.0, None + now = time.monotonic() + worst_seconds = -1.0 + worst_kind: str | None = None + for kind, state in states: + baseline = max(state.last_prune_at, self._started_at) + stale = max(0.0, now - baseline) + if stale > worst_seconds: + worst_seconds, worst_kind = stale, kind + return max(0.0, worst_seconds), worst_kind # ── internals ────────────────────────────────────────────────────────── @@ -714,6 +748,12 @@ class CascadeWorker: now = time.monotonic() should_prune = ( state is None + # 0.0 means "never attempted" — always prune, don't compare clocks. + # ``monotonic()`` is boot-relative, so ``now - 0 >= interval`` is + # false for the first ~cadence of machine/container uptime and the + # catch-up prune would be skipped exactly when a fresh process most + # needs it. + or state.last_prune_attempt_at == 0.0 or (now - state.last_prune_attempt_at) >= self._optimize_prune_interval ) try: @@ -754,9 +794,22 @@ class CascadeWorker: # lost the optimistic-concurrency race against a live writer. # Expected under churn, self-heals next beat — log at debug and # do NOT count it toward the streak (which would otherwise pin - # high on a busy table) or trigger a fallback rebuild. The heavy - # beat runs under the write lock, so it can't hit this benignly. - if not should_prune and _is_benign_commit_conflict(exc): + # high on a busy table) or trigger a fallback rebuild. + # + # This applies to the HEAVY beat too: the per-table write lock is + # in-process only (see LanceRepoBase.prune), so a second process + # — a long `cascade backfill`, a `cascade sync` — can still + # preempt prune's Rewrite commit. Counting those as real failures + # let ~25min of cross-process churn reach the threshold and fire a + # spurious fallback rebuild, which drops every index before + # recreating it; if the rebuild lost the race too, its failure was + # swallowed as a warning and the table sat with no FTS index (all + # `/search` on that kind 500s) until the next 12h sweep. A prune + # that genuinely stops succeeding is caught by the prune-staleness + # health signal instead (per-kind, see _prune_staleness) — that is + # the right detector for it, and unlike a rebuild it does not + # destroy indexes to "fix" a lost race. + if _is_benign_commit_conflict(exc): logger.debug( "cascade_lancedb_optimize_conflict", kind=kind, diff --git a/src/everos/memory/search/manager.py b/src/everos/memory/search/manager.py index 08fdf8e..1ae5bfd 100644 --- a/src/everos/memory/search/manager.py +++ b/src/everos/memory/search/manager.py @@ -41,7 +41,7 @@ from everos.component.rerank import get_rerank_capability from everos.component.utils.datetime import to_display_tz from everos.config import load_settings from everos.core.context import resolve_request_id -from everos.core.errors import ProviderNotConfiguredError +from everos.core.errors import ConfigurationError, ProviderNotConfiguredError from everos.core.observability.logging import get_logger from everos.core.observability.tracing import ( capture_input, @@ -716,7 +716,26 @@ class SearchManager: async def _embed_query(self, query: str) -> list[float]: if self._embedding is None: return [] - return await self._embedding.embed(query) + vector = await self._embedding.embed(query) + expected = getattr(self._embedding, "dim", None) + if expected and len(vector) != expected: + # Fail here, not inside LanceDB. A mismatched query vector reaches + # the engine as an opaque `ValueError: Invalid input, No vector + # column found to match...` after the query has already been set + # up — measured at 13-14s per request in a soak run, versus + # microseconds here, and it surfaces as an unhandled 500 with a + # ~6k-line traceback instead of a named error. + # + # ConfigurationError (not InvalidInputError): the caller only ever + # sends query *text*; the vector is produced by our own provider, + # so a width that disagrees with the provider's declared ``dim`` + # is a server-side configuration or provider-implementation fault. + raise ConfigurationError( + f"embedding provider returned a {len(vector)}-dimension query " + f"vector but declares dim={expected}; the vector index cannot " + f"be searched with a mismatched width" + ) + return vector # ── Limits / filters ──────────────────────────────────────────── diff --git a/tests/integration/test_cascade_cli_integration.py b/tests/integration/test_cascade_cli_integration.py index 8a91dd6..22511dd 100644 --- a/tests/integration/test_cascade_cli_integration.py +++ b/tests/integration/test_cascade_cli_integration.py @@ -286,6 +286,33 @@ def test_rebuild_recovers_drifted_index_and_reindexes( assert asyncio.run(_atomic_fact_row_count()) == 2 +def test_rebuild_refuses_to_run_while_a_server_holds_the_lock( + cli_runtime: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """``rebuild`` must refuse when a daemon is running on this memory root. + + It drops and recreates the LanceDB tables; a live daemon holds cached + table handles and would keep writing to the dropped dataset, leaving a + corrupted rebuild plus a permanent-failure backlog. Detection reuses the + OME jobstore lock that ``backfill`` already gates on, and the exit code + matches backfill's ``3`` (SERVER_RUNNING). + """ + monkeypatch.setattr(cascade_mod, "ome_lock_is_free", lambda: False) + + result = CliRunner().invoke(cascade_mod.app, ["rebuild", "--yes"]) + + assert result.exit_code == 3, result.output + # The explanation goes to stderr (click 8.2 keeps the streams separate). + assert "server" in result.stderr.lower() + assert "stop `everos server`" in result.stderr.lower() + # And it must bail out BEFORE touching anything — none of the step + # progress lines may appear. + combined = result.output + result.stderr + assert "LanceDB table(s)" not in combined + assert "cascade queue row(s)" not in combined + assert "rebuild complete" not in combined + + # Reduce false negatives on date drift. def test_resolve_relative_via_command_arg(cli_runtime: Path) -> None: """An absolute path under the root works through ``cascade sync ``.""" diff --git a/tests/integration/test_infra/test_lancedb_schema_migration.py b/tests/integration/test_infra/test_lancedb_schema_migration.py index 15aa98c..2b93833 100644 --- a/tests/integration/test_infra/test_lancedb_schema_migration.py +++ b/tests/integration/test_infra/test_lancedb_schema_migration.py @@ -107,7 +107,8 @@ async def test_migrate_table_schemas_raises_on_alter_failure( """Any table's failing ``alter_columns`` must fail startup loudly: :class:`LanceDBMigrationError` is raised, the version marker stays unwritten, and the message names the failed table plus escalating - recovery hints (restart, then wipe). Cascade would otherwise write + recovery hints (restart, then ``cascade rebuild``). Cascade would + otherwise write NULL vectors into a still-NOT-NULL column and silently drop every row.""" for schema in BUSINESS_SCHEMAS_WITH_VECTOR: @@ -127,11 +128,13 @@ async def test_migrate_table_schemas_raises_on_alter_failure( assert Episode.TABLE_NAME in message lancedb_dir = tmp_path / ".index" / "lancedb" assert str(lancedb_dir) in message - # Escalating recovery: restart first, wipe second. + # Escalating recovery: restart first, `cascade rebuild` second. Never + # "delete the index dir" — that leaves the queue done and the index empty. restart_idx = message.find("restart the process") - wipe_idx = message.find("wipe the index directory") - assert restart_idx != -1 and wipe_idx != -1 - assert restart_idx < wipe_idx + rebuild_idx = message.find("everos cascade rebuild") + assert restart_idx != -1 and rebuild_idx != -1 + assert restart_idx < rebuild_idx + assert "wipe the index directory" not in message marker = lancedb_dir / ".table_schema_version" assert not marker.exists() diff --git a/tests/integration/test_tiers/test_tier3_full.py b/tests/integration/test_tiers/test_tier3_full.py index 8e6d76b..41dc0ec 100644 --- a/tests/integration/test_tiers/test_tier3_full.py +++ b/tests/integration/test_tiers/test_tier3_full.py @@ -195,3 +195,13 @@ async def test_health_reports_tier3_capabilities(tier3_runtime: AsyncClient) -> "knowledge", } ) + + # Cascade readiness block is present on a real app that ran the cascade + # lifespan. This pins the wiring, which is stringly-typed on both ends + # (lifespan provider name → lifespan_data key); a rename would silently + # drop the block from production /health while the route's own unit tests, + # which hand-build lifespan_data, stayed green. + cascade = body["cascade"] + assert cascade is not None, "cascade block missing — lifespan_data wiring broke" + assert cascade["healthy"] is True + assert cascade["reasons"] == [] diff --git a/tests/unit/test_core/test_observability/test_logging_factory.py b/tests/unit/test_core/test_observability/test_logging_factory.py index 9094e3c..d7cf854 100644 --- a/tests/unit/test_core/test_observability/test_logging_factory.py +++ b/tests/unit/test_core/test_observability/test_logging_factory.py @@ -109,3 +109,31 @@ def test_get_logger_with_same_name_returns_equivalent( assert isinstance(a, structlog.stdlib.BoundLogger | structlog.BoundLoggerBase) or ( hasattr(a, "info") and hasattr(b, "info") ) + + +def _console_renderer_exception_formatter(): + """Pull the exception formatter out of the configured ProcessorFormatter.""" + import logging + + configure_logging(level="INFO") + for handler in logging.getLogger().handlers: + fmt = getattr(handler, "formatter", None) + procs = getattr(fmt, "processors", None) or () + for proc in procs: + if isinstance(proc, structlog.dev.ConsoleRenderer): + return proc._exception_formatter + return None + + +def test_traceback_rendering_omits_locals_and_caps_frames() -> None: + formatter = _console_renderer_exception_formatter() + assert formatter is not None, "ConsoleRenderer not found on the root handler" + assert isinstance(formatter, structlog.dev.RichTracebackFormatter) + assert formatter.show_locals is False, ( + "locals rendering is what blew a soak run's logs up to 85MB and can " + "print request payloads into them" + ) + assert formatter.max_frames <= 30, ( + f"max_frames={formatter.max_frames} — deep async stacks render " + "thousands of lines per exception" + ) 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 0f3bdf4..06f7ade 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 @@ -744,6 +744,129 @@ async def test_prune_holds_write_lock_and_is_cross_process_safe( assert captured["cleanup_older_than"] == dt.timedelta(seconds=42) +async def test_prune_times_out_and_releases_the_write_lock( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A hung lance cleanup must not hold the per-table write lock forever. + + On expiry the body is cancelled, the lock is released so writers on that + table are not wedged, and the timeout surfaces as + :class:`VectorStoreBusyError` — deliberately a *retryable* error, so the + cascade worker retries the row instead of marking it permanently failed. + """ + from everos.core.errors import ExternalServiceError, VectorStoreBusyError + from everos.core.persistence.lancedb import repository as repo_mod + + monkeypatch.setattr(repo_mod, "_PRUNE_TIMEOUT_SECONDS", 0.05) + + class _HangingTable: + async def optimize(self, **_kw): # type: ignore[no-untyped-def] + await asyncio.sleep(30) # never returns within the timeout + + async def uri(self) -> str: + return str(tmp_path) + + repo = _NoteRepo(table=_HangingTable()) # type: ignore[arg-type] + with pytest.raises(VectorStoreBusyError) as excinfo: + await repo.prune(dt.timedelta(seconds=60)) + + assert isinstance(excinfo.value, ExternalServiceError), ( + "must be retryable — under VectorStoreError the worker would mark the " + "row permanently failed and need a manual `cascade fix`" + ) + assert not repo._write_lock(repo.table_name).locked(), ( + "the write lock must be released after the timeout, otherwise a hung " + "cleanup wedges every writer on this table" + ) + # And the lock is genuinely reusable afterwards. + async with repo._write_lock(repo.table_name): + pass + + +async def test_waiting_for_a_stuck_holder_also_times_out( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """**No path may wait for this lock indefinitely.** + + The deadline covers acquisition, not just the body. Without that, one + operation that hangs while holding the lock wedges the table for good: + every writer blocks on acquire, and the maintenance scheduler skips a kind + whose task never finishes, so that table stops reclaiming versions forever + (observed in a soak run — 150 versions retained, disk 11x live size, and + *no* error logged anywhere, because nothing failed; it simply never + returned). + """ + from everos.core.errors import VectorStoreBusyError + from everos.core.persistence.lancedb import repository as repo_mod + + monkeypatch.setattr(repo_mod, "_WRITE_TIMEOUT_SECONDS", 0.05) + + class _NoopTable: + async def add(self, _records): # type: ignore[no-untyped-def] + return None + + repo = _NoteRepo(table=_NoopTable()) # type: ignore[arg-type] + + # Simulate a holder that never gives the lock back. + lock = repo._write_lock(repo.table_name) + await lock.acquire() + try: + with pytest.raises(VectorStoreBusyError): + await repo.add([_row(owner="u1", entry="n1")]) + finally: + lock.release() + + # Once the stuck holder is gone, the table works again — the timeout + # bounded the wait without breaking anything. + await repo.add([_row(owner="u1", entry="n2")]) + + +def test_write_budgets_are_sized_from_measurements_not_guesses() -> None: + """Write budgets must stay in the tens of seconds, not hundreds. + + The budget doubles as the detection latency for a wedged table: a stuck + holder is invisible until its deadline expires. Measured write durations + are 2-25ms (worst observed 63ms across 10k-100k row tables and 50-500 row + batches), so tens of seconds is already ~10^3 headroom. A budget in the + hundreds of seconds would mean minutes of blocked writers before anything + is reported, which is what this whole change exists to prevent. + """ + from everos.core.persistence.lancedb.repository import ( + _PRUNE_TIMEOUT_SECONDS, + _REBUILD_TIMEOUT_SECONDS, + _WRITE_TIMEOUT_SECONDS, + ) + + assert 5.0 <= _WRITE_TIMEOUT_SECONDS <= 30.0, ( + "row writes are millisecond operations; a budget outside this range is " + "either too tight to survive a contended lock or too slack to detect a " + "wedged table promptly" + ) + # Rebuild is the one legitimately slow section, so it gets more — but the + # ordering must hold: a rebuild budget below prune's would make the slowest + # operation the most eagerly killed. + assert _REBUILD_TIMEOUT_SECONDS > _PRUNE_TIMEOUT_SECONDS > _WRITE_TIMEOUT_SECONDS + + +def test_prune_timeout_is_well_below_the_prune_cadence() -> None: + """The timeout is a hang-catcher, not a bound on normal runtime. + + A real cleanup is milliseconds even on a heavily churned table, so the + value only matters when lance hangs — and then it must expire well before + the next heavy beat is due, or the lock is held for most of every cadence + (the ~97%-duty-cycle bug: timeout == cadence, so a hung prune was retried + ~immediately after each expiry). + """ + from everos.core.persistence.lancedb.repository import _PRUNE_TIMEOUT_SECONDS + from everos.memory.cascade.worker import ( + DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS, + ) + + assert _PRUNE_TIMEOUT_SECONDS < DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS / 2, ( + "prune timeout must leave a real write window before the next beat" + ) + + async def test_prune_removes_empty_index_dir_husks(tmp_path: Path) -> None: """After cleanup, ``prune`` removes the empty ``_indices//`` dirs that ``cleanup_older_than`` leaves behind, but keeps non-empty ones.""" diff --git a/tests/unit/test_infra/test_lancedb/test_migration_lock_wiring.py b/tests/unit/test_infra/test_lancedb/test_migration_lock_wiring.py index 3cee920..0f6ec60 100644 --- a/tests/unit/test_infra/test_lancedb/test_migration_lock_wiring.py +++ b/tests/unit/test_infra/test_lancedb/test_migration_lock_wiring.py @@ -194,7 +194,7 @@ async def test_migrate_table_schemas_still_raises_on_alter_failure( ) -> None: """Fail-loud regression: a genuine ``alter_columns`` failure still raises :class:`LanceDBMigrationError`, and the message lists the - escalating recovery steps in order (restart, then wipe).""" + escalating recovery steps in order (restart, then ``cascade rebuild``).""" lock = _TrackingLock() monkeypatch.setattr(lancedb_infra, "memory_root_lock", lock) @@ -217,13 +217,15 @@ async def test_migrate_table_schemas_still_raises_on_alter_failure( message = str(excinfo.value) assert Episode.TABLE_NAME in message restart_idx = message.find("restart the process") - wipe_idx = message.find("wipe the index directory") - # Both hints present and restart comes before wipe (escalating). + rebuild_idx = message.find("everos cascade rebuild") + # Both hints present, restart first (escalating least- to most-destructive). assert restart_idx != -1 - assert wipe_idx != -1 - assert restart_idx < wipe_idx - # Never mention destructive `rm -rf` before the softer step. - assert "restart" in message + assert rebuild_idx != -1 + assert restart_idx < rebuild_idx + # The recovery must NOT be "delete the index dir": that leaves the cascade + # queue marked done, so nothing re-indexes and the index comes back empty. + assert "wipe the index directory" not in message + assert "Do NOT just delete" in message # Marker must not be written on failure. marker = tmp_path / ".index" / "lancedb" / ".table_schema_version" diff --git a/tests/unit/test_memory/test_cascade/test_backfill_optimize.py b/tests/unit/test_memory/test_cascade/test_backfill_optimize.py index 299efc4..e9d44e8 100644 --- a/tests/unit/test_memory/test_cascade/test_backfill_optimize.py +++ b/tests/unit/test_memory/test_cascade/test_backfill_optimize.py @@ -41,6 +41,9 @@ from everos.memory.cascade._backfill import ( _TableBacklog, _TableSpec, ) +from everos.memory.cascade.worker import ( + DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS, +) @pytest.fixture(autouse=True) @@ -167,11 +170,16 @@ async def test_backfill_table_calls_optimize_when_rows_processed( assert result.rows_failed == 0 assert len(repo.update_calls) == 3 assert repo.optimize_calls == 1 - # Compact then reclaim: prune fires once with a zero retention (reclaim - # everything now) — this is what the removed ``cleanup_older_than`` kwarg - # used to do inline (review P0-1). + # Compact then reclaim: prune fires once — this is what the removed + # ``cleanup_older_than`` kwarg used to do inline (review P0-1). assert repo.prune_calls == 1 - assert repo.last_prune_older_than == dt.timedelta(0) + # It must pass the daemon's retention window, NOT zero: backfill runs in a + # separate process, so the in-process write lock cannot fence a daemon + # /search that still holds a reference to a just-superseded version. + # Reclaiming at zero age can delete files out from under that read. + assert repo.last_prune_older_than == dt.timedelta( + seconds=DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS + ) # Happy path must not fire the failure log. assert "cascade_backfill_table_optimize_failed" not in caplog.text diff --git a/tests/unit/test_memory/test_cascade/test_worker.py b/tests/unit/test_memory/test_cascade/test_worker.py index 7b2bdf7..2fb1c54 100644 --- a/tests/unit/test_memory/test_cascade/test_worker.py +++ b/tests/unit/test_memory/test_cascade/test_worker.py @@ -637,6 +637,48 @@ async def test_failed_prune_backs_off_a_cadence_and_keeps_health_signal( assert len(fake.optimize_calls) == 1, "second beat backs off to the light path" +async def test_prune_recurs_once_per_cadence_across_light_beats( + patched_repo: _FakeRepo, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A heavy (prune) beat must fire again after a full cadence, no matter + how many light beats ran in between. + + The attempt clock advances only on the heavy path. If a light beat also + pushed it forward, frequent light beats would keep resetting the cadence + and prune would run exactly once per process lifetime — version cleanup + silently stops and the index dir grows unbounded, which is the incident + this whole change exists to prevent. + """ + from everos.memory.cascade import worker as wmod + + clock = {"t": 1_000.0} + monkeypatch.setattr(wmod.time, "monotonic", lambda: clock["t"]) + fake = _FakeLanceRepo() + w = CascadeWorker( + {"episode": _OkHandlerWithRepo(fake)}, + retry_backoff_seconds=0, + optimize_min_interval_seconds=0.0, + optimize_prune_interval_seconds=10.0, + ) + + w._schedule_optimize("episode") + await w._flush_optimizers() + assert len(fake.prune_calls) == 1, "first beat prunes (never pruned yet)" + + # 11 light-ish beats, one simulated second apart — they cross the cadence. + for _ in range(11): + clock["t"] += 1.0 + w._schedule_optimize("episode") + await w._flush_optimizers() + + assert len(fake.prune_calls) == 2, ( + "prune must recur one cadence after the last prune ATTEMPT; light " + "beats in between must not push the cadence forward" + ) + assert len(fake.optimize_calls) == 10, "the other beats took the light path" + + # ── Rebuild scheduler tests ──────────────────────────────────────────────── @@ -840,18 +882,60 @@ def test_worker_health_dataclass_thresholds() -> None: assert len(reasons) == 2 # drain + prune, not the sub-threshold optimize -def test_worker_health_idle_is_not_stale() -> None: +def test_worker_health_idle_is_not_stale(monkeypatch: pytest.MonkeyPatch) -> None: """A worker with no optimize activity is never prune-stale (nothing to reclaim), even if it started long ago.""" from everos.memory.cascade import worker as wmod + # Freeze the clock: a bare `monotonic() - N` goes negative on a runner + # whose uptime is below N, making the "started long ago" premise fiction. + now = 10_000.0 + monkeypatch.setattr(wmod.time, "monotonic", lambda: now) w = CascadeWorker({"episode": _OkHandlerWithRepo(_FakeLanceRepo())}) - w._started_at = time.monotonic() - (wmod._PRUNE_STALE_SECONDS_ALERT + 1000) + w._started_at = now - (wmod._PRUNE_STALE_SECONDS_ALERT + 1000) h = w.health() assert h.prune_stale_seconds == 0.0 + assert h.prune_stale_kind is None assert h.reasons() == [] +def test_worker_health_reports_worst_kind_not_newest_prune( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Staleness is the WORST kind's, and ``reasons`` names it. + + Production registers several lance-backed kinds. Reporting the newest + prune across kinds let one healthy kind mask a kind whose cleanup had + died — the dead kind's index dir grows unbounded while ``/health`` stays + green, which is the incident this signal exists to catch. + """ + from everos.memory.cascade import worker as wmod + + now = 100_000.0 + monkeypatch.setattr(wmod.time, "monotonic", lambda: now) + w = CascadeWorker( + { + "episode": _OkHandlerWithRepo(_FakeLanceRepo()), + "atomic_fact": _OkHandlerWithRepo(_FakeLanceRepo()), + } + ) + w._started_at = now - 10_000.0 + # atomic_fact pruned just now; episode has not pruned in 3x the threshold. + fresh = wmod._KindOptimizerState() + fresh.last_prune_at = now - 10.0 + dead = wmod._KindOptimizerState() + dead.last_prune_at = now - 3 * wmod._PRUNE_STALE_SECONDS_ALERT + w._optimizer_states["atomic_fact"] = fresh + w._optimizer_states["episode"] = dead + + h = w.health() + assert h.prune_stale_kind == "episode", "must report the worst kind" + assert h.prune_stale_seconds >= wmod._PRUNE_STALE_SECONDS_ALERT + reasons = h.reasons() + assert any("cleanup stalled" in r for r in reasons) + assert any("episode" in r for r in reasons), "operator needs the kind named" + + def test_worker_health_reports_prune_staleness(monkeypatch: pytest.MonkeyPatch) -> None: """An active kind that has never successfully pruned since start goes stale once past the alert threshold.""" @@ -931,3 +1015,88 @@ async def test_light_beat_commit_conflict_is_debug_and_uncounted( assert "cascade_lancedb_optimize_failed" not in events assert "cascade_lancedb_optimize_fallback_rebuild" not in events assert "error" not in levels and "warning" not in levels + + +async def test_heavy_beat_commit_conflict_is_also_benign( + patched_repo: _FakeRepo, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A HEAVY-beat (prune) commit conflict is benign too. + + The per-table write lock is in-process only, so a second process (a long + ``cascade backfill``, a ``cascade sync``) can preempt prune's Rewrite + commit. Counting those as real failures let sustained cross-process churn + reach the alert threshold and fire a spurious fallback rebuild, which + drops every index before recreating it — leaving the table without an FTS + index (and `/search` 500ing on that kind) if the rebuild lost the race + too. A prune that genuinely stops succeeding is caught by the per-kind + prune-staleness signal instead. + """ + from everos.memory.cascade import worker as wmod + + class _ConflictRepo(_FakeLanceRepo): + """Records which beat was attempted, then loses the commit race.""" + + def __init__(self) -> None: + super().__init__() + self.attempts: list[str] = [] + + async def optimize(self) -> None: + self.attempts.append("optimize") + raise RuntimeError("Retryable commit conflict for version 215") + + async def prune(self, older_than: dt.timedelta) -> None: + self.attempts.append("prune") + raise RuntimeError("Retryable commit conflict for version 215") + + # Freeze the clock: which beat runs must not depend on the runner's uptime + # (`monotonic()` is boot-relative — a fresh CI runner reads ~100s). + now = 10_000.0 + monkeypatch.setattr(wmod.time, "monotonic", lambda: now) + repo = _ConflictRepo() + w = CascadeWorker( + {"episode": _OkHandlerWithRepo(repo)}, + retry_backoff_seconds=0, + optimize_prune_interval_seconds=10.0, + ) + state = wmod._KindOptimizerState() # last_prune_attempt_at=0 → HEAVY beat + w._optimizer_states["episode"] = state + + await w._run_optimize_once("episode") + + assert repo.attempts == ["prune"], "must have taken the heavy (prune) path" + assert state.optimize_failures == 0, ( + "a heavy-beat commit conflict is a lost cross-process race, not a " + "failure — counting it re-arms the spurious fallback rebuild" + ) + + +async def test_non_conflict_failure_counts_even_when_message_says_retryable( + patched_repo: _FakeRepo, +) -> None: + """The benign filter must match ONLY ``commit conflict``. + + Widening it to a bare ``retryable`` substring would swallow unrelated + recoverable errors — an ``ExternalServiceError`` repr carries + ``retryable=True`` — so a genuinely stuck optimize would log at debug + forever: no streak, no escalation, no fallback rebuild, health green. + """ + from everos.memory.cascade import worker as wmod + + repo = _OptimizeFailingRepo( + error=RuntimeError("ExternalServiceError(provider='x', retryable=True): boom") + ) + w = CascadeWorker( + {"episode": _OkHandlerWithRepo(repo)}, + retry_backoff_seconds=0, + ) + state = wmod._KindOptimizerState() + state.last_prune_attempt_at = time.monotonic() # light beat + w._optimizer_states["episode"] = state + + await w._run_optimize_once("episode") + + assert state.optimize_failures == 1, ( + "an error whose message merely contains 'retryable' is NOT a commit " + "conflict and must count toward the streak" + ) diff --git a/tests/unit/test_memory/test_search/test_manager.py b/tests/unit/test_memory/test_search/test_manager.py index 407059d..73dd666 100644 --- a/tests/unit/test_memory/test_search/test_manager.py +++ b/tests/unit/test_memory/test_search/test_manager.py @@ -418,6 +418,34 @@ def _atomic_fact_row(fid: str, *, parent_id: str, score: float) -> Candidate: # ── VECTOR (MaxSim atomic) ──────────────────────────────────────────── +async def test_embed_query_rejects_a_width_that_disagrees_with_dim() -> None: + """A provider returning a width other than its declared ``dim`` must fail + at the embed step, not inside LanceDB. + + Sending a mismatched query vector into the engine surfaces as an opaque + ``ValueError: Invalid input, No vector column found to match…`` only after + the query has been set up — 13-14s per request in a soak run, as an + unhandled 500 with a ~6k-line traceback. Here it is microseconds and a + named error. ``ConfigurationError`` (server-side) rather than + ``InvalidInputError``: callers only ever send query *text*, so the width is + entirely our provider's doing. + """ + from everos.core.errors import ConfigurationError + + class _LyingEmbedding(_StubEmbedding): + async def embed(self, text: str) -> list[float]: + return [0.0] * (self.dim // 2) # declares dim, returns half of it + + mgr = _build_manager(embedding=_LyingEmbedding(dim=8)) + + with pytest.raises(ConfigurationError) as excinfo: + await mgr._embed_query("anything") + + msg = str(excinfo.value) + assert "4-dimension" in msg, msg + assert "dim=8" in msg, msg + + async def test_vector_method_requires_embedding( monkeypatch: pytest.MonkeyPatch, ) -> None: