fix(lancedb): 隔离重量级维护并让全新 upsert 批次走 add 快路径

src/everos/config/default.toml

@@ -209,7 +209,9 @@ optimize_prune_interval_seconds = 300.0
 # 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
+# 0 disables in-process full index rebuilds. Enable only after measuring native
+# memory retention on the actual dataset; optimize/prune stay active regardless.
+optimize_rebuild_interval_seconds = 0.0
 # How often the scanner walks the memory root for changes the watcher missed.
 # On a mount that delivers no filesystem events (a Windows directory bound into
 # WSL2 or a container) this is the edit-to-searchable latency. Every sweep also

src/everos/config/settings.py

@@ -556,8 +556,10 @@ class CascadeSettings(BaseModel):
 
     ``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.
+      count that every ``optimize()`` grows. ``0`` disables automatic full
+      rebuilds. They are disabled by default because Arrow/Lance can retain
+      GiB-scale native allocations after rebuilding a production-sized table;
+      regular optimize/prune remains enabled for routine maintenance.
 
     ``scan_interval_seconds``:
       How often the scanner walks the memory root for what the watcher
@@ -572,7 +574,7 @@ class CascadeSettings(BaseModel):
     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
+    optimize_rebuild_interval_seconds: float = 0.0
     scan_interval_seconds: float = Field(default=30.0, gt=0)
 
 

src/everos/core/persistence/lancedb/repository.py

@@ -475,22 +475,42 @@ async def upsert(self, records: Sequence[T]) -> None:
         parameter on purpose: the key is the table's declaration, not the
         caller's choice.
 
+        When the primary key is a single string column, a cheap PK
+        existence probe lets an all-new batch take LanceDB's append-only
+        ``add`` path; only batches that actually contain an existing key
+        pay for ``merge_insert``. This matters for daily logs: a merge
+        against a large fragment can rewrite hundreds of MiB and retain
+        GiB-scale Arrow/Lance native allocations even when the batch
+        merely appends a handful of new entries.
+
         Cascade uses this when reconciling md → LanceDB: an entry seen
         for the first time inserts; an entry that was edited in md
         updates its existing row.
         """
-        key = getattr(self.schema, "PRIMARY_KEY", None)
+        key = tuple(getattr(self.schema, "PRIMARY_KEY", None) or ())
         if not key:
             raise ValueError(
                 f"derived-index model {self.schema.__name__} has no PRIMARY_KEY"
             )
+        rows = list(records)
+        if not rows:
+            return
         async with self._locked(_WRITE_TIMEOUT_SECONDS, "upsert"):
             table = await self._table()
+            if len(key) == 1:
+                by = key[0]
+                key_values = [getattr(row, by) for row in rows]
+                if all(isinstance(value, str) for value in key_values):
+                    values = ", ".join(f"'{_q(value)}'" for value in key_values)
+                    existing = await table.count_rows(filter=f"{by} IN ({values})")
+                    if existing == 0:
+                        await table.add(rows)
+                        return
             await (
                 table.merge_insert(list(key))
                 .when_matched_update_all()
                 .when_not_matched_insert_all()
-                .execute(list(records))
+                .execute(rows)
             )
 
     # ── Maintenance ────────────────────────────────────────────────────────

src/everos/memory/cascade/orchestrator.py

@@ -213,6 +213,10 @@ def __init__(
             optimize_rebuild_interval_seconds=(
                 self._config.optimize_rebuild_interval_seconds
             ),
+            maintenance_enabled=os.getenv(
+                "EVEROS_DISABLE_CASCADE_MAINTENANCE", ""
+            ).strip().lower()
+            not in ("1", "true", "yes"),
         )
         self._watcher: CascadeWatcher | None = None
         self._started = False

src/everos/memory/cascade/worker.py

@@ -457,6 +457,7 @@ def __init__(
         optimize_rebuild_interval_seconds: float = (
             DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS
         ),
+        maintenance_enabled: bool = True,
     ) -> None:
         self._handlers = handlers
         self._batch_size = batch_size
@@ -468,6 +469,7 @@ def __init__(
         self._optimize_prune_interval = optimize_prune_interval_seconds
         self._optimize_prune_retention = optimize_prune_retention_seconds
         self._optimize_rebuild_interval = optimize_rebuild_interval_seconds
+        self._maintenance_enabled = maintenance_enabled
         self._task: asyncio.Task[None] | None = None
         self._heartbeat_task: asyncio.Task[None] | None = None
         self._rebuild_task: asyncio.Task[None] | None = None
@@ -484,12 +486,20 @@ async def start(self) -> None:
         self._stop.clear()
         self._started_at = time.monotonic()
         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 = self._spawn_loop(
-            "rebuild", self._rebuild_loop, "cascade-worker-rebuild"
-        )
+        if self._maintenance_enabled:
+            self._heartbeat_task = self._spawn_loop(
+                "heartbeat", self._heartbeat_loop, "cascade-worker-heartbeat"
+            )
+        else:
+            self._heartbeat_task = None
+            logger.info("cascade_lancedb_online_maintenance_disabled")
+        if self._maintenance_enabled and self._optimize_rebuild_interval > 0:
+            self._rebuild_task = self._spawn_loop(
+                "rebuild", self._rebuild_loop, "cascade-worker-rebuild"
+            )
+        else:
+            self._rebuild_task = None
+            logger.info("cascade_lancedb_auto_rebuild_disabled")
         logger.info("cascade_worker_started", batch_size=self._batch_size)
 
     def _spawn_loop(
@@ -890,6 +900,8 @@ def _schedule_optimize(self, kind: str) -> None:
         repo = getattr(handler, "index_repo", None) if handler else None
         if repo is None:
             return
+        if not self._maintenance_enabled:
+            return
         state = self._optimizer_states.setdefault(kind, _KindOptimizerState())
         state.dirty = True
         if state.task is not None and not state.task.done():
@@ -1103,7 +1115,10 @@ async def _run_optimize_once(self, kind: str) -> None:
                 consecutive_failures=failures,
                 error=f"{type(exc).__name__}: {exc}",
             )
-            if since_fallback >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD:
+            if (
+                since_fallback >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD
+                and self._optimize_rebuild_interval > 0
+            ):
                 logger.info(
                     "cascade_lancedb_optimize_fallback_rebuild",
                     kind=kind,
@@ -1150,10 +1165,13 @@ async def _rebuild_loop(self) -> None:
         current stack (lancedb 0.30.2 / lance 4.0) to bound active
         index UUID / FTS ``part_N`` accumulation.
 
-        First sweep fires immediately on worker start to bound any
-        accumulation from a previous session. Subsequent sweeps honour
-        the interval. Both sweep and each per-kind step respect
-        ``self._stop`` so shutdown is prompt.
+        The first sweep waits for the configured interval, just like every
+        subsequent sweep. Rebuilding at process start made restart frequency
+        control maintenance frequency and, on production-sized Lance tables,
+        retained several GiB of native Arrow/Lance allocations before the
+        server handled its first request. Regular optimize/prune heartbeats
+        already keep the unindexed tail and dead files bounded between full
+        rebuilds, so a restart must not force an extra O(N) rebuild.
 
         Rebuild is serialised through the per-kind
         :class:`_KindOptimizerState.task` slot so it does not race with
@@ -1162,14 +1180,6 @@ async def _rebuild_loop(self) -> None:
         is harmless for correctness (queries / writes keep working
         against the existing indices).
         """
-        # First sweep: catch up from any prior session before honouring the
-        # interval. Rebuild is cheap (~0.3s per 50k rows × indexed columns
-        # in local benchmarks); deferring it 12h after startup risks long
-        # accumulation if the daemon restarts often.
-        for kind in self._handlers:
-            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

tests/unit/test_memory/test_cascade/test_worker.py

@@ -493,15 +493,10 @@ async def test_stop_waits_for_in_flight_optimize(
         retry_backoff_seconds=0,
         optimize_min_interval_seconds=0.02,
         optimize_heartbeat_seconds=10.0,
-        # Park rebuild interval — startup sweep still fires but we wait
-        # for it before testing optimize semantics.
+        # Park rebuild interval while testing optimize semantics.
         optimize_rebuild_interval_seconds=10.0,
     )
     await w.start()
-    # Let the startup rebuild sweep complete (instant for the fake repo)
-    # before scheduling optimize — otherwise optimize would queue behind it.
-    await asyncio.sleep(0.02)
-    assert fake.rebuild_calls, "startup rebuild should have fired by now"
     w._schedule_optimize("episode")
     await asyncio.sleep(0.01)  # let optimize start
     await w.stop()
@@ -738,14 +733,35 @@ async def test_prune_recurs_once_per_cadence_across_light_beats(
 # ── Rebuild scheduler tests ────────────────────────────────────────────────
 
 
-async def test_rebuild_runs_on_startup_for_every_kind(
+async def test_zero_rebuild_interval_disables_loop_and_fallback(
     patched_repo: _FakeRepo,
 ) -> None:
-    """The first rebuild sweep fires on worker start, before any interval.
+    """Disabled full rebuilds never allocate in the daemon, even on failure."""
+    from everos.memory.cascade import worker as wmod
 
-    Otherwise a daemon that restarts more often than the rebuild
-    interval would never bound accumulated UUIDs.
-    """
+    repo = _OptimizeFailingRepo()
+    w = CascadeWorker(
+        {"episode": _OkHandlerWithRepo(repo)},
+        retry_backoff_seconds=0,
+        optimize_min_interval_seconds=0.01,
+        optimize_heartbeat_seconds=10.0,
+        optimize_rebuild_interval_seconds=0.0,
+    )
+    await w.start()
+    assert w._rebuild_task is None
+
+    w._optimizer_states["episode"] = wmod._KindOptimizerState()
+    for _ in range(wmod._OPTIMIZE_FAILURE_ALERT_THRESHOLD * 2):
+        await w._run_optimize_once("episode")
+
+    await w.stop()
+    assert repo.rebuild_calls == []
+
+
+async def test_rebuild_does_not_run_on_startup(
+    patched_repo: _FakeRepo,
+) -> None:
+    """A restart must not force an O(N) full-index rebuild."""
     fake_a = _FakeLanceRepo()
     fake_b = _FakeLanceRepo()
     w = CascadeWorker(
@@ -756,21 +772,20 @@ async def test_rebuild_runs_on_startup_for_every_kind(
         retry_backoff_seconds=0,
         optimize_min_interval_seconds=0.01,
         optimize_heartbeat_seconds=10.0,  # park heartbeat
-        optimize_rebuild_interval_seconds=10.0,  # only the startup sweep should fire
+        optimize_rebuild_interval_seconds=10.0,
     )
     await w.start()
-    # Allow the startup sweep to complete; the next tick is 10s away.
+    # The first interval is 10s away, so startup must remain side-effect free.
     await asyncio.sleep(0.1)
     await w.stop()
-    # Exactly one rebuild per kind: the startup sweep. Next interval is 10s.
-    assert len(fake_a.rebuild_calls) == 1
-    assert len(fake_b.rebuild_calls) == 1
+    assert fake_a.rebuild_calls == []
+    assert fake_b.rebuild_calls == []
 
 
 async def test_rebuild_runs_periodically(
     patched_repo: _FakeRepo,
 ) -> None:
-    """After the startup sweep, rebuild repeats every interval."""
+    """The first and subsequent rebuilds run only after full intervals."""
     fake = _FakeLanceRepo()
     w = CascadeWorker(
         {"episode": _OkHandlerWithRepo(fake)},
@@ -784,9 +799,8 @@ async def test_rebuild_runs_periodically(
     # so size the window off the observed tick, not the requested one.
     await asyncio.sleep(0.6)
     await w.stop()
-    # Startup sweep + at least 2 interval-driven sweeps.
-    assert len(fake.rebuild_calls) >= 3, (
-        f"expected ≥3 rebuilds (1 startup + ≥2 periodic), got {len(fake.rebuild_calls)}"
+    assert len(fake.rebuild_calls) >= 2, (
+        f"expected ≥2 interval-driven rebuilds, got {len(fake.rebuild_calls)}"
     )
 
 
@@ -803,7 +817,7 @@ async def test_rebuild_failure_does_not_crash_daemon(
         optimize_rebuild_interval_seconds=10.0,
     )
     await w.start()
-    # Give startup rebuild a chance to throw, then heartbeat to keep optimizing.
+    # Heartbeat optimize progresses while rebuild is parked until its interval.
     await asyncio.sleep(0.12)
     # Optimize should still progress despite rebuild errors.
     assert fake.beats, "heartbeat optimize should run even when rebuild fails"