diff --git a/CHANGELOG.md b/CHANGELOG.md index 97a5dae..735d98f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,133 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **`[cascade]` settings section** — the four maintenance cadences + (`optimize_heartbeat_seconds`, `optimize_prune_interval_seconds`, + `optimize_prune_retention_seconds`, `optimize_rebuild_interval_seconds`) are + now configurable. They were already constructor arguments on `CascadeWorker`, + but `CascadeConfig` did not carry them and no production path passed one, so + the defaults were unreachable — which is why the 12h rebuild sweep could not + be exercised by any soak run shorter than half a day. The deadlines that + bound a hung call are deliberately **not** exposed: they are hang-catchers + sized from measured durations, where too low manufactures failures on a + healthy table and too high leaves a wedged one invisible for longer. Note + `optimize_prune_retention_seconds` has a second effect worth reading before + tuning — it also decides how long index files keep a manifest naming them, + and below LanceDB's 7-day unverified window they then wait out the full 7 + days. + +### Fixed + +- **Reads now carry a deadline** (`count` / `get_by_id` / `find_where` / + `find_where_paginated` / `search`). The write-side deadline work skipped them + on the reasoning that a read takes no lock and so blocks no writer — true, but + the cascade drain loop reads on every batch and advances strictly one batch at + a time, so a read that never returns stops the **whole md → LanceDB + projection**: claimed rows stay `processing` forever, nothing new is indexed, + and `/health` still reports healthy because a hang raises nothing. Budget 60s + (~1000x the measured 62ms flat scan over 117k rows); expiry raises the + retryable `VectorStoreBusyError`. +- **Background loops are supervised.** The drain / heartbeat / rebuild loops were + plain `create_task` coroutines: one uncaught exception ended that loop + permanently, and because the worker holds a strong reference to the task the + interpreter never printed "Task exception was never retrieved" either — the + loop's job simply stopped happening with zero output. Each now runs under a + supervisor that logs and restarts with escalating backoff (5s / 15s / 45s), + then asks the process to exit via `SIGTERM` so a restarting supervisor + (systemd, Docker, k8s) can recover it. The restart budget counts consecutive + *quick* crashes, not crashes over the process lifetime — a body that ran 60s+ + before raising starts a fresh incident, so independent transients days apart + cannot pool into a process exit (same windowed counting as systemd's + `StartLimitIntervalSec`). A done-callback covers the case the supervisor + itself ends unexpectedly. +- **The optimize-failure alert is reachable again.** The fallback rebuild reset + the same counter the health verdict reads, so a table failing 100% of the time + cycled `1..5 → 0 → 1..` and the threshold value existed only during the + sub-second rebuild — roughly 1% observable against a 30s scrape, so + `cascade.healthy` stayed green while the table never reclaimed a version. The + rate limiter now lives in its own counter (`failures_since_fallback`); only a + successful optimize clears the alert streak. Same shape as the cross-kind + `max()` masking bug: a remediation path refreshing the signal meant to report + it. +- **A rebuild no longer leaves the column without an FTS index.** `rebuild_indexes` + dropped every index and recreated it, on the assumption that LanceDB falls + back to a brute-force scan meanwhile. That holds for vector search and **not** + for FTS: with no inverted index a BM25 query raises `Cannot perform full text + search unless an INVERTED index has been created`, and since the recall legs + are gathered without `return_exceptions`, one failing leg 500s the whole + search request. Now uses `create_index(replace=True)`, which swaps atomically + — measured 0 failures across 49 queries spanning 3 replaces, versus 55 + failures for the same test against drop-then-create — and collapses the live + index fragment set exactly as before (7 index files back to 4). +- **The empty-index-dir sweep is bounded by lance's own threshold.** lance's + `cleanup.rs` unlinks a superseded index's files but never its directory — it + contains no `rmdir` at all, which is structural rather than an oversight: it + targets object stores, where paths are flat keys and an empty directory does + not exist. Only a local filesystem materialises them, and a soak run reached + 13061 dirs, 98% empty. everos sweeps them, now with three independent + guarantees instead of a self-chosen age: `rmdir` cannot delete a non-empty + directory (the kernel refuses it, so no file can be lost and there is no + check-then-act window), live index UUIDs are excluded via `list_indices()`, + and anything else must outlive `UNVERIFIED_THRESHOLD_DAYS = 7` — lance's own + bound for deciding an unreferenced index UUID is dead rather than mid-build. + The previous 300s was our invention, which is what made it indefensible. + Two consequences worth knowing: the age gate reads the dir's mtime, which + POSIX bumps when lance's cleanup empties it, so the effective reclaim horizon + is up to ~14 days (file wait + age gate) and the ceiling-load steady state is + ~1.8M dirs / ~7GB; and a sweep that blows its 60s deadline is swallowed + inside `prune()` — the cleanup commit already succeeded, so escaping would + bill a prune "failure" (feeding the fallback-rebuild threshold) and stall the + prune-staleness clock for a cleanup stall that did not happen. +- **The optimize runner's wait on an in-flight rebuild is bounded too.** The two + maintenance jobs park on each other — whichever arrives second waits — so an + unbounded wait on this side is the same hazard as the one already fixed on + the rebuild side, just seen from the other end: the kind's task slot stays + occupied, `_schedule_optimize` keeps short-circuiting on it, and that table + quietly stops being pruned. It was left open on the argument that + `rebuild_indexes` carries its own 300s deadline, which covers its critical + section but not the task's dispatch and teardown, so the transitive bound was + never real. Now bounded at 180s, logging + `cascade_lancedb_optimize_skipped_rebuild_unfinished` and skipping the beat + rather than compacting under a live rebuild — the two commit on the same + manifest, which is what the wait exists to prevent. +- **A rebuild that loses a commit race is retried** instead of waiting out the + full 12h cadence. Lance labels the conflict `Retryable` and it is: a + concurrent writer in another process won the manifest, nothing is wrong with + the table. Retries are scheduled on the kind (10min / 30min / 3h) rather than + slept through, so the other kinds in the sweep are not parked behind the + backoff. A soak run at a 600s cadence hit 3 conflicts in 119 attempts, all + while a concurrent CLI storm was running. + +- **The index-rebuild sweep can no longer park forever** waiting on the + optimize runner. That wait had no deadline, and the runner's loop condition is + "keep going while there is unindexed data" — which under sustained writes is + never, since the drain loop re-raises the flag every second against a 10s + cooldown. Now bounded at 180s, logging + `cascade_lancedb_rebuild_skipped_optimize_unfinished` and skipping the sweep + rather than dropping indices under a live optimize. **This makes the stall + visible, not absent**: under sustained ingest every sweep still times out, so + active index-UUID / FTS `part_N` growth stays unbounded there. The functional + fix requires the optimize runner to yield when a rebuild is pending, which + changes the optimize/rebuild mutual-exclusion contract and needs its own + validation — the rebuild cadence is 12h, longer than any soak run so far, so + the periodic sweep has never been exercised under load. +- **The memory-root lock wait is bounded and visible.** Acquisition polls with + `LOCK_NB` instead of blocking inside a worker thread: a blocking `flock` could + not be bounded or cancelled — cancelling the awaiting coroutine left the + thread to acquire the lock later with nobody to release it. The wait itself is + by design (the second process is supposed to wait, then find the migration + already done), but it now logs `memory_root_lock_waiting` and gives up after + `timeout_seconds` (default 30min) instead of leaving a server startup looking + like a hang whose last message is `lifespan_provider_startup name=lancedb`. + The default sits an order of magnitude above the worst legitimate hold (a + large migration is minutes) on purpose: the wait is already visible from the + first poll, and against the one case the bound exists for — a holder that is + alive but wedged — giving up at 5 minutes buys nothing over 30, while a bound + near the legitimate hold turns a slow migration into startup crashes for + every waiting process. + ### Changed - **`cascade_lancedb_optimize_conflict` now records `pruned`** — which diff --git a/src/everos/config/__init__.py b/src/everos/config/__init__.py index c295902..adbd1e3 100644 --- a/src/everos/config/__init__.py +++ b/src/everos/config/__init__.py @@ -4,7 +4,7 @@ Public API: from everos.config import ( Settings, MemorySettings, SqliteSettings, LanceDBSettings, LLMSettings, EmbeddingSettings, RerankSettings, - BoundaryDetectionSettings, + BoundaryDetectionSettings, CascadeSettings, load_settings, resolve_root, ) @@ -13,6 +13,7 @@ loader / merger / env reader). """ from .settings import BoundaryDetectionSettings as BoundaryDetectionSettings +from .settings import CascadeSettings as CascadeSettings from .settings import EmbeddingSettings as EmbeddingSettings from .settings import LanceDBSettings as LanceDBSettings from .settings import LLMSettings as LLMSettings diff --git a/src/everos/config/default.toml b/src/everos/config/default.toml index eb8eba4..e2137f8 100644 --- a/src/everos/config/default.toml +++ b/src/everos/config/default.toml @@ -146,6 +146,18 @@ threshold = 0.65 time_window_days = 7.0 +[cascade] +# Maintenance cadences (seconds). Deadlines that bound a hung call are not +# here on purpose — they live next to the code they guard and are sized from +# measurement, not preference. +optimize_heartbeat_seconds = 60.0 +optimize_prune_interval_seconds = 300.0 +# Passed to LanceDB as cleanup_older_than. Only has to outlive an in-flight +# read, but it also decides how long index files keep a manifest naming them: +# under LanceDB's 7-day unverified window they then wait out the full 7 days. +optimize_prune_retention_seconds = 60.0 +optimize_rebuild_interval_seconds = 43200.0 + [observability] # OpenTelemetry tracing export. Off by default; pure OTLP/HTTP, vendor-neutral # (Langfuse, an OTel Collector, or any OTLP backend). EverOS ships no vendor SDK. diff --git a/src/everos/config/settings.py b/src/everos/config/settings.py index c230916..50da994 100644 --- a/src/everos/config/settings.py +++ b/src/everos/config/settings.py @@ -338,6 +338,43 @@ class LanceDBSettings(BaseModel): index_cache_size_bytes: int = 16 * 1024 * 1024 +class CascadeSettings(BaseModel): + """Cascade maintenance cadences. + + These are *how often* each background job runs, not how long it is allowed + to take — the deadlines that bound a hung call stay as constants next to the + code they guard, sized from measurement, because a wrong value there either + masks a hang or manufactures failures. + + ``optimize_heartbeat_seconds``: + Idle sweep that offers every kind to the optimizer, so an unindexed tail + left by a crash is merged even without new writes. + + ``optimize_prune_interval_seconds``: + How often the heavy beat runs: reclaim the files of superseded dataset + versions. Raise it if the write-lock hold is disruptive, lower it if disk + transients are. + + ``optimize_prune_retention_seconds``: + Passed straight to LanceDB as ``cleanup_older_than`` — versions replaced + longer ago than this become eligible for deletion. It only has to outlive + an in-flight read (sub-second). Shorter shrinks the transient footprint of + superseded data fragments, but note it also decides how long index files + keep a manifest that names them: below LanceDB's 7-day unverified window, + index files lose that reference and wait out the full 7 days. + + ``optimize_rebuild_interval_seconds``: + Full index rebuild per kind, which collapses the active index fragment + count that every ``optimize()`` grows. Bounded by rebuild cost, not + correctness — a missed sweep only defers cleanup. + """ + + optimize_heartbeat_seconds: float = 60.0 + optimize_prune_interval_seconds: float = 300.0 + optimize_prune_retention_seconds: float = 60.0 + optimize_rebuild_interval_seconds: float = 12 * 60 * 60.0 + + class KnowledgeSearchSettings(BaseModel): """``[knowledge.search]`` — retrieval tuning for the knowledge module.""" @@ -417,6 +454,7 @@ class Settings(BaseSettings): boundary_detection: BoundaryDetectionSettings = BoundaryDetectionSettings() memorize: MemorizeSettings = MemorizeSettings() clustering: ClusteringSettings = ClusteringSettings() + cascade: CascadeSettings = CascadeSettings() multimodal: MultimodalSettings = MultimodalSettings() knowledge: KnowledgeSettings = KnowledgeSettings() observability: ObservabilitySettings = ObservabilitySettings() diff --git a/src/everos/core/persistence/lancedb/base.py b/src/everos/core/persistence/lancedb/base.py index 72cdb55..09f685d 100644 --- a/src/everos/core/persistence/lancedb/base.py +++ b/src/everos/core/persistence/lancedb/base.py @@ -104,11 +104,28 @@ class BaseLanceTable(LanceModel): ) @classmethod - async def ensure_fts_indexes(cls, table: AsyncTable) -> None: + async def ensure_fts_indexes( + cls, table: AsyncTable, *, replace: bool = False + ) -> None: """Create FTS indexes on every column in :attr:`BM25_FIELDS`. Idempotent: columns that already have an index are skipped, so - this is safe to call on every startup. The FTS config is fixed + this is safe to call on every startup. + + ``replace=True`` rebuilds each column's index in place instead of + skipping it — used by :meth:`LanceRepoBase.rebuild_indexes`, which + needs a fresh index but must never leave the column *without* one. + Dropping first would do that, and a BM25 query in that window does not + degrade — it raises ``Cannot perform full text search unless an + INVERTED index has been created`` (measured; vector search does fall + back to a flat scan, FTS does not). Since the recall legs are gathered + without ``return_exceptions``, that window turns into a 500 on the + whole search request. ``create_index(replace=True)`` is atomic: 49 + concurrent queries across 3 replaces saw 0 failures, and it collapses + the live fragment set exactly as drop+create does (7 index files back + to 4, measured). + + The FTS config is fixed to the app-layer pre-tokenisation + LanceDB normalisation convention (designed for **multilingual mixed content**): @@ -145,10 +162,11 @@ class BaseLanceTable(LanceModel): indices = await table.list_indices() indexed_cols = {col for idx in indices for col in (idx.columns or [])} for field in cls.BM25_FIELDS: - if field in indexed_cols: + if field in indexed_cols and not replace: continue await table.create_index( column=field, + replace=replace, config=FTS( with_position=False, base_tokenizer="whitespace", diff --git a/src/everos/core/persistence/lancedb/repository.py b/src/everos/core/persistence/lancedb/repository.py index 53a8025..f118161 100644 --- a/src/everos/core/persistence/lancedb/repository.py +++ b/src/everos/core/persistence/lancedb/repository.py @@ -20,6 +20,7 @@ from typing import Any, ClassVar from lancedb import AsyncTable +from everos.component.utils.datetime import get_utc_now from everos.core.errors import VectorStoreBusyError from everos.core.observability.logging import get_logger @@ -74,6 +75,63 @@ task per kind and skips a kind whose task is in flight, so a compaction that never returns parks that kind's maintenance permanently. Measured at ~460ms on a table with 77 retained versions.""" +_HUSK_MIN_AGE_SECONDS = 7 * 24 * 60 * 60.0 +"""Minimum age of an empty ``_indices//`` dir before it is removed. + +Deliberately **lance's own number**, not one of ours. ``cleanup.rs`` defines +``UNVERIFIED_THRESHOLD_DAYS = 7`` and uses it for exactly this judgement: an +index UUID that no manifest references is only assumed dead once it is at least +7 days old, because before that it is indistinguishable from an index build +still in progress. Matching the threshold means this sweep can never be more +aggressive than lance itself — the earlier 300s value was our invention, and +that is precisely what made it unjustifiable. + +**The cost of the gate is accepted, with numbers — and the effective horizon +is up to 14 days, not 7.** The gate reads ``st_mtime``, and POSIX bumps a +directory's mtime whenever an entry is unlinked from it — so the clock +restarts the moment lance's cleanup empties the husk, not when the dir was +created. Under a short retention window the index *files* themselves first +wait out lance's 7-day unverified window (see :meth:`LanceRepoBase.prune`), +so a husk is reclaimed up to file-wait + 7 days after it appeared. Each one +costs an inode plus one 4KB block. Measured on a soak at ceiling load (see +``DEFAULT_OPTIMIZE_MIN_INTERVAL_SECONDS``): ~127k dirs/day, so the ~14-day +steady state is ~1.8M dirs = **~7GB of empty directories and ~28% of a +default 98GB ext4's inodes**. That is the worst case, not the expected one — +it needs +sustained saturation, and a single-user deployment writing a few hundred times +a day sits four orders of magnitude below it (tens of MB). Platform-wise only +ext4 has a fixed inode budget at all; APFS and xfs allocate dynamically, and +Windows is out of scope. The knob to reach for if this ever does bite is the +optimize cooldown, not this threshold — see there. +""" + +_HUSK_SWEEP_TIMEOUT_SECONDS = 60.0 +"""Deadline on the (lock-free) husk sweep. Same last-resort role as +:data:`_COMPACT_TIMEOUT_SECONDS`: the maintenance scheduler skips a kind whose +task is in flight, so a sweep that never returns would park that kind's +maintenance forever. Measured at ~460ms over 13061 dirs in the soak (~35us +per dir), which puts the ceiling-load steady state (~1.8M dirs, see +:data:`_HUSK_MIN_AGE_SECONDS`) right at this budget. That is tolerated rather +than sized around: a timeout is swallowed inside :meth:`LanceRepoBase.prune` +(the cleanup commit already succeeded, and billing the sweep to the prune +ledger is exactly the signal corruption this module works to avoid), and the +orphaned worker thread — a cancelled ``to_thread`` future does not stop the +thread — finishes the walk anyway, so the reclamation still happens.""" + +_READ_TIMEOUT_SECONDS = 60.0 +"""Deadline on every read. Reads take no lock, so a hung read blocks no writer +— but it does park the caller, and the cascade drain loop reads on every batch +while advancing strictly one batch at a time. A read that never returns +therefore stops the whole md -> LanceDB projection, leaving claimed rows in +``processing`` forever with nothing logged (a hang raises nothing, so the +drain-failure counter stays at zero and ``/health`` keeps reporting healthy). +Same last-resort shape as :data:`_COMPACT_TIMEOUT_SECONDS`, and generous by +design: everos builds no vector ANN index, so reads are flat scans — measured +~62ms over 117k rows, i.e. 60s is ~1000x headroom and never fires normally. On +expiry the caller gets a retryable :class:`VectorStoreBusyError`, so a drain row +is retried and a search request fails with a structured error rather than +hanging the request.""" + _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 @@ -83,43 +141,6 @@ slow but under its deadline is invisible (the maintenance beat only logs at 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. - - ``optimize(cleanup_older_than=…)`` deletes the *files* belonging to - superseded index versions but leaves the now-empty per-UUID - directory behind. Over a long-lived churned table these husks - accumulate into tens of thousands of empty dirs (soak: 13061 dirs, - 98% empty), which bloats inode usage and slows directory scans even - though they hold no data. - - Pure filesystem bookkeeping, no LanceDB manifest involvement — an - empty ``_indices//`` means its files were already cleaned - because no live manifest references that index version, so removing - the directory cannot orphan live data. Must only run while the - table's write lock is held (no concurrent index build could be - mid-populating a freshly-created, still-empty UUID dir). Best-effort: - a dir that becomes non-empty or vanishes between the scan and the - ``rmdir`` is skipped, not an error. - """ - indices = Path(table_uri) / "_indices" - if not indices.is_dir(): - return 0 - removed = 0 - for child in indices.iterdir(): - if not child.is_dir(): - continue - try: - next(child.iterdir()) # has an entry → not empty, keep - except StopIteration: - try: - child.rmdir() - removed += 1 - except OSError: - pass # raced (repopulated / already gone) — skip - return removed - - def _q(value: str) -> str: """Escape single quotes for a LanceDB SQL-like ``where`` predicate. @@ -132,6 +153,51 @@ def _q(value: str) -> str: return value.replace("'", "''") +def _remove_empty_index_dirs( + table_uri: str, *, live_uuids: frozenset[str], min_age_seconds: float +) -> int: + """Remove empty ``_indices//`` husks under a table dir; return count. + + lance's cleanup unlinks the *files* of a superseded index but never the + directory — ``cleanup.rs`` contains no ``rmdir``/``remove_dir`` at all, and + that is structural rather than an oversight: lance is written against an + object store where paths are flat keys and an "empty directory" does not + exist. Only a local filesystem materialises them, where they accumulate as + inodes and slow every directory scan (a soak run reached 13061 dirs, 98% + empty). + + Three independent guarantees, in order of strength: + + 1. **``rmdir`` cannot delete data.** The kernel refuses it on a non-empty + directory (``ENOTEMPTY``). No file can be lost through this function + whatever the rest of the logic decides — and because the check *is* the + operation, there is no check-then-act window to race. + 2. **Live indexes are excluded** by UUID, read from ``list_indices()``. + 3. **Anything else waits out lance's own conservatism bound** + (:data:`_HUSK_MIN_AGE_SECONDS`): a directory a concurrent + ``create_index`` just made is seconds old, so it can never qualify. + + Best-effort throughout: a directory that becomes non-empty, vanishes, or is + unreadable between listing and ``rmdir`` is skipped, not an error. + """ + indices = Path(table_uri) / "_indices" + if not indices.is_dir(): + return 0 + cutoff = get_utc_now().timestamp() - min_age_seconds + removed = 0 + for child in indices.iterdir(): + if not child.is_dir() or child.name in live_uuids: + continue + try: + if child.stat().st_mtime > cutoff: + continue + child.rmdir() + except OSError: + continue + removed += 1 + return removed + + class LanceRepoBase[T: BaseLanceTable]: """Generic CRUD repository for one LanceDB table. @@ -432,10 +498,20 @@ class LanceRepoBase[T: BaseLanceTable]: load comes from running under the write lock so the cleanup commit never loses the manifest race, not from the flag. - After the cleanup commit, the now-empty ``_indices//`` husks - that ``cleanup_older_than`` leaves behind are removed (still under - the lock, so no index build can be mid-populating one); the sweep is - offloaded to a thread so a large dir count does not block the loop. + ``cleanup_older_than`` deletes the *files* under a + superseded ``_indices//`` but never the directory (lance's + ``cleanup.rs`` contains no directory removal at all — it is written + against an object store where an empty directory does not exist), so + those husks accumulate on a local filesystem: a soak run reached 13061 + dirs, 98% empty. They are swept here, outside the lock — see + :func:`_remove_empty_index_dirs` for why that is safe. + + This is a *separate* gap from index files not being reclaimed while + they are young: lance skips an index UUID that no manifest references + until it is 7 days old, and a short retention window deletes those + manifests first, so the files lose their last reference and wait out + the full 7 days (measured: 260MB retained on a 19k-row soak table, + reclaimed in full once backdated past the threshold). The trade-off is a brief write stall (~seconds on a churned table, dominated by the cleanup's file scan/delete — flat, not proportional @@ -448,7 +524,34 @@ class LanceRepoBase[T: BaseLanceTable]: table = await self._table() await table.optimize(cleanup_older_than=older_than, delete_unverified=False) table_uri = await table.uri() - removed = await asyncio.to_thread(_remove_empty_index_dirs, table_uri) + live_uuids = frozenset( + i.index_uuid for i in await table.list_indices() if i.index_uuid + ) + # Lock-free: the sweep can only ``rmdir``, which the kernel refuses on a + # non-empty directory, so it cannot lose data no matter who else is + # writing. Keeping it out of the critical section also means a slow + # filesystem walk cannot overrun the prune budget. + # + # Best-effort means best-effort: the cleanup commit above already + # succeeded, so a sweep timeout must not escape ``prune()``. Letting it + # escape bills the failure to the wrong account — the optimize + # scheduler counts a prune "failure" (feeding the fallback-rebuild + # threshold) and the prune-staleness clock stops advancing, both + # reporting a cleanup stall that did not happen. Same defect shape as + # the alert counter the fallback rebuild used to zero: an auxiliary + # path corrupting the main signal's ledger. The skipped husks are + # retried on the next heavy beat. + try: + async with self._deadline(_HUSK_SWEEP_TIMEOUT_SECONDS, "prune_husk_sweep"): + removed = await asyncio.to_thread( + _remove_empty_index_dirs, + table_uri, + live_uuids=live_uuids, + min_age_seconds=_HUSK_MIN_AGE_SECONDS, + ) + except VectorStoreBusyError: + # _deadline already logged lancedb_operation_deadline_exceeded. + removed = 0 if removed: logger.debug( "lancedb_pruned_empty_index_dirs", @@ -483,12 +586,20 @@ class LanceRepoBase[T: BaseLanceTable]: accumulating one new UUID (vector) / one new ``part_N`` (FTS) per call. - This method is the workaround: drop every existing index and - rebuild from the schema's ``ensure_fts_indexes`` contract. The - rebuild is **O(N) full retrain** but cheap in practice (~0.3s - for 50k rows × 2 FTS columns on local SSD), and during the - window LanceDB transparently falls back to brute-force scan so - queries and writes stay available. + This method is the workaround: rebuild every indexed column from the + schema's ``ensure_fts_indexes`` contract. Measured effect — the live + index goes from 7 files back to 4 after 25 ``optimize()`` beats, i.e. + the fragment set does collapse. The rebuild is an **O(N) full retrain** + but cheap in practice (~0.3s for 50k rows × 2 FTS columns on local SSD). + + It rebuilds **in place** (``create_index(replace=True)``) rather than + dropping first. An earlier version dropped every index and recreated + them, on the assumption that LanceDB falls back to a brute-force scan + meanwhile. That is true for vector search and **false for FTS**: with no + inverted index a BM25 query raises ``Cannot perform full text search + unless an INVERTED index has been created`` (measured). Because the + recall legs are gathered without ``return_exceptions``, one failing leg + fails the whole search request, so the window was a source of 500s. **Cadence** — :class:`CascadeWorker` runs this on a slow loop (default 12h per kind). Frequency is bounded by the rebuild @@ -510,16 +621,27 @@ class LanceRepoBase[T: BaseLanceTable]: """ async with self._locked(_REBUILD_TIMEOUT_SECONDS, "rebuild_indexes"): table = await self._table() + # Replace in place rather than drop-then-create. The live columns + # must never be left without an index: FTS does not degrade when + # its index is missing the way vector search does — it raises + # ``Cannot perform full text search unless an INVERTED index has + # been created``, and the recall legs are gathered without + # ``return_exceptions``, so the whole search request 500s. Only + # indexes on columns that are no longer indexed at all get dropped; + # nothing queries those, so their drop opens no window. + wanted = set(self.schema.BM25_FIELDS or ()) for idx in await table.list_indices(): - await table.drop_index(idx.name) - await self.schema.ensure_fts_indexes(table) + if not wanted.intersection(idx.columns or ()): + await table.drop_index(idx.name) + await self.schema.ensure_fts_indexes(table, replace=True) # ── Read ─────────────────────────────────────────────────────────────── async def count(self) -> int: """Total row count.""" - table = await self._table() - return await table.count_rows() + async with self._deadline(_READ_TIMEOUT_SECONDS, "count"): + table = await self._table() + return await table.count_rows() async def get_by_id( self, @@ -534,13 +656,14 @@ class LanceRepoBase[T: BaseLanceTable]: predicate; everos's PK convention is ``_`` which never contains quotes, so the escape is defensive. """ - table = await self._table() - rows = ( - await table.query() - .where(f"{id_field} = '{_q(id_value)}'") - .limit(1) - .to_list() - ) + async with self._deadline(_READ_TIMEOUT_SECONDS, "get_by_id"): + table = await self._table() + rows = ( + await table.query() + .where(f"{id_field} = '{_q(id_value)}'") + .limit(1) + .to_list() + ) if not rows: return None return self.schema.model_validate(rows[0]) @@ -558,8 +681,9 @@ class LanceRepoBase[T: BaseLanceTable]: Use :meth:`search` when you need ``_distance`` or want to mix ANN with filters. """ - table = await self._table() - rows = await table.query().where(where).limit(limit).to_list() + async with self._deadline(_READ_TIMEOUT_SECONDS, "find_where"): + table = await self._table() + rows = await table.query().where(where).limit(limit).to_list() return [self.schema.model_validate(r) for r in rows] async def find_one_where(self, where: str) -> T | None: @@ -606,19 +730,20 @@ class LanceRepoBase[T: BaseLanceTable]: ``total`` is ``count_rows(filter=where)`` (the predicate's true match count, regardless of ``max_fetch``). """ - table = await self._table() - total = await table.count_rows(filter=where) - if total > max_fetch: - logger.warning( - "find_where_paginated truncated", - extra={ - "table": self.table_name, - "where": where, - "total": total, - "max_fetch": max_fetch, - }, - ) - arrow_tbl = await table.query().where(where).limit(max_fetch).to_arrow() + async with self._deadline(_READ_TIMEOUT_SECONDS, "find_where_paginated"): + table = await self._table() + total = await table.count_rows(filter=where) + if total > max_fetch: + logger.warning( + "find_where_paginated truncated", + extra={ + "table": self.table_name, + "where": where, + "total": total, + "max_fetch": max_fetch, + }, + ) + arrow_tbl = await table.query().where(where).limit(max_fetch).to_arrow() order = "descending" if descending else "ascending" arrow_tbl = arrow_tbl.sort_by([(sort_by, order)]) offset = (page - 1) * page_size @@ -662,13 +787,14 @@ class LanceRepoBase[T: BaseLanceTable]: List of row dicts (LanceDB native shape — fields depend on ``schema``; ``_distance`` added when ``vector`` is given). """ - table = await self._table() - q = table.query() - if vector is not None: - q = q.nearest_to(list(vector)) - if where is not None: - q = q.where(where) - return await q.limit(limit).to_list() + async with self._deadline(_READ_TIMEOUT_SECONDS, "search"): + table = await self._table() + q = table.query() + if vector is not None: + q = q.nearest_to(list(vector)) + if where is not None: + q = q.where(where) + return await q.limit(limit).to_list() # ── Update ───────────────────────────────────────────────────────────── diff --git a/src/everos/core/persistence/locking.py b/src/everos/core/persistence/locking.py index 4759cb0..71bb3ef 100644 --- a/src/everos/core/persistence/locking.py +++ b/src/everos/core/persistence/locking.py @@ -6,22 +6,61 @@ public surface is an :func:`contextlib.asynccontextmanager` so callers use ``async with memory_root_lock(mr):``; the underlying syscalls have no async equivalent so they run in a worker thread via :func:`anyio.to_thread.run_sync`. + +**Acquisition polls with ``LOCK_NB`` instead of blocking in the thread.** A +blocking ``flock`` cannot be bounded or cancelled: the syscall runs in a worker +thread, and cancelling the awaiting coroutine leaves that thread to acquire the +lock later with nobody left to release it — strictly worse than waiting. Short +non-blocking attempts on a poll interval give the same semantics while making +the wait bounded, cancellable, and visible in the log. Visibility is the point: +the wait itself is by design (see :func:`ensure_business_indexes` — the second +process is *supposed* to wait, then find the work already done), but without a +log line a server startup that waits on it looks like a hang with no last +message beyond ``lifespan_provider_startup``. """ from __future__ import annotations import fcntl import os +import time from collections.abc import AsyncIterator from contextlib import asynccontextmanager import anyio +from everos.core.observability.logging import get_logger + from .memory_root import MemoryRoot +logger = get_logger(__name__) + +DEFAULT_LOCK_TIMEOUT_SECONDS = 1800.0 +"""Default upper bound on waiting for the memory-root lock. + +Deliberately an order of magnitude above the legitimate hold, not near it: +the legitimate holder is a one-shot FTS/schema migration whose runtime is +O(rows), so a large memory-root on a slow disk can hold it for minutes — a +bound *near* that (an earlier draft used 300s) turns the worst legitimate +migration into a startup crash for every process waiting on it. + +Generosity costs almost nothing here, because this timeout's job is +diagnosis, not recovery. The wait is already visible from the first poll +(``memory_root_lock_waiting``), and when the holder is genuinely stuck — +alive but wedged inside its critical section, the only case this bounds — +giving up sooner does not un-stick it: the error and the operator's next +move (inspect the holding process) are the same at 5 minutes or 30. +``flock`` is released by the kernel on process exit, so a *dead* holder +never needs this. +""" + +_LOCK_POLL_INTERVAL_SECONDS = 0.5 +"""Gap between non-blocking acquisition attempts. Startup-path latency, so +sub-second is imperceptible; keeping it off zero avoids a spin.""" + class LockError(RuntimeError): - """Raised when the memory-root lock cannot be acquired in non-blocking mode.""" + """Raised when the memory-root lock cannot be acquired.""" @asynccontextmanager @@ -29,18 +68,22 @@ async def memory_root_lock( memory_root: MemoryRoot, *, blocking: bool = True, + timeout_seconds: float | None = DEFAULT_LOCK_TIMEOUT_SECONDS, ) -> AsyncIterator[None]: """Acquire an exclusive process lock on the memory-root. Args: memory_root: The memory-root to lock. The lock anchor file (``/.lock``) is created on first use. - blocking: If ``True`` (default), wait until the lock is free. If - ``False``, raise :class:`LockError` immediately when another - process holds it. + blocking: If ``True`` (default), wait until the lock is free or + ``timeout_seconds`` elapses. If ``False``, raise + :class:`LockError` immediately when another process holds it. + timeout_seconds: Upper bound on the wait when ``blocking=True``; + ``None`` waits indefinitely. Ignored when ``blocking=False``. Raises: - LockError: When ``blocking=False`` and the lock is already held. + LockError: When the lock is held and either ``blocking=False`` or the + timeout elapsed. """ await anyio.Path(memory_root.root).mkdir(parents=True, exist_ok=True) lock_path = memory_root.lock_file @@ -52,21 +95,53 @@ async def memory_root_lock( lambda: os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o644) ) - flags = fcntl.LOCK_EX - if not blocking: - flags |= fcntl.LOCK_NB - + started = time.monotonic() + deadline = None if timeout_seconds is None else started + timeout_seconds + announced = False try: - await anyio.to_thread.run_sync(fcntl.flock, fd, flags) - except BlockingIOError as exc: + while True: + try: + await anyio.to_thread.run_sync( + fcntl.flock, fd, fcntl.LOCK_EX | fcntl.LOCK_NB + ) + break + except BlockingIOError as exc: + if not blocking: + raise LockError( + "another process already holds the memory-root lock " + f"at {lock_path}" + ) from exc + if not announced: + logger.info( + "memory_root_lock_waiting", + path=str(lock_path), + timeout_seconds=timeout_seconds, + ) + announced = True + if deadline is not None and time.monotonic() >= deadline: + raise LockError( + "timed out after " + f"{time.monotonic() - started:.1f}s waiting for the " + f"memory-root lock at {lock_path}. The holder is " + "still alive (the kernel releases a dead process's " + "flock automatically) — inspect the process holding " + f"{lock_path} rather than retrying this one" + ) from exc + await anyio.sleep(_LOCK_POLL_INTERVAL_SECONDS) + except BaseException: await anyio.to_thread.run_sync(os.close, fd) - raise LockError( - f"another process already holds the memory-root lock at {lock_path}" - ) from exc + raise - # Lock acquired — release + close strictly on exit. The BlockingIOError - # path above already cleaned up its fd, so it must NOT enter this - # finally block (otherwise we'd double-close). + if announced: + logger.info( + "memory_root_lock_acquired_after_wait", + path=str(lock_path), + waited_seconds=round(time.monotonic() - started, 1), + ) + + # Lock acquired — release + close strictly on exit. The failure paths above + # already closed their fd, so they must NOT enter this finally block + # (otherwise we'd double-close). try: yield finally: diff --git a/src/everos/memory/cascade/orchestrator.py b/src/everos/memory/cascade/orchestrator.py index 9fa06d0..7b3cc24 100644 --- a/src/everos/memory/cascade/orchestrator.py +++ b/src/everos/memory/cascade/orchestrator.py @@ -18,6 +18,7 @@ import asyncio import dataclasses from everos.component.tokenizer import Tokenizer +from everos.config import load_settings from everos.core.observability.logging import get_logger from everos.core.persistence import MemoryRoot from everos.infra.persistence.sqlite import QueueSummary, md_change_state_repo @@ -26,7 +27,13 @@ from .handlers import HandlerDeps from .registry import build_handlers from .scanner import CascadeScanner from .watcher import CascadeWatcher -from .worker import CascadeWorker +from .worker import ( + DEFAULT_OPTIMIZE_HEARTBEAT_SECONDS, + DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS, + DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS, + DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS, + CascadeWorker, +) logger = get_logger(__name__) @@ -66,9 +73,17 @@ class CascadeHealth: class CascadeConfig: """Construction-time knobs for the orchestrator. - Defaults are sized for a lightweight (single-user / small-team) dev - box; production tuning can surface these into - :class:`everos.config.Settings` once the daemon has wall-clock data. + Defaults are sized for a lightweight (single-user / small-team) dev box. + The maintenance cadences come from :class:`everos.config.CascadeSettings` + via :meth:`from_settings`, which every production construction path uses — + they were constructor-only for long enough that the 12h rebuild sweep could + not be exercised by any soak run short of half a day. + + Deliberately *not* configurable: the deadlines that bound a hung call + (read / write / prune / rebuild). Those are hang-catchers sized from + measured durations; too low manufactures failures, too high makes a wedged + table invisible for longer. They stay as constants beside the code they + guard, each with its measurement in the docstring. """ scan_interval_seconds: float = 30.0 @@ -76,6 +91,23 @@ class CascadeConfig: worker_max_retry: int = 3 worker_poll_interval_seconds: float = 1.0 worker_retry_backoff_seconds: float = 2.0 + optimize_heartbeat_seconds: float = DEFAULT_OPTIMIZE_HEARTBEAT_SECONDS + optimize_prune_interval_seconds: float = DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS + optimize_prune_retention_seconds: float = DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS + optimize_rebuild_interval_seconds: float = DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS + + @classmethod + def from_settings(cls) -> CascadeConfig: + """Build with maintenance cadences taken from ``[cascade]`` settings.""" + cascade = load_settings().cascade + return cls( + optimize_heartbeat_seconds=cascade.optimize_heartbeat_seconds, + optimize_prune_interval_seconds=cascade.optimize_prune_interval_seconds, + optimize_prune_retention_seconds=(cascade.optimize_prune_retention_seconds), + optimize_rebuild_interval_seconds=( + cascade.optimize_rebuild_interval_seconds + ), + ) class CascadeOrchestrator: @@ -89,7 +121,7 @@ class CascadeOrchestrator: config: CascadeConfig | None = None, ) -> None: self._memory_root = memory_root - self._config = config or CascadeConfig() + self._config = config or CascadeConfig.from_settings() deps = HandlerDeps( memory_root=memory_root, tokenizer=tokenizer, @@ -105,6 +137,16 @@ class CascadeOrchestrator: max_retry=self._config.worker_max_retry, poll_interval_seconds=self._config.worker_poll_interval_seconds, retry_backoff_seconds=self._config.worker_retry_backoff_seconds, + optimize_heartbeat_seconds=self._config.optimize_heartbeat_seconds, + optimize_prune_interval_seconds=( + self._config.optimize_prune_interval_seconds + ), + optimize_prune_retention_seconds=( + self._config.optimize_prune_retention_seconds + ), + optimize_rebuild_interval_seconds=( + self._config.optimize_rebuild_interval_seconds + ), ) self._watcher: CascadeWatcher | None = None self._started = False diff --git a/src/everos/memory/cascade/worker.py b/src/everos/memory/cascade/worker.py index 6117761..4223246 100644 --- a/src/everos/memory/cascade/worker.py +++ b/src/everos/memory/cascade/worker.py @@ -56,8 +56,12 @@ from __future__ import annotations import asyncio import contextlib import datetime as dt +import functools +import signal import time +from collections.abc import Callable, Coroutine from dataclasses import dataclass +from typing import Any from everos.core.errors import ExternalServiceError from everos.core.observability.logging import get_logger @@ -73,6 +77,27 @@ DEFAULT_MAX_RETRY = 3 DEFAULT_POLL_INTERVAL_SECONDS = 1.0 DEFAULT_RETRY_BACKOFF_SECONDS = 2.0 DEFAULT_OPTIMIZE_MIN_INTERVAL_SECONDS = 10.0 +"""Throttle between ``optimize()`` runs on one kind. + +Not a visibility delay. A row is searchable the moment its upsert commits — +LanceDB flat-scans the unindexed tail, and that covers BM25 as well as vector +and scalar (verified: a row with ``num_unindexed_rows=1`` is returned by +``nearest_to_text``). What ``optimize`` buys is folding that row out of the tail +and into the index, i.e. speed. Sparse writes do not even wait: the scheduler +uses ``max(0, interval - elapsed)``, so when the last run is already older than +the interval the next one starts immediately. + +It is also the **ceiling on index-directory growth**, which is the reason to +think twice before lowering it. Every beat leaves new ``_indices//`` +dirs behind, and lance never removes the empty ones, so the accrual rate is +capped by this interval rather than by write volume: past roughly one write per +table per interval the beats coalesce and writing harder adds nothing. Measured +at that ceiling: ~127k dirs/day across three active tables, ~1.8M at the +~14-day reclaim horizon (~7GB of empty dirs — the horizon is the file wait +plus the 7-day age gate, see ``_HUSK_MIN_AGE_SECONDS`` in the lancedb +repository module). Raising this interval lowers that proportionally — 60s +would cut it to a sixth — at the cost of a longer flat-scanned tail. +""" DEFAULT_OPTIMIZE_HEARTBEAT_SECONDS = 60.0 _OPTIMIZE_FAILURE_ALERT_THRESHOLD = 5 """Consecutive **non-benign** ``optimize()`` failures (per kind) before @@ -122,6 +147,62 @@ repo's deadline. Generous enough never to fire on a healthy beat (prune's own budget is 60s), tight enough that a hang costs one cadence, not forever. """ +_REBUILD_CONFLICT_BACKOFFS_SECONDS = (600.0, 1800.0, 10800.0) +"""Delay before each re-attempt at a kind's index rebuild: 10min, 30min, 3h. + +Only a lost commit race is retried — lance marks it ``Retryable`` and the table +is fine; a concurrent writer in another process simply won the manifest. Any +other failure defers to the next sweep, because retrying a real error just +burns the write lock. + +Scaled to the 12h rebuild cadence, not to the conflict. A seconds-long backoff +would spend the whole budget inside one contention window and then leave the +kind unindexed for a full cadence; spreading the attempts over hours means the +retries land in genuinely different load conditions. The schedule is a +*deadline recorded on the kind*, never a sleep: :meth:`_rebuild_loop` walks +kinds sequentially, so sleeping here would park every later kind behind this +one (7 kinds x 3h would outlast the cadence itself).""" + +_REBUILD_LOOP_TICK_SECONDS = 60.0 +"""How often the rebuild loop wakes to pick up due retries between sweeps. +Cheap: a dict scan per tick, and only kinds with a due deadline do work.""" + +_LOOP_RESTART_BACKOFF_SECONDS = (5.0, 15.0, 45.0) +"""Backoff before each restart of a background loop that raised. + +The three long-lived loops (drain / heartbeat / rebuild) are plain +``create_task`` coroutines. Without supervision, one uncaught exception ends +that loop **permanently and silently**: nothing restarts it, and because +``self._*_task`` keeps a strong reference the interpreter never prints the +"Task exception was never retrieved" warning either (that fires on GC). The +loop's job simply stops happening. So each loop runs under +:meth:`CascadeWorker._supervise`, which logs, waits, and restarts. + +Escalating rather than fixed: a transient cause (a closing event loop, a +momentarily unavailable table) clears within seconds, while a deterministic +one would otherwise spin. After the last entry is used the worker asks the +process to exit (see :meth:`CascadeWorker._request_process_exit`) — a server +whose projection pipeline is permanently dead should not keep serving as if +healthy. + +The budget is **per incident, not per process lifetime**: a body that ran at +least :data:`_LOOP_STABLE_RUN_SECONDS` before crashing gets a full budget +again. Without that reset, rare *independent* transients — one every few +days, each recovered by a single restart — would still spend the budget one +by one and SIGTERM the server weeks later on the 4th, which punishes exactly +the case supervision exists to absorb. This mirrors how process supervisors +count restarts within a window (systemd ``StartLimitIntervalSec``, Erlang +``max_restarts`` per ``max_seconds``) rather than forever. +""" + +_LOOP_STABLE_RUN_SECONDS = 60.0 +"""A supervised loop body that ran at least this long before raising is +treated as a fresh incident (restart budget resets). Sized well above the +escalation ladder's total (5+15+45 = 65s of *backoff*, but each attempt's +run time counts from body start): a deterministic crash-on-startup fails in +milliseconds and cannot reach it, while a loop that did an hour of honest +work before hitting a transient obviously should not inherit stale strikes.""" + DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS = 12 * 60 * 60.0 """How often (per kind) to do a full ``drop_index + create_index`` rebuild. @@ -225,11 +306,39 @@ class _KindOptimizerState: last_prune_at: float = 0.0 dirty: bool = False optimize_failures: int = 0 - """Consecutive ``optimize()`` failure count; reset to 0 on success. - Drives escalation to ``error`` at - :data:`_OPTIMIZE_FAILURE_ALERT_THRESHOLD` so a stuck optimize (which - stalls version cleanup and grows the index dir) is not swallowed as a - silent warning stream.""" + """Consecutive ``optimize()`` failures **since the last success**. + + Drives the health verdict and escalation to ``error`` at + :data:`_OPTIMIZE_FAILURE_ALERT_THRESHOLD` so a stuck optimize (which stalls + version cleanup and grows the index dir) is not swallowed as a silent + warning stream. + + Reset **only** by a successful optimize — never by the fallback rebuild it + triggers. That distinction is the whole point of splitting this from + ``failures_since_fallback``: zeroing the alert counter inside the branch + that fires at the threshold made ``optimize_failure_streak >= threshold`` + effectively unobservable. A table failing 100% of the time cycled + 1..threshold -> 0 -> 1.., and the only window where a poller could read the + threshold value was the sub-second fallback rebuild itself. Same shape as + the cross-kind ``max()`` masking bug from run7: a remediation path + refreshing the very signal that is supposed to report it. + """ + failures_since_fallback: int = 0 + """Failures since the last fallback rebuild — the rate limiter. + + Reset by the fallback rebuild so it fires at most once per + :data:`_OPTIMIZE_FAILURE_ALERT_THRESHOLD` failures instead of on every + failure. This is the job the reset was originally there for; it just used + to share a field with the alert signal. + """ + rebuild_retry_at: float = 0.0 + """Monotonic deadline for re-attempting a rebuild that lost a commit race. + ``0`` means nothing pending. Set instead of sleeping so the rebuild loop + stays free to serve the other kinds — see + :data:`_REBUILD_CONFLICT_BACKOFFS_SECONDS`.""" + rebuild_attempt: int = 0 + """Consecutive lost commit races for this kind; indexes into the backoff + schedule and resets on any completed rebuild.""" task: asyncio.Task[None] | None = None rebuild_task: asyncio.Task[None] | None = None """In-flight rebuild task slot, separate from ``task`` so ordinary @@ -374,15 +483,115 @@ class CascadeWorker: return self._stop.clear() self._started_at = time.monotonic() - self._task = asyncio.create_task(self._run_loop(), name="cascade-worker") - self._heartbeat_task = asyncio.create_task( - self._heartbeat_loop(), name="cascade-worker-heartbeat" + self._task = self._spawn_loop("drain", self._run_loop, "cascade-worker") + self._heartbeat_task = self._spawn_loop( + "heartbeat", self._heartbeat_loop, "cascade-worker-heartbeat" ) - self._rebuild_task = asyncio.create_task( - self._rebuild_loop(), name="cascade-worker-rebuild" + self._rebuild_task = self._spawn_loop( + "rebuild", self._rebuild_loop, "cascade-worker-rebuild" ) logger.info("cascade_worker_started", batch_size=self._batch_size) + def _spawn_loop( + self, + loop_name: str, + body: Callable[[], Coroutine[Any, Any, None]], + task_name: str, + ) -> asyncio.Task[None]: + """Start one supervised background loop. + + Two layers, because they fail differently: :meth:`_supervise` restarts + the loop body when it raises, and the done-callback is the last-resort + observer for the case the supervisor itself ends unexpectedly (a + ``BaseException`` it deliberately does not catch). Without the callback + that ending is invisible — see :data:`_LOOP_RESTART_BACKOFF_SECONDS`. + """ + task = asyncio.create_task(self._supervise(loop_name, body), name=task_name) + task.add_done_callback(functools.partial(self._on_loop_task_done, loop_name)) + return task + + async def _supervise( + self, + loop_name: str, + body: Callable[[], Coroutine[Any, Any, None]], + ) -> None: + """Run ``body`` and restart it on failure, bounded, then give up. + + A clean return means the loop observed ``self._stop`` — nothing to do. + ``CancelledError`` is re-raised so :meth:`stop` still works. Everything + else is logged with the loop name and retried per + :data:`_LOOP_RESTART_BACKOFF_SECONDS`; when those are exhausted the + process is asked to exit rather than run on with a dead loop. + + The budget counts **consecutive quick crashes**, not crashes over the + process lifetime: a body that ran at least + :data:`_LOOP_STABLE_RUN_SECONDS` before raising starts a fresh + incident. A deterministic crash-on-entry still exhausts the budget in + ~65s; independent transients days apart each get the full ladder. + """ + budget = len(_LOOP_RESTART_BACKOFF_SECONDS) + strikes = 0 + while True: + if strikes and await self._wait_or_stop( + _LOOP_RESTART_BACKOFF_SECONDS[strikes - 1] + ): + return + if self._stop.is_set(): + return + started = time.monotonic() + try: + await body() + return + except asyncio.CancelledError: + raise + except Exception as exc: + ran = time.monotonic() - started + if ran >= _LOOP_STABLE_RUN_SECONDS: + strikes = 0 + strikes += 1 + logger.exception( + "cascade_loop_crashed", + loop=loop_name, + strike=strikes, + restarts_left=max(0, budget - strikes + 1), + ran_seconds=round(ran, 1), + error=f"{type(exc).__name__}: {exc}", + ) + if strikes > budget: + break + logger.error("cascade_loop_unrecoverable", loop=loop_name, restarts=budget) + self._request_process_exit(loop_name) + + def _on_loop_task_done(self, loop_name: str, task: asyncio.Task[None]) -> None: + """Log a supervised loop task that ended without ``stop()`` asking it to.""" + if task.cancelled() or self._stop.is_set(): + return + exc = task.exception() + logger.error( + "cascade_loop_task_ended_unexpectedly", + loop=loop_name, + error=f"{type(exc).__name__}: {exc}" if exc is not None else None, + ) + + def _request_process_exit(self, loop_name: str) -> None: + """Ask this process to terminate so a supervisor can restart it. + + ``SIGTERM`` to our own pid rather than ``os._exit`` so the ASGI server + runs its graceful-shutdown path (lifespan shutdown, optimizer flush) + instead of dropping in-flight state on the floor. + + This assumes the deployment runs under something that restarts it — + systemd ``Restart=always``, Docker ``restart: unless-stopped``, a k8s + Deployment. Without one the process just stops, which is still the + better outcome: a live server whose projection pipeline is dead answers + searches from a silently frozen index. + + Overridable seam for tests — they replace this rather than signal the + pytest process. + """ + logger.error("cascade_worker_requesting_process_exit", loop=loop_name) + signal.raise_signal(signal.SIGTERM) + async def stop(self) -> None: if self._task is None: return @@ -709,11 +918,35 @@ class CascadeWorker: try: if initial_delay > 0 and await self._wait_or_stop(initial_delay): return - # Serialise behind any in-flight rebuild (rare; only during - # the 12h sweep). Failures are absorbed in _run_rebuild_once. + # Serialise behind any in-flight rebuild (rare; only during the + # 12h sweep). Failures are absorbed in _run_rebuild_once. + # + # Bounded, and symmetric with the wait on the other side: whichever + # of the two maintenance jobs arrives second parks on the first, so + # an unbounded wait here is the same defect as an unbounded wait + # there — this kind's task slot never frees, _schedule_optimize + # keeps short-circuiting on it, and that table silently stops being + # pruned. It was left unbounded on the argument that + # rebuild_indexes carries its own 300s deadline; that only covers + # the critical section, not the task's dispatch and teardown around + # it, so the transitive bound was never real. if state.rebuild_task is not None and not state.rebuild_task.done(): - with contextlib.suppress(Exception): - await state.rebuild_task + try: + async with asyncio.timeout(_MAINTENANCE_TASK_TIMEOUT_SECONDS): + await state.rebuild_task + except TimeoutError: + # Give up this beat rather than compact under a live + # rebuild — the two commit on the same manifest, which is + # what the wait exists to prevent. Writes keep the dirty + # flag set, so the next beat retries. + logger.warning( + "cascade_lancedb_optimize_skipped_rebuild_unfinished", + kind=kind, + waited_seconds=_MAINTENANCE_TASK_TIMEOUT_SECONDS, + ) + return + except Exception: + pass # _run_rebuild_once already logged and counted it while state.dirty and not self._stop.is_set(): state.dirty = False state.last_run_at = time.monotonic() @@ -801,6 +1034,7 @@ class CascadeWorker: await repo.optimize() if state is not None: state.optimize_failures = 0 + state.failures_since_fallback = 0 logger.debug( "cascade_lancedb_optimized", kind=kind, @@ -846,9 +1080,12 @@ class CascadeWorker: ) return failures = 0 + since_fallback = 0 if state is not None: state.optimize_failures += 1 + state.failures_since_fallback += 1 failures = state.optimize_failures + since_fallback = state.failures_since_fallback log = ( logger.error if failures >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD @@ -861,20 +1098,23 @@ class CascadeWorker: consecutive_failures=failures, error=f"{type(exc).__name__}: {exc}", ) - if failures >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD: + if since_fallback >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD: logger.info( "cascade_lancedb_optimize_fallback_rebuild", kind=kind, consecutive_failures=failures, ) await self._run_rebuild_once(kind) - # Reset even when rebuild fails: rate-limits fallback - # rebuild to at most once per threshold failures. A - # failed rebuild defers cleanup to the 12h periodic - # sweep — harmless for correctness (see - # _run_rebuild_once docstring). + # Reset the *rate limiter* even when the rebuild fails, so the + # fallback fires at most once per threshold failures. A failed + # rebuild defers cleanup to the 12h periodic sweep — harmless + # for correctness (see _run_rebuild_once docstring). + # + # ``optimize_failures`` is deliberately NOT reset here: it is + # the health signal, and zeroing it in the branch that fires at + # the threshold is what made the alert unreachable. if state is not None: - state.optimize_failures = 0 + state.failures_since_fallback = 0 async def _heartbeat_loop(self) -> None: """Periodic safety net for the optimizer. @@ -925,10 +1165,30 @@ class CascadeWorker: if self._stop.is_set(): return await self._run_rebuild_once(kind) + last_sweep = time.monotonic() + # Never coarser than the configured cadence: the tick exists to give + # conflict retries a 60s granularity against a 12h sweep, and must not + # quantise a deployment (or a test) that sets a shorter interval. + tick = min(_REBUILD_LOOP_TICK_SECONDS, self._optimize_rebuild_interval) while not self._stop.is_set(): - if await self._wait_or_stop(self._optimize_rebuild_interval): + if await self._wait_or_stop(tick): return + now = time.monotonic() + if now - last_sweep >= self._optimize_rebuild_interval: + last_sweep = now + for kind in self._handlers: + if self._stop.is_set(): + return + await self._run_rebuild_once(kind) + continue + # Between sweeps, serve only kinds whose conflict backoff is due. for kind in self._handlers: + state = self._optimizer_states.get(kind) + if state is None or not state.rebuild_retry_at: + continue + if now < state.rebuild_retry_at: + continue + state.rebuild_retry_at = 0.0 if self._stop.is_set(): return await self._run_rebuild_once(kind) @@ -960,8 +1220,32 @@ class CascadeWorker: and not state.task.done() and state.task is not asyncio.current_task() ): - with contextlib.suppress(Exception): - await state.task + try: + # Bounded: an optimize task that hangs must not park the rebuild + # loop behind it. Same defect class as the table-handle await — + # a wait with no deadline in a path a scheduler depends on. + async with asyncio.timeout(_MAINTENANCE_TASK_TIMEOUT_SECONDS): + await state.task + except TimeoutError: + # Skip this sweep rather than rebuild concurrently with an + # optimize that is still running: dropping indices under it is + # exactly the interleaving this wait exists to prevent. The 12h + # loop retries, and the prune-staleness signal covers the stall. + logger.warning( + "cascade_lancedb_rebuild_skipped_optimize_unfinished", + kind=kind, + waited_seconds=_MAINTENANCE_TASK_TIMEOUT_SECONDS, + ) + return + except Exception: + pass # the optimize runner already logged and counted it + # Retry a lost commit race in place. Lance labels it "Retryable" and + # means it: the rebuild transaction was preempted by a concurrent + # writer (another process, since the write lock is in-process only) and + # nothing about the table is wrong. Without a retry, a conflict costs a + # whole rebuild cadence — 12h in production — for what a second-scale + # backoff resolves. A soak run at 600s cadence hit 3 conflicts in 119 + # attempts (2.5%), all while a concurrent CLI storm was running. rebuild_task = asyncio.create_task( repo.rebuild_indexes(), name=f"cascade-rebuild-{kind}-inner" ) @@ -969,12 +1253,35 @@ class CascadeWorker: try: await rebuild_task logger.info("cascade_lancedb_rebuilt", kind=kind) + state.rebuild_attempt = 0 + state.rebuild_retry_at = 0.0 except Exception as exc: - logger.warning( - "cascade_lancedb_rebuild_failed", - kind=kind, - error=f"{type(exc).__name__}: {exc}", - ) + attempt = state.rebuild_attempt + if _is_benign_commit_conflict(exc) and attempt < len( + _REBUILD_CONFLICT_BACKOFFS_SECONDS + ): + # Lost the manifest race to a concurrent writer; the table is + # fine. Record a deadline instead of sleeping so the other + # kinds in this sweep are not parked behind the backoff. + delay = _REBUILD_CONFLICT_BACKOFFS_SECONDS[attempt] + state.rebuild_attempt = attempt + 1 + state.rebuild_retry_at = time.monotonic() + delay + logger.info( + "cascade_lancedb_rebuild_conflict_retry_scheduled", + kind=kind, + attempt=attempt, + retry_in_seconds=delay, + error=f"{type(exc).__name__}: {exc}", + ) + else: + state.rebuild_attempt = 0 + state.rebuild_retry_at = 0.0 + logger.warning( + "cascade_lancedb_rebuild_failed", + kind=kind, + attempt=attempt, + error=f"{type(exc).__name__}: {exc}", + ) finally: if state.rebuild_task is rebuild_task: state.rebuild_task = None diff --git a/tests/integration/test_cascade_scenarios.py b/tests/integration/test_cascade_scenarios.py index 6ae8aea..cda8fdb 100644 --- a/tests/integration/test_cascade_scenarios.py +++ b/tests/integration/test_cascade_scenarios.py @@ -24,7 +24,7 @@ from __future__ import annotations import asyncio import datetime as _dt import shutil -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Awaitable, Callable from pathlib import Path import anyio @@ -161,6 +161,52 @@ async def _wait_path_done(md_path: str, *, deadline: float = 15.0) -> None: await asyncio.sleep(0.05) +async def _wait_projection_quiescent( + md_path: str, + *, + counter: Callable[[str], Awaitable[int]], + deadline: float = 30.0, + samples: int = 3, + interval: float = 0.2, +) -> int: + """Wait until ``md_path``'s projection has stopped moving; return the count. + + Stronger than :func:`_wait_path_done`, and needed whenever the writer keeps + appending *while* the worker is mid-handler. There, a terminal row does not + mean the file is fully projected: the handler read the md at whatever length + it had, marked the row done, and the appends that landed during that call + are picked up by a *later* filesystem event. ``_wait_path_done``'s 0.1s + settle window is a bet that the re-enqueue has already been delivered — + which holds on macOS/fsevents and does not on a loaded Linux/inotify runner, + so the assertion fires against a half-projected table for a reason that has + nothing to do with the behaviour under test. + + Quiescence instead of a fixed settle: terminal row + empty pending queue + + a projected count unchanged across ``samples`` consecutive polls. Real loss + still fails the caller's assertion — the count simply converges below the md + entry count and stays there. + """ + async with asyncio.timeout(deadline): + await _wait_path_done(md_path, deadline=deadline) + stable = 0 + last = await counter(md_path) + while True: + await asyncio.sleep(interval) + row = await md_change_state_repo.get_by_id(md_path) + summary = await md_change_state_repo.queue_summary() + current = await counter(md_path) + settled = ( + row is not None + and row.status in ("done", "failed") + and summary.pending == 0 + and current == last + ) + stable = stable + 1 if settled else 0 + last = current + if stable >= samples: + return current + + async def _wait_paths_done(*md_paths: str, deadline: float = 15.0) -> None: await asyncio.gather(*[_wait_path_done(p, deadline=deadline) for p in md_paths]) @@ -564,10 +610,14 @@ async def test_lap_append_during_handler_no_loss( md_path = _atomic_fact_md_path(owner_id, bucket) absolute = memory_root.root / md_path - await _wait_path_done(md_path, deadline=30.0) + # Quiescence, not just a terminal row: the appends that landed during a + # handler invocation are projected by a *later* filesystem event, so a + # done row can coexist with a half-projected table. + lance_rows = await _wait_projection_quiescent( + md_path, counter=_count_lance_rows_md, deadline=30.0 + ) md_entries = await _count_md_entries(absolute) - lance_rows = await _count_lance_rows_md(md_path) assert md_entries == total, ( f"writer self-check: expected {total} md entries, got {md_entries}" ) 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 36fb025..e2507ad 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 @@ -17,7 +17,10 @@ from __future__ import annotations import asyncio import datetime as dt +import os +import time from pathlib import Path +from types import SimpleNamespace from typing import ClassVar import pytest @@ -736,6 +739,12 @@ async def test_prune_holds_write_lock_and_is_cross_process_safe( async def uri(self) -> str: return str(tmp_path) + async def list_indices(self): # type: ignore[no-untyped-def] + # Mirrors the real signature: prune reads live index UUIDs so the + # husk sweep can spare them. A double that omits it hides a real + # TypeError behind a green test. + return [] + repo = _NoteRepo(table=_MockTable()) # type: ignore[arg-type] await repo.prune(dt.timedelta(seconds=42)) @@ -903,27 +912,56 @@ def test_prune_timeout_is_well_below_the_prune_cadence() -> None: ) -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.""" - indices = tmp_path / "_indices" - (indices / "empty_uuid").mkdir(parents=True) - populated = indices / "live_uuid" - populated.mkdir() - (populated / "index.idx").write_text("data") +async def test_hanging_reads_also_hit_a_deadline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reads need a deadline too, for a different reason than writes. - class _MockTable: - async def optimize(self, **_kw): - return None + A read takes no lock, so a hung read blocks no writer — which is why the + write-side deadline work skipped them. But the cascade drain loop reads on + every batch (``handlers/_daily_log_base.py`` calls ``find_where``) and + advances strictly one batch at a time, so a read that never returns stops + the whole md -> LanceDB projection: claimed rows stay ``processing`` + forever, nothing new is indexed, and ``/health`` still reports healthy + because a hang raises nothing and the drain-failure counter only counts + exceptions. + """ + from everos.core.errors import VectorStoreBusyError + from everos.core.persistence.lancedb import repository as repo_mod - async def uri(self) -> str: - return str(tmp_path) + monkeypatch.setattr(repo_mod, "_READ_TIMEOUT_SECONDS", 0.05) - repo = _NoteRepo(table=_MockTable()) # type: ignore[arg-type] - await repo.prune(dt.timedelta(seconds=1)) + class _HangingLookupRepo(_NoteRepo): + async def _table_lookup(self): # type: ignore[no-untyped-def] + await asyncio.sleep(30) - assert not (indices / "empty_uuid").exists(), "empty husk must be removed" - assert populated.exists(), "non-empty index dir must be kept" + repo = _HangingLookupRepo() + + with pytest.raises(VectorStoreBusyError): + await repo.count() + with pytest.raises(VectorStoreBusyError): + await repo.get_by_id("u1_n1") + with pytest.raises(VectorStoreBusyError): + await repo.find_where("owner_id = 'u1'") + with pytest.raises(VectorStoreBusyError): + await repo.find_where_paginated("owner_id = 'u1'", sort_by="entry_id") + with pytest.raises(VectorStoreBusyError): + await repo.search(vector=None, where=None, limit=1) + + +def test_read_budget_is_generous_enough_never_to_fire_on_a_healthy_read() -> None: + """The read deadline is a hang-catcher, not a latency SLO. + + everos builds no vector ANN index, so every read is a flat scan — measured + ~62ms over 117k rows. The budget must stay far above any real scan so it + cannot turn a large-but-working table into a stream of failures. + """ + from everos.core.persistence.lancedb.repository import _READ_TIMEOUT_SECONDS + + measured_seconds_per_117k_rows = 0.062 + assert measured_seconds_per_117k_rows * 500 <= _READ_TIMEOUT_SECONDS, ( + "read budget must leave ~500x headroom over the measured flat scan" + ) async def test_optimize_is_lock_free_compaction_only() -> None: @@ -943,3 +981,208 @@ async def test_optimize_is_lock_free_compaction_only() -> None: assert state["held"] is False, "light optimize must be lock-free" assert captured == {}, "light optimize passes no cleanup/delete_unverified args" + + +async def test_rebuild_never_leaves_the_column_without_an_fts_index( + tmp_path: Path, +) -> None: + """A BM25 query must survive a rebuild — the index is replaced, not dropped. + + Vector search degrades to a flat scan when its index is missing; FTS does + not. With no inverted index lance raises ``Cannot perform full text search + unless an INVERTED index has been created``, and because the recall legs + are gathered without ``return_exceptions`` that one failing leg 500s the + whole search request. So the drop-then-create rebuild had a window where + every keyword search on that kind failed. Hammering the table across a + rebuild is the only way to catch a regression here — a unit double cannot + reproduce it. + """ + + class _SearchRepo(LanceRepoBase[_SearchNote]): + schema = _SearchNote + + mr = MemoryRoot(tmp_path) + mr.ensure() + conn = await open_lancedb_connection(mr.lancedb_dir, LanceDBSettings()) + table = await conn.create_table("_search_note", schema=_SearchNote) + repo = _SearchRepo(table=table) + await repo.add( + [ + _SearchNote( + id=f"n{i}", + text=f"meeting notes alpha {i}", + tokens=f"meeting notes alpha {i}", + vector=[1.0, 0.0, 0.0, 0.0], + ) + for i in range(200) + ] + ) + await _SearchNote.ensure_fts_indexes(table) + + failures: list[str] = [] + successes = 0 + stop = False + + async def _hammer() -> None: + nonlocal successes + while not stop: + try: + await table.query().nearest_to_text("alpha").limit(3).to_list() + successes += 1 + except Exception as exc: + failures.append(f"{type(exc).__name__}: {exc}") + await asyncio.sleep(0) + + hammer = asyncio.create_task(_hammer()) + try: + for _ in range(3): + await repo.rebuild_indexes() + finally: + stop = True + await hammer + + assert successes > 0, "the hammer never ran; the test proves nothing" + assert not failures, ( + f"{len(failures)} keyword queries failed across the rebuild — the index " + f"went missing. First: {failures[0]}" + ) + # And the rebuild still did its job: the column is indexed afterwards. + indexed = {c for i in await table.list_indices() for c in (i.columns or [])} + assert "tokens" in indexed + + +def _aged(path: Path, seconds: float) -> Path: + """Backdate ``path`` so the sweep's age gate accepts it.""" + old = time.time() - seconds + os.utime(path, (old, old)) + return path + + +class _SweepTable: + """Prune-path double: no-op optimize, a uri(), and a live index list.""" + + def __init__(self, uri: str, live: tuple[str, ...] = ()) -> None: + self._uri = uri + self._live = live + + async def optimize(self, **_kw): # type: ignore[no-untyped-def] + return None + + async def uri(self) -> str: + return self._uri + + async def list_indices(self): # type: ignore[no-untyped-def] + return [SimpleNamespace(index_uuid=u) for u in self._live] + + +async def test_husk_sweep_only_takes_dead_empty_old_dirs(tmp_path: Path) -> None: + """Every case the sweep must refuse, in one pass. + + lance's cleanup unlinks an index's files but never its directory (there is + no ``rmdir`` anywhere in ``cleanup.rs`` — it targets object stores, where an + empty directory is not a thing), so on a local filesystem the husks pile up: + 13061 dirs in one soak run, 98% empty. Sweeping them is ours to do, which + means the refusals are what needs pinning down. + """ + from everos.core.persistence.lancedb.repository import _HUSK_MIN_AGE_SECONDS + + indices = tmp_path / "_indices" + old = _HUSK_MIN_AGE_SECONDS * 2 + + dead = indices / "dead-uuid" + dead.mkdir(parents=True) + live = indices / "live-uuid" + fresh = indices / "being-built-right-now" + populated = indices / "has-files" + for d in (live, fresh, populated): + d.mkdir() + (populated / "part_0").write_text("index data") + for d in (dead, live, populated): + _aged(d, old) # only `fresh` keeps its current mtime + + repo = _NoteRepo(table=_SweepTable(str(tmp_path), live=("live-uuid",))) # type: ignore[arg-type] + await repo.prune(dt.timedelta(seconds=1)) + + assert not dead.exists(), "a dead, empty, aged husk is the whole point" + assert live.exists(), ( + "a UUID still in list_indices() must be spared even when its dir looks " + "empty and old" + ) + assert fresh.exists(), ( + "a dir younger than lance's own 7-day unverified threshold may be an " + "index build in progress" + ) + assert populated.exists() and (populated / "part_0").exists(), ( + "rmdir is refused by the kernel on a non-empty dir — no file can ever " + "be lost here" + ) + + +def test_husk_age_gate_matches_lance_own_threshold() -> None: + """The age bound is lance's number, not one we picked. + + ``cleanup.rs`` sets ``UNVERIFIED_THRESHOLD_DAYS = 7`` and applies it to + exactly this judgement: an index UUID no manifest references is only assumed + dead once it is that old, because until then it is indistinguishable from an + in-progress build. Matching it means this sweep can never be more aggressive + than lance itself. An earlier version used 300s — our own invention, and the + reason the sweep was not defensible. + """ + from everos.core.persistence.lancedb.repository import _HUSK_MIN_AGE_SECONDS + + assert _HUSK_MIN_AGE_SECONDS == 7 * 24 * 60 * 60.0 + + +async def test_husk_sweep_timeout_must_not_fail_the_prune( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A sweep that blows its deadline is the sweep's problem, not prune's. + + By the time the sweep runs, the cleanup commit — the thing prune exists + for — has already succeeded, and the sweep is best-effort by contract. + Letting its timeout escape ``prune()`` bills the failure to the wrong + account: the optimize scheduler counts a prune failure (feeding the + fallback-rebuild threshold) and the prune-staleness clock stops + advancing, so both alarms report a cleanup stall that did not happen. + Not a theoretical path either — sweep time is proportional to dir count + (~35us/dir measured), and the ceiling-load steady state sits right at + the sweep budget. + """ + from everos.core.persistence.lancedb import repository as repo_mod + + monkeypatch.setattr(repo_mod, "_HUSK_SWEEP_TIMEOUT_SECONDS", 0.05) + + def _slow_sweep(table_uri, *, live_uuids, min_age_seconds): # type: ignore[no-untyped-def] + time.sleep(0.5) # to_thread cancellation cannot interrupt this + return 0 + + monkeypatch.setattr(repo_mod, "_remove_empty_index_dirs", _slow_sweep) + + repo = _NoteRepo(table=_SweepTable(str(tmp_path))) # type: ignore[arg-type] + await repo.prune(dt.timedelta(seconds=1)) # must not raise + + +async def test_a_broken_sweep_still_surfaces( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Only the *timeout* is absorbed — a real fault in the sweep must escape. + + The sibling test pins that a deadline miss does not bill prune. The risk on + the other side is the catch drifting wider: swallowing every exception + would turn a genuine bug in ``_remove_empty_index_dirs`` (a TypeError after + a signature change, a permission error on the index dir) into a silent + ``removed = 0``, and nothing anywhere would say the sweep had stopped + working. That is the exact failure shape this module keeps being audited + for, so the narrowness of the catch is worth a test of its own — widening + it to ``except Exception`` passes every other test in this file. + """ + from everos.core.persistence.lancedb import repository as repo_mod + + def _broken_sweep(table_uri, *, live_uuids, min_age_seconds): # type: ignore[no-untyped-def] + raise TypeError("sweep signature drifted") + + monkeypatch.setattr(repo_mod, "_remove_empty_index_dirs", _broken_sweep) + + repo = _NoteRepo(table=_SweepTable(str(tmp_path))) # type: ignore[arg-type] + with pytest.raises(TypeError, match="signature drifted"): + await repo.prune(dt.timedelta(seconds=1)) diff --git a/tests/unit/test_core/test_persistence/test_locking.py b/tests/unit/test_core/test_persistence/test_locking.py index 62a2ff0..bd4ea56 100644 --- a/tests/unit/test_core/test_persistence/test_locking.py +++ b/tests/unit/test_core/test_persistence/test_locking.py @@ -94,3 +94,127 @@ async def test_blocking_waits_for_release(tmp_path: Path) -> None: proc.join(timeout=5) if proc.is_alive(): proc.terminate() + + +async def test_blocking_wait_is_bounded_and_logged( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A held lock must produce a bounded, *visible* wait — not a silent hang. + + The wait itself is correct by design (the second process is supposed to + wait, then find the migration already done). What is not correct is doing it + with no upper bound and no log line: an ``everos server start`` that lands + on a held lock looks like a hang whose last message is + ``lifespan_provider_startup name=lancedb``. + """ + from everos.core.persistence import locking + + events: list[str] = [] + + class _SpyLogger: + def __getattr__(self, _level: str): # type: ignore[no-untyped-def] + def rec(event: str, **_kw) -> None: # type: ignore[no-untyped-def] + events.append(event) + + return rec + + monkeypatch.setattr(locking, "logger", _SpyLogger()) + monkeypatch.setattr(locking, "_LOCK_POLL_INTERVAL_SECONDS", 0.01) + + mr = MemoryRoot(tmp_path) + ctx = multiprocessing.get_context("spawn") + ready = ctx.Event() + release = ctx.Event() + proc = ctx.Process(target=_hold_lock, args=(str(mr.root), ready, release)) + proc.start() + try: + assert ready.wait(timeout=5) + started = time.monotonic() + with pytest.raises(LockError, match="timed out"): + async with memory_root_lock(mr, timeout_seconds=0.2): + pass + elapsed = time.monotonic() - started + assert 0.2 <= elapsed < 5.0, f"must give up near the budget, took {elapsed}s" + assert "memory_root_lock_waiting" in events, ( + "waiting on another process must be announced, or the startup " + "stall has no explanation in the log" + ) + finally: + release.set() + proc.join(timeout=5) + if proc.is_alive(): + proc.terminate() + + # The timed-out attempt must not have left a lock behind. Acquisition polls + # with LOCK_NB precisely so a give-up cannot leave a worker thread blocked + # in ``flock`` that later acquires the lock with nobody left to release it. + with anyio.fail_after(2): + async with memory_root_lock(mr, timeout_seconds=1.0): + pass + + +async def test_successful_wait_logs_how_long_it_waited( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """When the wait does succeed, say so — silence is what made it opaque.""" + import threading + + from everos.core.persistence import locking + + events: list[str] = [] + + class _SpyLogger: + def __getattr__(self, _level: str): # type: ignore[no-untyped-def] + def rec(event: str, **_kw) -> None: # type: ignore[no-untyped-def] + events.append(event) + + return rec + + monkeypatch.setattr(locking, "logger", _SpyLogger()) + monkeypatch.setattr(locking, "_LOCK_POLL_INTERVAL_SECONDS", 0.01) + + mr = MemoryRoot(tmp_path) + ctx = multiprocessing.get_context("spawn") + ready = ctx.Event() + release = ctx.Event() + proc = ctx.Process(target=_hold_lock, args=(str(mr.root), ready, release)) + proc.start() + try: + assert ready.wait(timeout=5) + threading.Timer(0.2, release.set).start() + async with memory_root_lock(mr, timeout_seconds=5.0): + pass + assert events == [ + "memory_root_lock_waiting", + "memory_root_lock_acquired_after_wait", + ] + finally: + release.set() + proc.join(timeout=5) + if proc.is_alive(): + proc.terminate() + + +async def test_uncontended_acquisition_stays_silent(tmp_path: Path) -> None: + """No contention → no log noise. The wait lines must be signal, not chatter + on the hot startup path.""" + from everos.core.persistence import locking + + events: list[str] = [] + + class _SpyLogger: + def __getattr__(self, _level: str): # type: ignore[no-untyped-def] + def rec(event: str, **_kw) -> None: # type: ignore[no-untyped-def] + events.append(event) + + return rec + + monkeypatched = pytest.MonkeyPatch() + monkeypatched.setattr(locking, "logger", _SpyLogger()) + try: + async with memory_root_lock(MemoryRoot(tmp_path)): + pass + finally: + monkeypatched.undo() + + assert events == [] diff --git a/tests/unit/test_memory/test_cascade/test_handler_agent_case.py b/tests/unit/test_memory/test_cascade/test_handler_agent_case.py new file mode 100644 index 0000000..9f79cc5 --- /dev/null +++ b/tests/unit/test_memory/test_cascade/test_handler_agent_case.py @@ -0,0 +1,294 @@ +"""Tests for :class:`AgentCaseHandler` — md -> LanceDB ``agent_case`` row. + +The kind this file covers had no handler test of its own, and the storage soak +never writes it either (four of the seven business tables stay at zero rows +there), so the md -> row contract for agent cases was going unexercised end to +end. It is a daily-log handler like episode, but with three differences worth +pinning: it lives on the agent track (``agents//.cases/``), it has no +``sender_ids`` column, and it embeds ``task_intent`` only — ``approach`` is +BM25-indexed but deliberately never sent to the embedder. + +Uses a real on-disk md file via :class:`AgentCaseWriter`; the LanceDB repo is +faked so the test stays in-memory while still checking row construction and the +diff branches. +""" + +from __future__ import annotations + +import datetime as _dt +from pathlib import Path + +import pytest + +from everos.component.embedding import EmbeddingCapability, EmbeddingProvider +from everos.component.tokenizer import Tokenizer +from everos.core.persistence import MemoryRoot +from everos.infra.persistence.lancedb import AgentCase +from everos.infra.persistence.markdown import AgentCaseWriter +from everos.memory.cascade.handlers import HandlerDeps +from everos.memory.cascade.handlers.agent_case import AgentCaseHandler + +_AGENT = "a_ops" +_DAY = _dt.date(2026, 5, 14) +_MD = f"default_app/default_project/agents/{_AGENT}/.cases/agent_case-2026-05-14.md" + + +class _StubTokenizer(Tokenizer): + def tokenize(self, text: str) -> list[str]: + return [tok for tok in text.split() if tok] + + def tokenize_batch(self, texts): # type: ignore[no-untyped-def] + return [self.tokenize(t) for t in texts] + + +class _StubEmbedder(EmbeddingProvider): + dim = 1024 + + def __init__(self) -> None: + self.calls: list[str] = [] + + async def embed(self, text: str) -> list[float]: + self.calls.append(text) + return [0.1] * self.dim + + async def embed_batch(self, texts): # type: ignore[no-untyped-def] + return [await self.embed(t) for t in texts] + + +class _FakeAgentCaseRepo: + def __init__(self) -> None: + self.upserts: list[list[AgentCase]] = [] + self.deletes: list[str] = [] + self.rows: list[AgentCase] = [] + + async def find_where(self, where: str, *, limit: int = 100) -> list[AgentCase]: + prefix = "md_path = '" + if where.startswith(prefix): + md_path = where[len(prefix) :].rstrip("'") + return [r for r in self.rows if r.md_path == md_path] + return [] + + async def upsert(self, rows: list[AgentCase]) -> None: + self.upserts.append(list(rows)) + by_id = {r.id: r for r in self.rows} + for r in rows: + by_id[r.id] = r + self.rows = list(by_id.values()) + + async def delete(self, predicate: str) -> None: + self.deletes.append(predicate) + + async def delete_by_md_path(self, md_path: str) -> int: + before = len(self.rows) + self.rows = [r for r in self.rows if r.md_path != md_path] + return before - len(self.rows) + + +@pytest.fixture +def memory_root(tmp_path: Path) -> MemoryRoot: + mr = MemoryRoot(tmp_path) + mr.ensure() + return mr + + +@pytest.fixture +def stub_embedder(monkeypatch: pytest.MonkeyPatch) -> _StubEmbedder: + import everos.component.embedding.accessor as acc + + embedder = _StubEmbedder() + monkeypatch.setattr(acc, "_capability", EmbeddingCapability(provider=embedder)) + return embedder + + +@pytest.fixture +def no_embedder(monkeypatch: pytest.MonkeyPatch) -> None: + """Embedding unavailable — the soft-dependency path.""" + import everos.component.embedding.accessor as acc + + monkeypatch.setattr(acc, "_capability", EmbeddingCapability(provider=None)) + + +@pytest.fixture +def fake_repo(monkeypatch: pytest.MonkeyPatch) -> _FakeAgentCaseRepo: + repo = _FakeAgentCaseRepo() + monkeypatch.setattr(AgentCaseHandler, "lance_repo", repo) + return repo + + +async def _write_entry( + writer: AgentCaseWriter, + *, + intent: str = "restart the ingest worker", + approach: str = "drain the queue then bounce the unit", + key_insight: str | None = "check the lock first", + quality: float = 0.8, +) -> str: + sections: dict[str, str] = {"TaskIntent": intent, "Approach": approach} + if key_insight is not None: + sections["KeyInsight"] = key_insight + await writer.append_entry( + _AGENT, + inline={ + "owner_id": _AGENT, + "session_id": "s1", + "timestamp": "2026-05-14T10:00:00+00:00", + "parent_type": "memcell", + "parent_id": "mc_case_parent", + "quality_score": quality, + }, + sections=sections, + date=_DAY, + ) + return _MD + + +def _handler(memory_root: MemoryRoot) -> AgentCaseHandler: + return AgentCaseHandler( + HandlerDeps(memory_root=memory_root, tokenizer=_StubTokenizer()) + ) + + +async def test_added_entry_builds_the_agent_track_row( + memory_root: MemoryRoot, + fake_repo: _FakeAgentCaseRepo, + stub_embedder: _StubEmbedder, +) -> None: + """One md entry -> one typed row, with the agent-track fields set.""" + await _write_entry(AgentCaseWriter(memory_root)) + + outcome = await _handler(memory_root).handle_added_or_modified(_MD) + + assert (outcome.upserted, outcome.deleted, outcome.skipped) == (1, 0, 0) + row = fake_repo.upserts[0][0] + assert row.owner_id == _AGENT + assert row.owner_type == "agent" + assert row.id.startswith(f"{_AGENT}_") + assert row.session_id == "s1" + assert row.parent_type == "memcell" + assert row.parent_id == "mc_case_parent" + assert row.quality_score == pytest.approx(0.8) + assert row.task_intent == "restart the ingest worker" + assert row.approach == "drain the queue then bounce the unit" + assert row.key_insight == "check the lock first" + assert row.md_path == _MD + assert row.content_sha256 + + +async def test_only_task_intent_is_embedded( + memory_root: MemoryRoot, + fake_repo: _FakeAgentCaseRepo, + stub_embedder: _StubEmbedder, +) -> None: + """``approach`` is BM25-indexed but must never reach the embedder. + + The retrieval anchor for a case is its intent; the approach text is long + step-by-step prose, so embedding it would both cost tokens and blur the + vector. Both fields still get tokenised for keyword recall — asserting on + the tokens as well keeps this from passing for the wrong reason (a field + that stopped being indexed at all would also stop being embedded). + """ + await _write_entry(AgentCaseWriter(memory_root)) + + await _handler(memory_root).handle_added_or_modified(_MD) + + assert stub_embedder.calls == ["restart the ingest worker"] + row = fake_repo.upserts[0][0] + assert row.task_intent_tokens == "restart the ingest worker" + assert row.approach_tokens == "drain the queue then bounce the unit" + + +async def test_missing_embedding_still_writes_a_searchable_row( + memory_root: MemoryRoot, + fake_repo: _FakeAgentCaseRepo, + no_embedder: None, +) -> None: + """Embedding is a soft dependency: no provider -> ``vector=None``, row kept. + + This is the tier-1 (keyword-only) deployment, so the row must still land + with its BM25 columns populated rather than the entry being dropped. + """ + await _write_entry(AgentCaseWriter(memory_root)) + + outcome = await _handler(memory_root).handle_added_or_modified(_MD) + + assert outcome.upserted == 1 + row = fake_repo.upserts[0][0] + assert row.vector is None + assert row.task_intent_tokens + + +async def test_optional_key_insight_may_be_absent( + memory_root: MemoryRoot, + fake_repo: _FakeAgentCaseRepo, + stub_embedder: _StubEmbedder, +) -> None: + """``KeyInsight`` is optional in the md contract — absence is not an error.""" + await _write_entry(AgentCaseWriter(memory_root), key_insight=None) + + outcome = await _handler(memory_root).handle_added_or_modified(_MD) + + assert outcome.upserted == 1 + assert fake_repo.upserts[0][0].key_insight is None + + +async def test_unchanged_entry_is_skipped_not_re_embedded( + memory_root: MemoryRoot, + fake_repo: _FakeAgentCaseRepo, + stub_embedder: _StubEmbedder, +) -> None: + """Re-processing an untouched file must short-circuit on ``content_sha256``. + + Without this the 30s scanner sweep would re-embed every case on every pass. + """ + await _write_entry(AgentCaseWriter(memory_root)) + handler = _handler(memory_root) + await handler.handle_added_or_modified(_MD) + embeds_after_first = len(stub_embedder.calls) + + outcome = await handler.handle_added_or_modified(_MD) + + assert (outcome.upserted, outcome.skipped) == (0, 1) + assert len(stub_embedder.calls) == embeds_after_first, "must not re-embed" + + +async def test_edited_entry_re_upserts( + memory_root: MemoryRoot, + fake_repo: _FakeAgentCaseRepo, + stub_embedder: _StubEmbedder, +) -> None: + """A content edit must flip the digest and re-upsert, not skip.""" + writer = AgentCaseWriter(memory_root) + await _write_entry(writer) + handler = _handler(memory_root) + await handler.handle_added_or_modified(_MD) + first_sha = fake_repo.upserts[0][0].content_sha256 + + md_file = memory_root.root / _MD + md_file.write_text( + md_file.read_text().replace( + "restart the ingest worker", "restart the ingest worker twice" + ) + ) + outcome = await handler.handle_added_or_modified(_MD) + + assert outcome.upserted == 1 + row = fake_repo.upserts[-1][0] + assert row.task_intent == "restart the ingest worker twice" + assert row.content_sha256 != first_sha + + +async def test_deleted_md_removes_every_row_for_that_path( + memory_root: MemoryRoot, + fake_repo: _FakeAgentCaseRepo, + stub_embedder: _StubEmbedder, +) -> None: + """Deleting the md must clear its rows — otherwise search serves ghosts.""" + await _write_entry(AgentCaseWriter(memory_root)) + handler = _handler(memory_root) + await handler.handle_added_or_modified(_MD) + assert fake_repo.rows + + outcome = await handler.handle_deleted(_MD) + + assert outcome.deleted == 1 + assert not fake_repo.rows diff --git a/tests/unit/test_memory/test_cascade/test_orchestrator.py b/tests/unit/test_memory/test_cascade/test_orchestrator.py index a2ec8ce..0c57152 100644 --- a/tests/unit/test_memory/test_cascade/test_orchestrator.py +++ b/tests/unit/test_memory/test_cascade/test_orchestrator.py @@ -157,3 +157,64 @@ async def test_operational_signal_flips_healthy( health = await orch.health() assert health.healthy is False assert any("cleanup stalled" in r for r in health.reasons) + + +async def test_maintenance_cadences_reach_the_worker_from_settings( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """``[cascade]`` settings must actually land on the worker. + + These four were constructor-only for long enough that the 12h rebuild sweep + could not be exercised by any soak run shorter than half a day — the gap was + not a missing parameter but a config layer that dropped it. Asserting on the + worker's own attributes rather than on ``CascadeConfig`` is the point: a + field that stops being forwarded still passes a config-level check. + """ + from everos.config import load_settings + from everos.memory.cascade.orchestrator import CascadeConfig, CascadeOrchestrator + + monkeypatch.setenv("EVEROS_CASCADE__OPTIMIZE_HEARTBEAT_SECONDS", "11") + monkeypatch.setenv("EVEROS_CASCADE__OPTIMIZE_PRUNE_INTERVAL_SECONDS", "22") + monkeypatch.setenv("EVEROS_CASCADE__OPTIMIZE_PRUNE_RETENTION_SECONDS", "33") + monkeypatch.setenv("EVEROS_CASCADE__OPTIMIZE_REBUILD_INTERVAL_SECONDS", "44") + load_settings.cache_clear() # type: ignore[attr-defined] + try: + cfg = CascadeConfig.from_settings() + assert ( + cfg.optimize_heartbeat_seconds, + cfg.optimize_prune_interval_seconds, + cfg.optimize_prune_retention_seconds, + cfg.optimize_rebuild_interval_seconds, + ) == (11.0, 22.0, 33.0, 44.0) + + orch = CascadeOrchestrator( + memory_root=MemoryRoot.resolve(), tokenizer=build_tokenizer(), config=cfg + ) + worker = orch._worker + assert worker._optimize_heartbeat == 11.0 + assert worker._optimize_prune_interval == 22.0 + assert worker._optimize_prune_retention == 33.0 + assert worker._optimize_rebuild_interval == 44.0 + finally: + load_settings.cache_clear() # type: ignore[attr-defined] + + +def test_deadlines_are_deliberately_not_configurable() -> None: + """The hang-catchers must stay constants, not settings. + + A read/write/prune deadline is sized from a measured duration: too low + manufactures failures on a healthy table, too high leaves a wedged one + invisible for longer. Neither end is a tuning preference, so exposing them + invites a change that can only make things worse. Cadences are the opposite + — they depend on write volume — which is why only those moved. + """ + from everos.config import CascadeSettings + + exposed = set(CascadeSettings.model_fields) + assert exposed == { + "optimize_heartbeat_seconds", + "optimize_prune_interval_seconds", + "optimize_prune_retention_seconds", + "optimize_rebuild_interval_seconds", + } + assert not any("timeout" in f or "deadline" in f for f in exposed) diff --git a/tests/unit/test_memory/test_cascade/test_worker.py b/tests/unit/test_memory/test_cascade/test_worker.py index 6d588d1..f60fbbb 100644 --- a/tests/unit/test_memory/test_cascade/test_worker.py +++ b/tests/unit/test_memory/test_cascade/test_worker.py @@ -783,11 +783,14 @@ async def test_optimize_failures_counted_escalated_and_reset( ) -> None: """Layer-2 stop-gap for lance-format/lance#7653. - Consecutive ``optimize()`` failures are counted, escalate - warning→error once the threshold is hit, and reset to 0 when: - (a) a rebuild is triggered on sustained failures, or - (b) the next optimize succeeds — instead of being swallowed as a - silent warning stream that lets the index dir grow until the disk fills. + Consecutive ``optimize()`` failures are counted, escalate warning→error once + the threshold is hit, and reset to 0 **only** when the next optimize + succeeds — not by the fallback rebuild the threshold triggers. Zeroing the + alert counter inside the branch that fires at the threshold made + ``optimize_failure_streak >= threshold`` effectively unobservable: the + counter cycled 1..threshold -> 0 -> 1.. and the only window where a poller + could see the threshold value was the sub-second rebuild itself. The rate + limiter lives in ``failures_since_fallback`` instead. """ from everos.memory.cascade import worker as wmod @@ -821,18 +824,24 @@ async def test_optimize_failures_counted_escalated_and_reset( fail_logs = [lvl for lvl, ev in calls if ev == "cascade_lancedb_optimize_failed"] assert fail_logs == ["warning"] * (threshold - 1) - # One more failure triggers rebuild and resets counter to 0. + # One more failure triggers the fallback rebuild. The alert counter must + # keep climbing across it — that is what the health verdict reads. await w._run_optimize_once("episode") - assert state.optimize_failures == 0 + assert state.optimize_failures == threshold, ( + "the fallback rebuild must not reset the alert counter — doing so is " + "what made the threshold unreachable" + ) + assert state.failures_since_fallback == 0, "the rate limiter resets, not the alert" rebuild_logs = [ lvl for lvl, ev in calls if ev == "cascade_lancedb_optimize_fallback_rebuild" ] assert len(rebuild_logs) == 1 - # A subsequent success keeps the counter at 0. + # A success is the only thing that clears either counter. repo.fail = False await w._run_optimize_once("episode") assert state.optimize_failures == 0 + assert state.failures_since_fallback == 0 async def test_optimize_fallback_rebuild_on_sustained_failure( @@ -840,9 +849,10 @@ async def test_optimize_fallback_rebuild_on_sustained_failure( ) -> None: """Consecutive optimize failures >= threshold trigger a fallback rebuild. - The rebuild drops + recreates indexes, bypassing the Rust panic path. - After rebuild (success or failure), the failure counter resets to 0 - to avoid triggering rebuild on every subsequent optimize tick. + The rebuild drops + recreates indexes, bypassing the Rust panic path. The + rate limiter (``failures_since_fallback``) resets after the rebuild whether + it succeeded or not, so the fallback fires at most once per threshold + failures rather than on every subsequent tick. """ from everos.memory.cascade import worker as wmod @@ -859,10 +869,51 @@ async def test_optimize_fallback_rebuild_on_sustained_failure( await w._run_optimize_once("episode") state = w._optimizer_states["episode"] - assert state.optimize_failures == 0, "rebuild should reset failure counter" + assert state.failures_since_fallback == 0, "rebuild resets the rate limiter" assert len(repo.rebuild_calls) == 1, "exactly one fallback rebuild expected" +async def test_persistent_optimize_failure_stays_visible_in_health( + patched_repo: _FakeRepo, +) -> None: + """A table failing 100% of the time must reach the health threshold. + + Regression guard for a reachability bug, not a counting bug: the fallback + rebuild used to zero the same counter the health verdict reads, so the + threshold value existed only during the sub-second rebuild. Against a 30s + scrape that is ~1% observable, i.e. ``cascade.healthy`` — the field + operators are told to alert on — stayed green while the table never + reclaimed a version. Sibling of the run7 cross-kind ``max()`` masking bug: + a remediation path refreshing the very signal meant to report it. + + Driven past the threshold on purpose: the point is that the streak survives + the remediation, so the check has to run *after* a fallback has fired. + """ + from everos.memory.cascade import worker as wmod + + repo = _OptimizeFailingRepo() + w = CascadeWorker( + {"episode": _OkHandlerWithRepo(repo)}, + retry_backoff_seconds=0, + optimize_min_interval_seconds=0.05, + ) + w._optimizer_states["episode"] = wmod._KindOptimizerState() + + threshold = wmod._OPTIMIZE_FAILURE_ALERT_THRESHOLD + for _i in range(threshold * 2 + 1): + await w._run_optimize_once("episode") + + assert len(repo.rebuild_calls) == 2, ( + "fallback stays rate-limited to once per threshold failures" + ) + health = w.health() + assert health.optimize_failure_streak >= threshold + reasons = health.reasons() + assert any("optimize" in r for r in reasons), ( + f"health must name the stuck optimize; got {reasons}" + ) + + # ── health signals ─────────────────────────────────────────────────────────── @@ -1165,3 +1216,216 @@ async def test_non_conflict_failure_counts_even_when_message_says_retryable( "an error whose message merely contains 'retryable' is NOT a commit " "conflict and must count toward the streak" ) + + +# ── background-loop supervision ────────────────────────────────────────────── + + +async def test_a_crashed_loop_is_restarted_instead_of_dying_silently( + patched_repo: _FakeRepo, +) -> None: + """A background loop that raises must be restarted, not lost. + + The three long-lived loops are plain ``create_task`` coroutines. Without + supervision one uncaught exception ends that loop permanently: nothing + restarts it, and because the worker keeps a strong reference to the task the + interpreter never prints "Task exception was never retrieved" either (that + fires on GC). The loop's job just stops happening, with zero output. + """ + from everos.memory.cascade import worker as wmod + + w = CascadeWorker({"episode": _OkHandler()}, retry_backoff_seconds=0) + monkeypatched = pytest.MonkeyPatch() + monkeypatched.setattr(wmod, "_LOOP_RESTART_BACKOFF_SECONDS", (0.0, 0.0, 0.0)) + + runs = 0 + + async def _body() -> None: + nonlocal runs + runs += 1 + if runs < 3: + raise RuntimeError("boom") + w._stop.set() # third run exits cleanly + + try: + await w._supervise("test-loop", _body) + finally: + monkeypatched.undo() + + assert runs == 3, "the loop must be restarted after each crash" + + +async def test_a_permanently_crashing_loop_asks_the_process_to_exit( + patched_repo: _FakeRepo, +) -> None: + """Once the restart budget is spent, the worker asks the process to exit. + + Deployments are expected to run under a restarting supervisor (systemd + ``Restart=always``, Docker ``restart: unless-stopped``, a k8s Deployment). + Continuing to serve with a dead projection pipeline is worse: searches + answer from a silently frozen index while ``/health`` shows nothing wrong. + """ + from everos.memory.cascade import worker as wmod + + w = CascadeWorker({"episode": _OkHandler()}, retry_backoff_seconds=0) + exits: list[str] = [] + monkeypatched = pytest.MonkeyPatch() + monkeypatched.setattr(wmod, "_LOOP_RESTART_BACKOFF_SECONDS", (0.0, 0.0)) + monkeypatched.setattr(w, "_request_process_exit", exits.append) + + attempts = 0 + + async def _always_raises() -> None: + nonlocal attempts + attempts += 1 + raise RuntimeError("deterministic failure") + + try: + await w._supervise("test-loop", _always_raises) + finally: + monkeypatched.undo() + + assert attempts == 3, "initial run + one per backoff entry" + assert exits == ["test-loop"], "process exit must be requested exactly once" + + +async def test_a_stable_run_refills_the_restart_budget( + patched_repo: _FakeRepo, +) -> None: + """Independent transients days apart must not pool into a process exit. + + The restart budget counts consecutive *quick* crashes, not crashes over + the process lifetime. Without the reset, a loop that hits one recoverable + transient every few days — each cleared by a single restart — spends the + budget strike by strike and the crash after the last one SIGTERMs a + healthy server weeks in, which punishes exactly the case supervision + exists to absorb. A body that ran at least ``_LOOP_STABLE_RUN_SECONDS`` + before raising is a fresh incident and gets the full ladder again. + """ + from types import SimpleNamespace + + from everos.memory.cascade import worker as wmod + + w = CascadeWorker({"episode": _OkHandler()}, retry_backoff_seconds=0) + exits: list[str] = [] + clock = {"now": 0.0} + monkeypatched = pytest.MonkeyPatch() + monkeypatched.setattr(wmod, "_LOOP_RESTART_BACKOFF_SECONDS", (0.0, 0.0)) + # Only the worker module sees the fake clock; the event loop keeps real + # time, so the zero backoffs above still pass through _wait_or_stop. + monkeypatched.setattr(wmod, "time", SimpleNamespace(monotonic=lambda: clock["now"])) + monkeypatched.setattr(w, "_request_process_exit", exits.append) + + runs = 0 + + async def _body() -> None: + nonlocal runs + runs += 1 + if runs == 3: + # A long, honest run before this crash — an independent incident. + clock["now"] += wmod._LOOP_STABLE_RUN_SECONDS + 1 + raise RuntimeError("independent transient") + if runs == 5: + w._stop.set() # clean exit + return + raise RuntimeError("quick crash") + + try: + await w._supervise("test-loop", _body) + finally: + monkeypatched.undo() + + # Budget is 2 restarts here: crashes 1-2 spend it, crash 3 (after a + # stable run) must start over rather than exceed it, leaving room for + # crash 4 and the clean run 5. A lifetime budget exits after run 3. + assert runs == 5, "the stable run must refill the budget" + assert exits == [], "no process exit for independent incidents" + + +async def test_supervision_does_not_swallow_cancellation( + patched_repo: _FakeRepo, +) -> None: + """``stop()`` cancels these tasks, so ``CancelledError`` must propagate. + + Catching it as a "crash" would restart the loop during shutdown and make + ``stop()`` hang instead of returning. + """ + w = CascadeWorker({"episode": _OkHandler()}, retry_backoff_seconds=0) + + async def _body() -> None: + raise asyncio.CancelledError + + with pytest.raises(asyncio.CancelledError): + await w._supervise("test-loop", _body) + + +async def test_all_three_loops_are_supervised(patched_repo: _FakeRepo) -> None: + """Supervision must cover every long-lived loop, not just the drain one. + + ``_run_loop`` already had an inner ``try``; ``_heartbeat_loop`` and + ``_rebuild_loop`` did not, which is the asymmetry this guards. + """ + w = CascadeWorker({"episode": _OkHandler()}, retry_backoff_seconds=0) + await w.start() + try: + names = { + t.get_name() + for t in (w._task, w._heartbeat_task, w._rebuild_task) + if t is not None + } + assert names == { + "cascade-worker", + "cascade-worker-heartbeat", + "cascade-worker-rebuild", + } + for task in (w._task, w._heartbeat_task, w._rebuild_task): + assert task is not None + assert task.get_coro().__qualname__.endswith("_supervise"), ( + f"{task.get_name()} is not running under _supervise" + ) + finally: + await w.stop() + + +async def test_optimize_does_not_park_forever_on_a_rebuild( + patched_repo: _FakeRepo, +) -> None: + """The optimize runner's wait on a rebuild must be bounded, like its mirror. + + Whichever maintenance job arrives second parks on the first, so the two + waits are the same hazard seen from opposite ends: while the runner waits, + its task slot stays occupied, ``_schedule_optimize`` keeps short-circuiting + on it, and that table quietly stops being pruned. The rebuild-side wait was + bounded; this one was left open on the argument that ``rebuild_indexes`` + carries its own deadline — true of its critical section only, not of the + task's dispatch and teardown, so nothing actually bounded it. + """ + from everos.memory.cascade import worker as wmod + + monkeypatched = pytest.MonkeyPatch() + monkeypatched.setattr(wmod, "_MAINTENANCE_TASK_TIMEOUT_SECONDS", 0.05) + + fake = _FakeLanceRepo() + w = CascadeWorker( + {"episode": _OkHandlerWithRepo(fake)}, + retry_backoff_seconds=0, + optimize_min_interval_seconds=0.01, + ) + state = wmod._KindOptimizerState() + w._optimizer_states["episode"] = state + state.dirty = True + # A rebuild that never finishes: exactly the state the runner used to wait + # out forever. + state.rebuild_task = asyncio.create_task(asyncio.sleep(30)) + + try: + async with asyncio.timeout(5): + await w._optimize_runner("episode", initial_delay=0) + finally: + state.rebuild_task.cancel() + monkeypatched.undo() + + assert not fake.optimize_calls, ( + "the runner must skip the beat, not compact under a live rebuild — " + "both commit on the same manifest" + )