fix(cascade): quiesce drains fully, the reconcile pass commits once

CHANGELOG.md

@@ -28,7 +28,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `.index/.index_reconcile_version`. Entries whose rows are present and
   unchanged are skipped by digest, so a healthy root pays one read pass
   and no embedding; only the rows that were lost are re-embedded.
-  `everos cascade sync --force-all` runs the same pass on demand.
+  `everos cascade sync --force-all` runs the same pass on demand. Two
+  roots need that command instead: a server started with
+  `EVEROS_DISABLE_CASCADE=1` never runs the pass (stop it and run
+  `everos cascade sync --force-all` once), and a multi-project root where
+  reflection ran before this release needs `everos cascade rebuild` once,
+  because the pass does not clear the `deprecated_by` stamps one project's
+  reflection left on another project's rows.
+
+- **`POST /cascade/quiesce` freezes a complete index.** Its drain stopped
+  at the worker's per-drain bound, and inside a server it could also stop
+  while the background worker still held the last batch, which the stop
+  that follows then orphaned in `processing`; the ledger an upgrade
+  re-queues made both routine. The drain now runs until the ledger holds
+  nothing pending, waiting for the worker's in-flight batch, and
+  `everos cascade sync` / `rebuild` use the same loop.
 
 - **Changing a knowledge document's category keeps it inside `knowledge/`.**
   `PATCH /knowledge/documents/{doc_id}` with `category_id` set to `..` moved

docs/cascade_runbook.md

@@ -133,7 +133,13 @@ upgrading from a release before 1.4.2 (a server start, or an unscoped
 `everos cascade sync`) force-enqueues every ledger row and writes
 `.index/.index_reconcile_version`, so no upgrade step is needed and later
 starts are ordinary. Delete the marker to make the next start repeat it,
-or run `--force-all` for an explicit pass.
+or run `--force-all` for an explicit pass. Two roots need the explicit
+pass: a server started with `EVEROS_DISABLE_CASCADE=1` never runs it
+(stop the server, run `--force-all` once), and a multi-project root where
+reflection ran before 1.4.2 needs `cascade rebuild` once -- the pass does
+not clear the `deprecated_by` stamps one project's reflection left on
+another project's rows, so those rows stay hidden from search until the
+index is rebuilt from markdown.
 
 The CLI builds the same `CascadeOrchestrator` as the daemon but only
 calls `sync_once` / `drain_once` — no watcher / scanner background task.

pyproject.toml

@@ -268,7 +268,6 @@ forbidden_modules = [
     "everos.infra.persistence.index.**",
     "everos.infra.persistence.records.**",
     "everos.infra.persistence.milvus.**",
-    "everos.infra.persistence.backends.**",
 ]
 # `forbidden` contracts run a *transitive closure* — any path from a
 # source module to a forbidden module fails the check, even when the

src/everos/entrypoints/api/routes/cascade.py

@@ -91,11 +91,13 @@ async def quiesce(request: Request) -> QuiesceResponse:
         )
 
     before = await orch.queue_summary()
-    # Drain BEFORE stopping. `sync_once` is a full scan + drain, so it also picks
-    # up markdown the watcher never saw -- which matters here because the watcher
-    # is the half most likely to be off (inotify watches are a per-user kernel
-    # resource a shared host can exhaust).
-    drained = await orch.sync_once()
+    # Drain BEFORE stopping. `drain_fully` is a full scan + drain to empty, so it
+    # also picks up markdown the watcher never saw -- which matters here because
+    # the watcher is the half most likely to be off (inotify watches are a
+    # per-user kernel resource a shared host can exhaust) -- and it does not stop
+    # at the worker's per-drain bound, which the ledger an upgrade re-queues can
+    # exceed.
+    drained = await orch.drain_fully()
     await orch.stop()
     after = await orch.queue_summary()
 

src/everos/entrypoints/cli/commands/cascade.py

@@ -224,19 +224,6 @@ def _build_orchestrator() -> CascadeOrchestrator:
     )
 
 
