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 <zhanghui@shanda.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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,

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

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}",
             )
 

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

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