From 45656d331e5af669d6c958f6af5a3400a5b0fb33 Mon Sep 17 00:00:00 2001 From: zhanghui Date: Fri, 10 Jul 2026 16:54:58 +0800 Subject: [PATCH] fix(lancedb): fix FTS with_position optimize crash + disk bloat (#336) FTS indexes built with with_position=True crash lance's optimize/compaction on lancedb >= 0.32 when merging an unindexed tail (Max offset exceeds length of values; upstream lance-format/lance#7653). The crash aborts optimize() including version cleanup, so the index dir grows unbounded until the disk fills. everos recall is OR-mode BM25 and never does phrase queries, so positions are never read -- disabling is lossless. - base: default with_position=False - infra: migrate_fts_indexes() rebuilds pre-fix indexes once at startup + reclaims orphans - cascade worker: count consecutive optimize failures, escalate warning->error Fixes #335. Co-authored-by: zhanghui Co-authored-by: Claude Opus 4.8 (1M context) --- src/everos/core/persistence/lancedb/base.py | 13 +++- .../infra/persistence/lancedb/__init__.py | 69 +++++++++++++++++-- src/everos/memory/cascade/worker.py | 32 ++++++++- .../test_lancedb/test_repository.py | 56 +++++++++++++++ .../test_memory/test_cascade/test_worker.py | 62 +++++++++++++++++ 5 files changed, 224 insertions(+), 8 deletions(-) diff --git a/src/everos/core/persistence/lancedb/base.py b/src/everos/core/persistence/lancedb/base.py index 6d2b5e7..7efa33d 100644 --- a/src/everos/core/persistence/lancedb/base.py +++ b/src/everos/core/persistence/lancedb/base.py @@ -127,7 +127,16 @@ class BaseLanceTable(LanceModel): filtering and a divided source of truth. - ``ascii_folding=True`` — strips diacritics (é→e) on Latin characters; no-op on CJK. - - ``with_position=True`` — enables phrase queries. + - ``with_position=False`` — everos does OR-mode BM25 recall + (``MatchQuery`` clauses; see ``search.recall.base.build_or_query``), + never phrase queries, so token positions are never read. + Building the position posting List is therefore pure overhead + **and** triggers a ``Max offset exceeds length of values`` + offset-overflow crash inside lance's compaction once the + position lists grow large (upstream lance-format/lance#7653). + That crash blocks ``optimize()`` — including version cleanup — + so the index dir grows unbounded until the disk fills. Keeping + positions off avoids both the overhead and the crash. Subclasses normally do not need to override this — declaring :attr:`BM25_FIELDS` is enough. @@ -142,7 +151,7 @@ class BaseLanceTable(LanceModel): await table.create_index( column=field, config=FTS( - with_position=True, + with_position=False, base_tokenizer="whitespace", lower_case=True, stem=True, diff --git a/src/everos/infra/persistence/lancedb/__init__.py b/src/everos/infra/persistence/lancedb/__init__.py index 832bd34..8f8b3c5 100644 --- a/src/everos/infra/persistence/lancedb/__init__.py +++ b/src/everos/infra/persistence/lancedb/__init__.py @@ -24,6 +24,12 @@ first access; row population is the cascade daemon's job (see ``12_cascade_design.md``). """ +import contextlib +import datetime as dt + +from everos.core.observability.logging import get_logger +from everos.core.persistence import MemoryRoot + # Importing ``tables`` registers every business :class:`BaseLanceTable` # schema so callers can rely on the package alone to surface every schema. from . import tables as tables @@ -69,19 +75,71 @@ class LanceDBSchemaMismatchError(RuntimeError): """ +_FTS_INDEX_SCHEMA_VERSION = 2 +"""Bump when the FTS index build config changes so existing on-disk +indexes get rebuilt at startup. v2 = ``with_position=False`` (see +:meth:`BaseLanceTable.ensure_fts_indexes` + lance-format/lance#7653).""" + + +async def migrate_fts_indexes() -> None: + """One-time rebuild of FTS indexes that predate the current config. + + Older indexes were built with ``with_position=True``; that position + posting List overflows lance's compaction once it grows large + (``Max offset exceeds length of values``, lance-format/lance#7653), + which aborts ``optimize()`` — including version cleanup — so the + index dir grows unbounded until the disk fills. + + Rebuilds every business table's FTS index with the current + :meth:`BaseLanceTable.ensure_fts_indexes` config (``with_position`` + now off) and reclaims the orphaned index files / data fragments the + crashed-optimize churn left behind. Guarded by a version marker in + the LanceDB dir so it runs at most once per bump; the rebuild is + O(N) but only on the first startup after upgrade. + """ + logger = get_logger(__name__) + marker = MemoryRoot.default().lancedb_dir / ".fts_index_version" + try: + current = int(marker.read_text().strip()) if marker.exists() else 0 + except (ValueError, OSError): + current = 0 + if current >= _FTS_INDEX_SCHEMA_VERSION: + return + logger.info("fts_index_migration_started", target=_FTS_INDEX_SCHEMA_VERSION) + for schema in _BUSINESS_SCHEMAS: + if not schema.BM25_FIELDS: + continue + table = await get_table(schema.TABLE_NAME, schema) + # Drop existing indexes (everos only builds FTS here; mirrors + # LanceRepoBase.rebuild_indexes) then rebuild with the new config. + for idx in await table.list_indices(): + await table.drop_index(idx.name) + await schema.ensure_fts_indexes(table) + # Reclaim the orphaned index dirs + data fragments the crashed + # optimize loop piled up. Safe now: the crashing index is gone, + # so compaction no longer decodes a position List. + with contextlib.suppress(Exception): + await table.optimize(cleanup_older_than=dt.timedelta(seconds=0)) + marker.write_text(str(_FTS_INDEX_SCHEMA_VERSION)) + logger.info("fts_index_migration_done", version=_FTS_INDEX_SCHEMA_VERSION) + + async def ensure_business_indexes() -> None: """Ensure FTS (BM25) indexes for every business table (idempotent). - Called once at startup by :class:`LanceDBLifespanProvider`. Walks - the 5 business schemas (each schema owns its ``TABLE_NAME`` + - ``BM25_FIELDS``), opens each table via :func:`get_table`, and - delegates to ``schema.ensure_fts_indexes(table)``. Already-indexed - columns are skipped, so re-runs are no-ops. + Called once at startup by :class:`LanceDBLifespanProvider`. First + runs :func:`migrate_fts_indexes` (one-time, marker-guarded) to + rebuild any pre-fix ``with_position=True`` indexes, then walks the + business schemas (each owns its ``TABLE_NAME`` + ``BM25_FIELDS``), + opens each table via :func:`get_table`, and delegates to + ``schema.ensure_fts_indexes(table)``. Already-indexed columns are + skipped, so re-runs are no-ops. Adding a new business table = adding it to ``_BUSINESS_SCHEMAS``; everything else (table name, columns to index) reads off the schema's ClassVars. """ + await migrate_fts_indexes() for schema in _BUSINESS_SCHEMAS: table = await get_table(schema.TABLE_NAME, schema) await schema.ensure_fts_indexes(table) @@ -134,6 +192,7 @@ __all__ = [ "get_connection", "get_table", "knowledge_topic_repo", + "migrate_fts_indexes", "user_profile_repo", "verify_business_schemas", ] diff --git a/src/everos/memory/cascade/worker.py b/src/everos/memory/cascade/worker.py index e826227..8525cb3 100644 --- a/src/everos/memory/cascade/worker.py +++ b/src/everos/memory/cascade/worker.py @@ -63,6 +63,13 @@ DEFAULT_POLL_INTERVAL_SECONDS = 1.0 DEFAULT_RETRY_BACKOFF_SECONDS = 2.0 DEFAULT_OPTIMIZE_MIN_INTERVAL_SECONDS = 10.0 DEFAULT_OPTIMIZE_HEARTBEAT_SECONDS = 60.0 +_OPTIMIZE_FAILURE_ALERT_THRESHOLD = 5 +"""Consecutive ``optimize()`` failures (per kind) before the log is +escalated from ``warning`` to ``error``. A one-off failure is benign +(next tick retries); a sustained streak means compaction + version +cleanup are stuck and the index dir will grow unbounded — that must +surface to health checks / alerting rather than rot as a warning nobody +reads (the failure mode behind lance-format/lance#7653).""" DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS = 12 * 60 * 60.0 """How often (per kind) to do a full ``drop_index + create_index`` rebuild. @@ -137,6 +144,12 @@ class _KindOptimizerState: last_run_at: float = 0.0 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.""" task: asyncio.Task[None] | None = None rebuild_task: asyncio.Task[None] | None = None """In-flight rebuild task slot, separate from ``task`` so ordinary @@ -485,16 +498,33 @@ class CascadeWorker: await repo.optimize(cleanup_older_than=cleanup) if should_prune and state is not None: state.last_prune_at = now + if state is not None: + state.optimize_failures = 0 logger.debug( "cascade_lancedb_optimized", kind=kind, pruned=should_prune, ) except Exception as exc: - logger.warning( + failures = 0 + if state is not None: + state.optimize_failures += 1 + failures = state.optimize_failures + # A one-off failure is benign (next tick retries). A sustained + # streak means optimize — compaction *and* version cleanup — is + # stuck, so the index dir grows unbounded; escalate to error so + # it surfaces to health checks / alerting instead of rotting as + # a warning nobody reads (see lance-format/lance#7653). + log = ( + logger.error + if failures >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD + else logger.warning + ) + log( "cascade_lancedb_optimize_failed", kind=kind, pruned=should_prune, + consecutive_failures=failures, error=f"{type(exc).__name__}: {exc}", ) diff --git a/tests/unit/test_core/test_persistence/test_lancedb/test_repository.py b/tests/unit/test_core/test_persistence/test_lancedb/test_repository.py index 91ce2b7..a8946f9 100644 --- a/tests/unit/test_core/test_persistence/test_lancedb/test_repository.py +++ b/tests/unit/test_core/test_persistence/test_lancedb/test_repository.py @@ -647,3 +647,59 @@ async def test_write_lock_is_per_table(tmp_path: Path) -> None: assert repo_a._write_lock(repo_a.table_name) is not repo_b._write_lock( repo_b.table_name ) + + +# ── migrate_fts_indexes (one-time rebuild of pre-fix with_position indexes) ── + + +async def test_migrate_fts_indexes_runs_once_and_rebuilds( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """migrate_fts_indexes rebuilds existing FTS indexes once, guarded by a + version marker in the LanceDB dir (fix for lance-format/lance#7653). + + White-box surfaces: the ``.fts_index_version`` marker file and + ``AsyncTable.list_indices`` on the global-connection table. + """ + monkeypatch.setenv("EVEROS_ROOT", str(tmp_path)) + import everos.infra.persistence.lancedb as lancedb_infra + from everos.core.persistence import MemoryRoot + from everos.infra.persistence.lancedb import ( + dispose_connection, + get_table, + migrate_fts_indexes, + ) + + monkeypatch.setattr(lancedb_infra, "_BUSINESS_SCHEMAS", (_SearchNote,)) + await dispose_connection() + try: + table = await get_table(_SearchNote.TABLE_NAME, _SearchNote) + await table.add( + [ + _SearchNote( + id="1", + text="hello world", + tokens="hello world", + vector=[1, 0, 0, 0], + ) + ] + ) + await _SearchNote.ensure_fts_indexes(table) + assert any("tokens" in (i.columns or []) for i in await table.list_indices()) + + marker = MemoryRoot.default().lancedb_dir / ".fts_index_version" + assert not marker.exists() + + # First run: migrates + writes the marker, index still present. + await migrate_fts_indexes() + assert marker.read_text().strip() == "2" + assert any("tokens" in (i.columns or []) for i in await table.list_indices()) + + # Marker present → second run is a no-op. Drop the index, re-run, + # and confirm it is NOT rebuilt (migration skipped, not re-executed). + for i in await table.list_indices(): + await table.drop_index(i.name) + await migrate_fts_indexes() + assert not list(await table.list_indices()) + finally: + await dispose_connection() diff --git a/tests/unit/test_memory/test_cascade/test_worker.py b/tests/unit/test_memory/test_cascade/test_worker.py index 0072bc8..b2622f0 100644 --- a/tests/unit/test_memory/test_cascade/test_worker.py +++ b/tests/unit/test_memory/test_cascade/test_worker.py @@ -571,3 +571,65 @@ async def test_rebuild_failure_does_not_crash_daemon( await w.stop() # Worker is still alive (stop() returned cleanly). assert w._task is None + + +class _OptimizeFailingRepo(_FakeLanceRepo): + """Fake repo whose ``optimize()`` raises until ``fail`` is cleared.""" + + def __init__(self, **kw) -> None: # type: ignore[no-untyped-def] + super().__init__(**kw) + self.fail = True + + async def optimize(self, *, cleanup_older_than: dt.timedelta | None = None) -> None: + if self.fail: + raise RuntimeError("Max offset of 9 exceeds length of values 3") + await super().optimize(cleanup_older_than=cleanup_older_than) + + +async def test_optimize_failures_counted_escalated_and_reset( + patched_repo: _FakeRepo, + monkeypatch: pytest.MonkeyPatch, +) -> 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 on the next + success — instead of being swallowed as a silent warning stream that + lets the index dir grow until the disk fills. + """ + from everos.memory.cascade import worker as wmod + + calls: list[tuple[str, str]] = [] + + class _SpyLogger: + def __getattr__(self, level: str): # type: ignore[no-untyped-def] + def rec(event: str, **_kw) -> None: # type: ignore[no-untyped-def] + calls.append((level, event)) + + return rec + + monkeypatch.setattr(wmod, "logger", _SpyLogger()) + + 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 _ in range(threshold): + await w._run_optimize_once("episode") + + state = w._optimizer_states["episode"] + assert state.optimize_failures == threshold + + fail_logs = [lvl for lvl, ev in calls if ev == "cascade_lancedb_optimize_failed"] + assert fail_logs[:-1] == ["warning"] * (threshold - 1) + assert fail_logs[-1] == "error" + + # A subsequent success resets the streak. + repo.fail = False + await w._run_optimize_once("episode") + assert state.optimize_failures == 0