-async def _drain_fully(orchestrator: CascadeOrchestrator) -> int:
-    """One scan, then drain until a pass processes nothing.
-
-    The worker bounds a single drain (``max_passes`` × batch size); the
-    ledger the one-time reconcile queues after an upgrade can exceed it,
-    and the command must not report success with rows still pending.
-    """
-    processed = await orchestrator.sync_once()
-    while more := await orchestrator.drain_once():
-        processed += more
-    return processed
-
-
 # ── sync ─────────────────────────────────────────────────────────────────
 
 
@@ -306,7 +293,7 @@ async def _run() -> None:
                     raise typer.Exit(code=1)
                 await md_change_state_repo.force_enqueue(rel, spec.name)
                 typer.echo(f"force-enqueued {rel} (kind={spec.name})")
-            processed = await _drain_fully(orchestrator)
+            processed = await orchestrator.drain_fully()
             typer.echo(f"sync complete — processed {processed} row(s)")
 
     asyncio.run(_run())
@@ -585,7 +572,7 @@ async def _run() -> None:
             await ensure_business_indexes()
             # Re-scan + drain: re-embed and re-insert every md entry.
             orchestrator = _build_orchestrator()
-            processed = await _drain_fully(orchestrator)
+            processed = await orchestrator.drain_fully()
             typer.echo(f"rebuild complete — re-indexed {processed} md file(s)")
 
     try:

src/everos/infra/persistence/sqlite/repos/md_change_state.py

@@ -28,6 +28,8 @@
 from __future__ import annotations
 
 import dataclasses
+from datetime import datetime
+from typing import Any
 
 from sqlalchemy import delete, func, select, text, update
 from sqlalchemy.dialects.sqlite import insert as sqlite_insert
@@ -129,49 +131,17 @@ async def upsert(
         now = get_utc_now()
         async with session_scope(self._factory) as s:
             new_lsn = await _next_lsn(s)
-            stmt = (
-                sqlite_insert(MdChangeState)
-                .values(
-                    md_path=md_path,
+            await s.execute(
+                _enqueue_stmt(
+                    md_path,
                     kind=kind,
                     change_type=change_type,
                     mtime=mtime,
-                    first_seen_at=now,
-                    last_changed_at=now,
                     lsn=new_lsn,
-                    status="pending",
-                    retryable=None,
-                    last_attempt_at=None,
-                    retry_count=0,
-                    error=None,
-                )
-                .on_conflict_do_update(
-                    index_elements=["md_path"],
-                    set_={
-                        "kind": kind,
-                        "change_type": change_type,
-                        "mtime": mtime,
-                        "last_changed_at": now,
-                        "lsn": new_lsn,
-                        "status": "pending",
-                        "retryable": None,
-                        "last_attempt_at": None,
-                        "error": None,
-                        "retry_count": 0
-                        if reset_retry_count
-                        else text(
-                            "CASE"
-                            " WHEN md_change_state.status = 'failed'"
-                            " AND ABS(md_change_state.mtime - :new_mtime)"
-                            f" < {MTIME_TOLERANCE_SECONDS}"
-                            " THEN md_change_state.retry_count"
-                            " ELSE 0"
-                            " END"
-                        ).bindparams(new_mtime=mtime),
-                    },
+                    now=now,
+                    reset_retry_count=reset_retry_count,
                 )
             )
-            await s.execute(stmt)
             await s.commit()
             return new_lsn
 
@@ -207,10 +177,30 @@ async def force_enqueue_all(self) -> list[MdChangeState]:
         status filter on purpose — ``failed`` rows must come back too.
         """
         rows = await self.list_all()
-        # ponytail: one upsert (one commit) per file; batch into a single
-        # session if a root with tens of thousands of md files makes startup wait.
-        for row in rows:
-            await self.force_enqueue(row.md_path, row.kind, mtime=row.mtime)
+        now = get_utc_now()
+        # One session, one commit: a commit per row cost ~0.9 ms each and held
+        # a server's startup for the whole pass on a root with tens of
+        # thousands of md files; one round trip per row is ~0.3 ms. Each row
+        # keeps its mtime (the scanner does not re-pick it) and gets a fresh
+        # LSN in ledger order.
+        # ponytail: still one execute per row; a chunked multi-row INSERT on
+        # `excluded` if a 100k-file root makes startup wait.
+        async with session_scope(self._factory) as s:
+            lsn = await _next_lsn(s)
+            for row in rows:
+                await s.execute(
+                    _enqueue_stmt(
+                        row.md_path,
+                        kind=row.kind,
+                        change_type="modified",
+                        mtime=row.mtime,
+                        lsn=lsn,
+                        now=now,
+                        reset_retry_count=True,
+                    )
+                )
+                lsn += 1
+            await s.commit()
         return rows
 
     # ── Worker claim ───────────────────────────────────────────────────────
@@ -517,6 +507,65 @@ async def queue_summary(self) -> QueueSummary:
         )
 
 
+def _enqueue_stmt(
+    md_path: str,
+    *,
+    kind: str,
+    change_type: str,
+    mtime: float,
+    lsn: int,
+    now: datetime,
+    reset_retry_count: bool,
+) -> Any:
+    """The INSERT ... ON CONFLICT that :meth:`upsert` documents.
+
+    Shared with :meth:`force_enqueue_all`, which issues one per ledger row
+    inside a single session.
+    """
+    return (
+        sqlite_insert(MdChangeState)
+        .values(
+            md_path=md_path,
+            kind=kind,
+            change_type=change_type,
+            mtime=mtime,
+            first_seen_at=now,
+            last_changed_at=now,
+            lsn=lsn,
+            status="pending",
+            retryable=None,
+            last_attempt_at=None,
+            retry_count=0,
+            error=None,
+        )
+        .on_conflict_do_update(
+            index_elements=["md_path"],
+            set_={
+                "kind": kind,
+                "change_type": change_type,
+                "mtime": mtime,
+                "last_changed_at": now,
+                "lsn": lsn,
+                "status": "pending",
+                "retryable": None,
+                "last_attempt_at": None,
+                "error": None,
+                "retry_count": 0
+                if reset_retry_count
+                else text(
+                    "CASE"
+                    " WHEN md_change_state.status = 'failed'"
+                    " AND ABS(md_change_state.mtime - :new_mtime)"
+                    f" < {MTIME_TOLERANCE_SECONDS}"
+                    " THEN md_change_state.retry_count"
+                    " ELSE 0"
+                    " END"
+                ).bindparams(new_mtime=mtime),
+            },
+        )
+    )
+
+
 async def _next_lsn(session: AsyncSession) -> int:
     """Pick the next global LSN (``MAX(lsn) + 1``).
 

src/everos/memory/cascade/orchestrator.py

@@ -55,6 +55,10 @@
 bump here."""
 
 _RECONCILE_MARKER = ".index_reconcile_version"
+_DRAIN_IDLE_LIMIT_SECONDS = 120.0
+"""How long :meth:`CascadeOrchestrator.drain_fully` keeps polling a ledger
+that holds rows nobody is landing (a handler hung mid-batch) before it
+returns and lets the caller report what is left."""
 
 
 async def reconcile_index_once(
@@ -311,6 +315,42 @@ async def drain_once(self) -> int:
         """Drain the queue exactly once without scanning first."""
         return await self._worker.drain_until_empty()
 
+    async def drain_fully(self) -> int:
+        """One scan, then drain until the ledger holds nothing pending.
+
+        Two things end a single drain early. The worker bounds one
+        ``drain_until_empty`` (``max_passes`` x batch size), and the ledger
+        the one-time reconcile queues after an upgrade can exceed it. Inside
+        a server the background worker claims batches too, so a pass of ours
+        can find nothing pending while the worker still holds the last
+        batch; ``POST /cascade/quiesce`` cancels the worker right after,
+        which would orphan that batch in ``processing`` and freeze a partial
+        index (seen on a real host: 50 rows left of 12000). So the ledger,
+        not our own pass, says when the drain is done -- ``pending`` counts
+        ``processing`` rows -- and a pass that finds nothing waits one poll
+        interval for the other drainer. A ledger that stops moving for
+        :data:`_DRAIN_IDLE_LIMIT_SECONDS` ends the wait; the caller reports
+        the rows left.
+        """
+        processed = await self.sync_once()
+        idle_seconds = 0.0
+        while (await self.queue_summary()).pending:
+            more = await self.drain_once()
+            processed += more
+            if more:
+                idle_seconds = 0.0
+                continue
+            if idle_seconds >= _DRAIN_IDLE_LIMIT_SECONDS:
+                logger.warning(
+                    "cascade_drain_fully_gave_up",
+                    idle_seconds=idle_seconds,
+                    reason="rows stayed pending or processing with nothing landing",
+                )
+                break
+            await asyncio.sleep(self._config.worker_poll_interval_seconds)
+            idle_seconds += self._config.worker_poll_interval_seconds
+        return processed
+
     async def queue_summary(self) -> QueueSummary:
         """Forward to the repo so callers don't reach past this class."""
         return await md_change_state_repo.queue_summary()

tests/integration/test_cascade_cli_integration.py

@@ -372,20 +372,16 @@ def test_sync_holds_the_lock_while_it_drains(
     seen: list[bool] = []
 
     class _Orchestrator:
-        async def sync_once(self) -> int:
-            seen.append(cascade_mod.ome_lock_is_free())
-            return 0
-
-        async def drain_once(self) -> int:
-            # the CLI keeps draining until a pass processes nothing
+        async def drain_fully(self) -> int:
+            # scan + every drain pass happen inside this one call
             seen.append(cascade_mod.ome_lock_is_free())
             return 0
 
     monkeypatch.setattr(cascade_mod, "_build_orchestrator", lambda: _Orchestrator())
     result = CliRunner().invoke(cascade_mod.app, ["sync"])
 
     assert result.exit_code == 0, result.output + result.stderr
-    assert seen == [False, False]  # the lock was ours for the scan and every drain
+    assert seen == [False]  # the lock was ours for the whole drain
     assert cascade_mod.ome_lock_is_free()  # and released afterwards
 
 

tests/unit/test_entrypoints/test_api/test_routes/test_cascade_quiesce_route.py

@@ -48,7 +48,7 @@ def _orch(*, before: int, after: int, drained: int) -> mock.MagicMock:
         _Summary(pending=before),
         _Summary(pending=after, failed_permanent=2),
     ]
-    orch.sync_once.return_value = drained
+    orch.drain_fully.return_value = drained
     return orch
 
 
@@ -64,7 +64,7 @@ async def test_drains_before_stopping() -> None:
     then reads as a healthy store that quietly under-recalls."""
     orch = _orch(before=7, after=0, drained=7)
     order: list[str] = []
-    orch.sync_once.side_effect = lambda **_: order.append("drain") or 7
+    orch.drain_fully.side_effect = lambda: order.append("drain") or 7
     orch.stop.side_effect = lambda: order.append("stop")
 
     async with _client(orch) as c:
@@ -74,6 +74,20 @@ async def test_drains_before_stopping() -> None:
     assert order == ["drain", "stop"], "stopped before draining"
 
 
+async def test_drains_past_the_worker_bound() -> None:
+    """One ``sync_once`` stops at the worker's bound (``max_passes`` x batch
+    size). The ledger the one-time reconcile queues after an upgrade can exceed
+    it, and a quiesce that stopped there would freeze a partial index while
+    reporting ``quiesced=true``."""
+    orch = _orch(before=5003, after=0, drained=5003)
+    async with _client(orch) as c:
+        resp = await c.post("/api/v1/cascade/quiesce")
+
+    assert resp.json()["drained"] == 5003
+    assert orch.drain_fully.await_count == 1
+    assert orch.sync_once.await_count == 0
+
+
 async def test_reports_the_queue_on_both_sides() -> None:
     """``pending_before`` is how a caller discovers the projection was behind --
     which the benchmark's ``add.done`` marker does not tell it, because that