fix: harden the stall defect class found auditing the 1.2.2 fix (#392)
* fix(lancedb): bound reads and unhook the husk sweep from the write lock Two instances of the same defect class the write-lock deadline work left behind: an await in a scheduler-gated path with nothing bounding it. Reads (count / get_by_id / find_where / find_where_paginated / search) were skipped last round on the reasoning that a read takes no lock and so blocks no writer. True, but incomplete: 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 (claim_pending_batch only takes `pending`, orphan recovery runs once at startup), and /health keeps reporting healthy because a hang raises nothing. Budget 60s, ~1000x the measured 62ms flat scan over 117k rows. The empty-index-dir sweep ran inside the prune critical section under a docstring contract requiring the write lock. That contract could not hold: the sweep runs via asyncio.to_thread, and a deadline cancels the future, not the thread, so an orphan sweep outlives the lock -- and Path.iterdir is a lazy os.scandir, so it can yield a dir created after the scan began. It could therefore rmdir a directory a concurrent create_index had just made, leaving the table with no FTS index and every search on that kind 500ing. Safety now comes from an age filter (skip dirs younger than 300s), which holds regardless of lock ownership; the sweep moved out of the critical section so a slow filesystem walk can no longer overrun the prune budget. Mutation-verified: moving the table handle back outside the read deadline hangs the new test; moving the sweep back inside the lock fails it; setting the age floor to 0 fails the fresh-dir test. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(cascade): supervise background loops and unmask the optimize alert The drain / heartbeat / rebuild loops were plain create_task coroutines. One uncaught exception ended that loop permanently: nothing restarted it, and because the worker holds a strong reference to the task the interpreter never printed "Task exception was never retrieved" either (that fires on GC). The loop's job just stopped happening with zero output. _run_loop had an inner try; its two siblings did not. Each loop now runs under _supervise: log, wait, restart with escalating backoff (5s / 15s / 45s), then request process exit via SIGTERM so a restarting supervisor (systemd Restart=always, Docker restart: unless-stopped, a k8s Deployment) can recover it. SIGTERM rather than os._exit so the ASGI server runs its graceful-shutdown path. A done-callback is the last-resort observer for the supervisor itself ending unexpectedly. Separately, the fallback rebuild reset the same counter the health verdict reads, so the optimize-failure threshold was effectively unreachable: a table failing 100% of the time cycled 1..5 -> 0 -> 1.. and the threshold value existed only during the sub-second rebuild, ~1% observable against a 30s scrape. cascade.healthy stayed green while the table never reclaimed a version. The rate limiter moves to failures_since_fallback; only a successful optimize clears the alert streak. Same shape as the run7 cross-kind max() masking bug -- a remediation path refreshing the signal meant to report it. Mutation-verified: dropping the restart budget, removing the exit request, and restoring the counter reset each turn the corresponding test red. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(persistence): bound and announce the memory-root lock wait Acquisition polls with LOCK_NB instead of blocking inside a worker thread. A blocking flock could be neither bounded nor cancelled: cancelling the awaiting coroutine leaves the thread to acquire the lock later with nobody left to release it, which is strictly worse than waiting. The wait itself is correct by design -- the second process is supposed to wait, then find the migration already done -- and flock is released by the kernel on process exit, so a dead holder never wedges it. What was wrong is that it had no upper bound and emitted nothing: a server startup landing on a held lock looked like a hang whose last log line was lifespan_provider_startup name=lancedb. It now logs memory_root_lock_waiting on first contention, reports how long it waited on success, and gives up after timeout_seconds (default 300s, generous because the legitimate holder is an O(rows) migration). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test(cascade): wait for projection quiescence in the lap-race scenario test_lap_append_during_handler_no_loss asserted no loss after _wait_path_done, whose settle window is 0.1s. That is a bet that the filesystem event for the appends which landed *during* a handler invocation has already been delivered — a terminal row does not mean the file is fully projected, because the handler read the md at whatever length it had then, marked the row done, and the rest arrive on a later event. The bet holds on macOS/fsevents and lost on a loaded Linux runner (md=30 lance=17), failing the assertion for a reason unrelated to the behaviour under test. Waits for quiescence instead: terminal row + empty pending queue + a projected count unchanged across three consecutive polls. Strictly stronger than the old condition, and real loss still fails — the count just converges below the md entry count and stays there. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(lancedb): replace indexes in place and retry a lost rebuild race rebuild_indexes dropped every index and recreated it. The docstring justified that with "LanceDB transparently falls back to brute-force scan", which 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". The recall legs are gathered without return_exceptions, so one failing leg fails the whole search request -- every keyword search landing in the window returned 500. Measured: 55 failures across 3 rebuilds with drop+create, 0 with create_index(replace=True). Replacing also collapses the live fragment set identically (7 index files back to 4 after 25 optimize beats), so nothing the rebuild existed for is lost. Only indexes on columns that are no longer indexed at all are still dropped -- nothing queries those. A rebuild that loses the manifest race is now retried rather than deferred to the next 12h sweep. Lance marks the conflict Retryable and means it: another process committed first. Retries are recorded as a deadline on the kind (10min / 30min / 3h) and picked up by the rebuild loop, not slept through -- the loop walks kinds sequentially, so sleeping would park every later kind behind the backoff (7 kinds x 3h outlasts the cadence itself). The loop tick is min(60s, cadence) so a shorter configured interval is not quantised. Removes the empty-index-dir sweep. cleanup_older_than deletes the files under a superseded _indices/<uuid>/ but leaves the directory, and everos was removing those with its own rmdir. No LanceDB contract says an empty index dir is garbage, so this is being raised upstream instead. Note it is a separate gap from index *files* not being reclaimed under delete_unverified=False (260MB retained on a 19k-row soak table) -- solving that still leaves the empty dirs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs(changelog): put only the sweep under Removed The rebuild-bound and lock-wait entries were swallowed into the Removed section when the husk-sweep entry was inserted above them. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(lancedb): bound the husk sweep by lance's own unverified threshold Restores the empty-index-dir sweep with a safety argument instead of a self-chosen number. Upstream context, read from lance's cleanup.rs: it unlinks a superseded index's files but never the directory, and contains no rmdir at all. That is structural, not an oversight -- lance targets object stores, where paths are flat keys and an empty directory does not exist. Only a local filesystem materialises them, where they accumulate as inodes (a soak run reached 13061 dirs, 98% empty) and slow every directory scan. Three independent guarantees, in order of strength: 1. rmdir cannot delete data. The kernel refuses it on a non-empty directory, so no file can be lost 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 must outlive UNVERIFIED_THRESHOLD_DAYS = 7, which is lance's own bound for deciding an unreferenced index UUID is dead rather than an index build in progress. Matching it means the sweep can never be more aggressive than lance itself; the previous 300s was our invention, and that is what made it indefensible. Each guarantee is pinned by the same test and mutation-verified: dropping the live-UUID check, zeroing the age gate, and swapping rmdir for a recursive delete each turn it red. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * feat(cascade): make the maintenance cadences configurable The four cadences were already constructor arguments on CascadeWorker, but CascadeConfig did not carry them and none of the three production construction paths passed a config, so the module defaults were unreachable from outside the code. That is why no soak run shorter than half a day could exercise the 12h rebuild sweep: not a missing parameter, a config layer that dropped it. Adds CascadeSettings ([cascade] in default.toml) and CascadeConfig.from_settings, which the orchestrator now uses when no config is passed -- so the CLI, backfill and server paths all pick settings up at once. Deliberately not exposed: the read / write / prune / rebuild deadlines. Those are hang-catchers sized from measured durations, and both directions are worse -- too low manufactures failures on a healthy table, too high leaves a wedged one invisible for longer. Cadences depend on write volume and are a real tuning axis; deadlines are not. A test pins the exposed field set so a later change has to state its intent. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test(cascade): cover the AgentCase handler agent_case was the one business kind with no handler test, and the storage soak never writes it either -- four of the seven tables stay at zero rows there -- so its md -> row contract was unexercised from both directions. Covers what makes this kind different from its daily-log siblings: it lives on the agent track, and it embeds task_intent only while approach is BM25-indexed but deliberately never sent to the embedder. Plus the branches every handler shares: soft-dependency embedding (no provider -> vector=None, row still written for keyword-only deployments), optional KeyInsight, the content_sha256 short-circuit that stops the 30s scanner re-embedding untouched files, edit detection, and delete-by-path. Mutation-verified: routing approach into the embedder, and dropping section:TaskIntent from content_change_keys, each turn the relevant test red. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs(lancedb): record the empty-dir cost and what caps it Two constants now carry the arithmetic instead of leaving it in a chat log. _HUSK_MIN_AGE_SECONDS states what the 7-day gate costs: nothing reclaims an empty index dir before then, each is an inode plus a 4KB block, and at ceiling load that is ~890k dirs / ~3.6GB / 14% of a default 98GB ext4's inodes at the 7-day steady state. Also that this is the worst case and needs sustained saturation -- a single-user deployment sits four orders of magnitude below it -- and that only ext4 has a fixed inode budget (APFS and xfs allocate dynamically, Windows is out of scope). DEFAULT_OPTIMIZE_MIN_INTERVAL_SECONDS gets two things it never said. First, it is not a visibility delay: a row is searchable as soon as its upsert commits, because LanceDB flat-scans the unindexed tail -- verified to cover BM25, not just vector and scalar, which was the leg worth doubting given a missing FTS index hard-fails rather than degrading. Sparse writes do not wait at all, since the scheduler uses max(0, interval - elapsed). Second, it is the ceiling on index-directory growth: past roughly one write per table per interval the beats coalesce, so the accrual rate is capped by this interval rather than by write volume, and raising it lowers the cost proportionally. That makes it the knob to reach for if the empty dirs ever bite -- not the husk threshold. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(cascade): reset the loop-restart budget after a stable run The supervisor's restart budget was spent per process lifetime: three strikes ever, regardless of how long the loop ran healthily between them. A loop hitting one recoverable transient every few days — each cleared by a single restart — would still pool those strikes and SIGTERM a healthy server weeks in, on the 4th, which punishes exactly the case supervision exists to absorb. The budget now counts consecutive quick crashes: a body that ran at least 60s before raising starts a fresh incident with the full ladder. A deterministic crash-on-entry still exhausts the budget in ~65s. Same windowed counting as systemd StartLimitIntervalSec / Erlang max_restarts-per-max_seconds. Also corrects the husk accrual numbers in the optimize-cooldown docstring to the ~14-day effective reclaim horizon (see the sibling lancedb commit for why the age gate doubles). Mutation-verified: with the reset removed, the new test exits after run 3 instead of surviving to run 5. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(lancedb): keep a husk-sweep timeout from failing the prune 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 deadline escape prune() billed the failure to the wrong account: the optimize scheduler counted a prune failure (feeding the fallback-rebuild threshold) and the prune-staleness clock stopped advancing, so both alarms reported 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. Reachable, not theoretical: sweep time is proportional to dir count (~35us/dir measured) and the ceiling-load steady state sits right at the 60s budget. The timeout is tolerable exactly because it is now swallowed — and the orphaned worker thread finishes the walk anyway, so the reclamation still happens. Also corrects the age-gate docstrings: the gate reads st_mtime, which POSIX bumps when lance's cleanup empties the husk, so the effective reclaim horizon is file wait + 7 days (~14 days total) and the ceiling-load steady state is ~1.8M dirs / ~7GB, twice the previously recorded figure. The "never more aggressive than lance" property is unaffected (it is strictly more conservative). Mutation-verified: with the try/except removed, the new test fails on the escaping VectorStoreBusyError. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(persistence): move the lock timeout an order above the legit hold 300s sat at the edge of what the docstring itself calls a legitimate hold (a large migration is minutes of O(rows) work), so the worst honest migration turned every waiting process's startup into a LockError crash. Now 1800s: the wait has been visible since the first poll (memory_root_lock_waiting), and against the one case the bound exists for — a holder alive but wedged — giving up at 5 minutes buys nothing over 30, because the timeout's job is diagnosis, not recovery. The timeout message now says which way to look: the kernel releases a dead holder's flock automatically, so reaching the timeout means the holder is alive — inspect that process instead of retrying this one. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(changelog): fold the review fixes into the unreleased notes Supervisor bullet gains the per-incident budget, the sweep bullet gains the ~14-day effective horizon and the swallowed timeout, and the lock bullet records the 30min default with its sizing rationale. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * test(lancedb): pin that only a sweep timeout is absorbed by prune The sibling test covers one side -- a deadline miss must not bill prune's ledger. Nothing covered the other: widening the catch to `except Exception` passes every other test in the file, and would turn a genuine fault in _remove_empty_index_dirs (a TypeError after a signature change, a permission error on the index dir) into a silent removed = 0 with no signal anywhere. That is the failure shape this module keeps being audited for, so the narrowness of the catch needs its own guard. Found by mutating the catch rather than by reading it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(cascade): bound the optimize runner's wait on a rebuild The two maintenance jobs park on each other -- whichever arrives second waits on the first -- so the two waits are one hazard seen from opposite ends. The rebuild side was bounded; this side was left open on the argument that rebuild_indexes carries its own 300s deadline. That deadline covers its critical section, not the task's dispatch and teardown around it, so the transitive bound was never real. While the runner waits, its per-kind task slot stays occupied, every _schedule_optimize call short-circuits on it, and that table silently stops being pruned -- the same shape as the stall this branch has been chasing. Bounded at 180s. On expiry the beat is skipped rather than run: compacting under a live rebuild is the interleaving the wait exists to prevent, and both commit on the same manifest. Writes keep the dirty flag set, so the next beat retries. Mutation-verified: replacing the timeout with a plain await hangs the new test until its own guard fires. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: zhanghui <zhanghui@shanda.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
zhanghui
GitHub
parent
d5668dcba1
commit
8024fe576e
127
CHANGELOG.md
127
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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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/<uuid>/`` 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/<uuid>/`` 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/<uuid>/`` 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/<uuid>/`` 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/<uuid>/`` 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/<uuid>/`` 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 ``<owner_id>_<entry_id>``
|
||||
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 ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
(``<root>/.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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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/<uuid>/``
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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}"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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/<uuid>/`` 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))
|
||||
|
|
|
|||
|
|
@ -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 == []
|
||||
|
|
|
|||
|
|
@ -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/<id>/.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
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
)
|
||||
|
|
|
|||
Loading…
Reference in New Issue