xiaoxue_admin
/
EverOS
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(cascade): per-kind prune staleness + rebuild safety (#384) 

* fix(cascade): per-kind prune staleness + rebuild safety

Adversarial review of the review-response fixes found five real defects,
all in code this PR introduced.

Health signal (P1): prune staleness reported the time since the NEWEST
successful prune across kinds, so on a multi-kind deployment (every real
one) a single kind whose cleanup died was masked by the others pruning
on schedule — /health stayed green while that table's index dir grew
unbounded, the exact incident the signal exists to catch. Report the
WORST kind instead and name it in the reason. The failure streak could
not cover this either: an intervening light beat resets it, so it never
reaches the threshold for a prune-only failure. Documented that split of
duties.

Spurious fallback rebuilds (P1): the benign-conflict carve-out excluded
the heavy beat, justified by "runs under the write lock, so it can't hit
this benignly" — but that lock is in-process only, so a second process
(a long `cascade backfill`, a `cascade sync`) preempts prune's Rewrite
commit. Those counted as real failures, and ~25min of cross-process
churn reached the threshold and fired a fallback rebuild, which drops
every index before recreating it; a rebuild that also lost the race was
swallowed as a warning, leaving the table with no FTS index (every
/search on that kind 500s) until the next 12h sweep. Treat commit
conflicts as benign on both beats and let prune-staleness detect a prune
that genuinely stops succeeding.

cascade rebuild (P1 ×2 + P2): it drops and recreates tables with no
guard while --help/docstrings advertised it as safe, so `rebuild --yes`
against a live daemon corrupted the rebuild (the daemon keeps writing
through cached handles). Refuse when the OME jobstore lock is held,
reusing backfill's detection and its exit code 3. It also ran the
pre-drop migration pass (`ensure_business_indexes`) against the damaged
table, so on the corruption classes it exists to repair (missing column,
un-alterable type) the recovery path died on the damage itself — skip it
via `_runtime(ensure=False)`. Reset the queue BEFORE dropping so every
crash window converges on "queue pending → re-index" instead of empty
tables with a fully-done queue (a silently empty deployment), and handle
Ctrl-C with exit 130 plus a resume hint.

Recovery guidance (P1): the nullable-vector migration error still told
users to wipe the index directory — which this PR's own runbook documents
as the wrong recovery (queue stays done, index comes back empty). Point
it at `everos cascade rebuild`. Dropped the schema-drift error's
"restart first" step too: the startup migrations only alter nullability,
never a name or type, so a name/type drift never self-heals.

Also: backfill's post-write prune passed a zero retention window from a
separate process, able to delete files under a daemon /search still
holding that version — pass the daemon's window instead. Runbook gains
the /health cascade block (thresholds, what flips healthy, why
failed_permanent does not) and its quoted schema-drift error now matches
the code.

Tests: the three safety mechanisms this PR adds were unpinned — a
one-line revert of any of them passed the suite. Added per-kind
staleness, heavy-beat benign conflict, benign-filter negative case
(an error whose message merely contains "retryable" must still count),
prune recurrence across light beats (mutation-verified: hoisting the
attempt-clock advance out of the heavy branch turns it red), the prune
timeout releasing the write lock, timeout-below-cadence, the rebuild
server guard, and a tier3 assertion that the /health cascade block is
actually wired. Froze the last fabricated-monotonic test.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Also: treat ``last_prune_attempt_at == 0.0`` as "never attempted" instead of
comparing clocks. ``monotonic()`` is boot-relative, so ``now - 0 >= cadence``
is false for the first ~cadence of container uptime — the catch-up prune was
skipped exactly when a fresh process most needs it (and made a test depend on
the runner uptime, which CI caught).

* fix(lancedb): bound every write-lock critical section

run7 (1h at 2.5x rate, concurrent CLI maintenance, doubled fuzz) reproduced a
table whose version cleanup stopped permanently: 150 versions retained, disk
11x live size, while the other two tables sat at 1 version each — and with no
error logged anywhere, because nothing failed. It simply never returned.

Three things combined. The maintenance scheduler allows one task per table (a
LanceDB table takes one writer), so it skips a kind whose task is still in
flight. The prune timeout sat *inside* the lock and covered only the cleanup
call. And the other six critical sections on that lock — add, upsert, update,
delete, delete_by_md_path, rebuild_indexes — had no deadline at all. So one
operation stuck anywhere outside that narrow window wedged the table for good:
every writer blocked on acquire, and every later heartbeat was turned away
because the stuck task never finished.

Make it structurally impossible instead of patching prune: all seven sections
now go through `LanceRepoBase._locked(budget, op)`, where the deadline covers
**acquisition and the body**. No path can wait for this lock, or hold it,
indefinitely. Budgets are hang-catchers, not throughput limits: 120s for row
writes, 600s for an index rebuild, the existing 60s for prune.

Expiry raises `VectorStoreBusyError`, deliberately under `ExternalServiceError`
so the cascade worker retries the row; under `VectorStoreError` a transient
lock contention would be marked permanently failed and need a manual
`cascade fix`.

Tests: a stuck holder now makes a waiter fail its deadline and release (the
lock is reusable afterwards), and the prune timeout is pinned as retryable.
Verified by mutation — moving the timeout back inside the lock makes a waiter
block until the enclosing observation window expires (1001ms vs 51ms), i.e.
wait forever in production.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(lancedb): size write-lock budgets from measurements

120s for a row write was a guess, and a bad one: the budget doubles as the
detection latency for a wedged table, so an over-slack value means minutes of
blocked writers before anything surfaces — the failure this change exists to
prevent.

Measured the four locked write ops on a local SSD across table sizes and batch
sizes (10k-100k rows, 50-500 rows per call): add 3-22ms, upsert (merge_insert,
the read-modify-write one) 6-25ms, update 2-4ms, delete 2-3ms; worst observation
63ms, and flat in both dimensions since these are append-and-commit, not scans.

So: writes 120s -> 15s (~240x the worst observation, enough for a contended disk
and several waiters queued ahead — the deadline includes acquisition and
asyncio.Lock is FIFO), rebuild 600s -> 300s (still the one genuinely slow
section at ~0.3s per 50k rows per indexed column). Prune stays 60s.

Test pins the sizing intent: writes stay in the tens of seconds, and
rebuild > prune > write so the slowest section is not the most eagerly killed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* feat(lancedb): record wait/hold time on write-lock critical sections

A soak run stalled one table's writes for ~16s and the logs could not say why:
maintenance beats only log at `debug`, so a section that is slow but still
inside its deadline is invisible, and the timeout warning did not distinguish
"never acquired the lock" from "acquired it and overran".

`_locked` now carries that apart. The deadline warning gains `acquired`,
`waited_seconds` and `held_seconds` — `acquired` alone answers whether a holder
was slow or this operation was — and a completed section that held the lock for
at least a second logs `lancedb_write_lock_slow_hold` at info, so a stall that
never reaches a deadline still leaves a trace.

Uses `time.monotonic` (elapsed measurement, not wall clock — the datetime
discipline bans `time.time`).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(search): reject a mismatched query vector before it reaches LanceDB

A soak run showed every slow search was a failing search: `search:vector` p50
251ms / p99 1.6s, but its 8 requests over 10s were exactly its 8 failures
(13-14s each). The cause was a query vector whose width disagreed with the
index. LanceDB only notices after the query is built and reports it as an
opaque `ValueError: Invalid input, No vector column found to match…`, which
escaped as an unhandled 500.

Validate at `_embed_query` — the single point every query vector passes
through — against the provider's declared `dim`. Microseconds instead of 13s,
and a named `ConfigurationError` (500 + CONFIGURATION_ERROR) instead of an
unhandled crash. Deliberately not `InvalidInputError`/422: callers only send
query *text*, so a bad width is our provider's fault, not the caller's.

Also cap traceback rendering. structlog's default is
`RichTracebackFormatter(show_locals=True, max_frames=100, extra_lines=3)`,
which on an async stack rendered 82 frames into 6423 log lines per exception —
85MB of server.log across 11 of them — at ~290ms of synchronous CPU each, and
risks printing request payloads into logs. With locals off and 15 frames the
same traceback is 103 lines and 10ms.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs(changelog): record the storage-reliability work under Unreleased

#379 merged without changelog entries, so this covers both it and the
follow-up work in this branch: the maintenance split (compaction vs
reclamation) that fixes unbounded index growth, bounded write-lock critical
sections, the /health cascade readiness block and its alert contract,
`cascade rebuild`, schema type-drift detection, the query-vector width check,
and the traceback-rendering cap.

Each entry states the operator-visible consequence, not just the change —
`cascade rebuild` now refusing to run against a live server, benign-conflict
warnings dropping in volume, and `/health` being able to report a stalled kind
that was previously invisible are all behaviour changes someone will notice.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: zhanghui <zhanghui@shanda.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
zhanghui 2026-08-04 20:03:01 +08:00 committed by GitHub
parent e5118c52a8
commit d256048a6d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 919 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
CHANGELOG.md
View File

@ -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/<uuid>/` 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

43
docs/cascade_runbook.md
View File

@ -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

6
pyproject.toml
View File

@ -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)

12
src/everos/core/errors.py
View File

@ -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."""

16
src/everos/core/observability/logging/factory.py
View File

@ -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,
),
),
],
)

116
src/everos/core/persistence/lancedb/repository.py
View File

@ -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/<uuid>/`` 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)

79
src/everos/entrypoints/cli/commands/cascade.py
View File

@ -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 ──────────────────────────────────────────────────────────────

28
src/everos/infra/persistence/lancedb/__init__.py
View File

@ -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."
)

2
src/everos/memory/cascade/__init__.py
View File

@ -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",
]

34
src/everos/memory/cascade/_backfill.py
View File

@ -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.

109
src/everos/memory/cascade/worker.py
View File

@ -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,

23
src/everos/memory/search/manager.py
View File

@ -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 ────────────────────────────────────────────

27
tests/integration/test_cascade_cli_integration.py
View File

@ -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 <path>``."""

13
tests/integration/test_infra/test_lancedb_schema_migration.py
View File

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

10
tests/integration/test_tiers/test_tier3_full.py
View File

@ -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"] == []

28
tests/unit/test_core/test_observability/test_logging_factory.py
View File

@ -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"
)

123
tests/unit/test_core/test_persistence/test_lancedb/test_repository.py
View File

@ -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/<uuid>/`` dirs
that ``cleanup_older_than`` leaves behind, but keeps non-empty ones."""

16
tests/unit/test_infra/test_lancedb/test_migration_lock_wiring.py
View File

@ -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"

16
tests/unit/test_memory/test_cascade/test_backfill_optimize.py
View File

@ -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

173
tests/unit/test_memory/test_cascade/test_worker.py
View File

@ -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"
)

28
tests/unit/test_memory/test_search/test_manager.py
View File

@ -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: