diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c0b4eb24..3ff9416fa 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/cascade_runbook.md b/docs/cascade_runbook.md index 85607240c..d43f661ae 100644 --- a/docs/cascade_runbook.md +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 95d1f652a..4eccb25d0 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/everos/entrypoints/api/routes/cascade.py b/src/everos/entrypoints/api/routes/cascade.py index 7b5d8bf95..9fc4ced0a 100644 --- a/src/everos/entrypoints/api/routes/cascade.py +++ b/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() diff --git a/src/everos/entrypoints/cli/commands/cascade.py b/src/everos/entrypoints/cli/commands/cascade.py index d59a55fb6..aedae4d24 100644 --- a/src/everos/entrypoints/cli/commands/cascade.py +++ b/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: diff --git a/src/everos/infra/persistence/sqlite/repos/md_change_state.py b/src/everos/infra/persistence/sqlite/repos/md_change_state.py index 6ba49edaa..51cf65121 100644 --- a/src/everos/infra/persistence/sqlite/repos/md_change_state.py +++ b/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``). diff --git a/src/everos/memory/cascade/orchestrator.py b/src/everos/memory/cascade/orchestrator.py index 3e8bdf342..57bf4ec6e 100644 --- a/src/everos/memory/cascade/orchestrator.py +++ b/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() diff --git a/tests/integration/test_cascade_cli_integration.py b/tests/integration/test_cascade_cli_integration.py index 03ef62a96..86aab7c7a 100644 --- a/tests/integration/test_cascade_cli_integration.py +++ b/tests/integration/test_cascade_cli_integration.py @@ -372,12 +372,8 @@ 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 @@ -385,7 +381,7 @@ async def drain_once(self) -> int: 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 diff --git a/tests/unit/test_entrypoints/test_api/test_routes/test_cascade_quiesce_route.py b/tests/unit/test_entrypoints/test_api/test_routes/test_cascade_quiesce_route.py index 4a945d279..5af2ba999 100644 --- a/tests/unit/test_entrypoints/test_api/test_routes/test_cascade_quiesce_route.py +++ b/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 diff --git a/tests/unit/test_entrypoints/test_cli/test_cascade_drain_fully.py b/tests/unit/test_entrypoints/test_cli/test_cascade_drain_fully.py deleted file mode 100644 index 7f49aa9a5..000000000 --- a/tests/unit/test_entrypoints/test_cli/test_cascade_drain_fully.py +++ /dev/null @@ -1,41 +0,0 @@ -"""``cascade sync`` / ``rebuild`` drain until a pass processes nothing. - -The worker bounds one ``drain_until_empty`` (``max_passes`` × batch size). -A ledger the one-time reconcile queues after an upgrade can exceed that -bound, and the command must not report success with rows still pending. -""" - -from __future__ import annotations - -from everos.entrypoints.cli.commands.cascade import _drain_fully - - -class _BoundedOrchestrator: - """Stands in for the real one: a drain returns at most ``bound`` rows.""" - - def __init__(self, queued: int, bound: int) -> None: - self.queued = queued - self.bound = bound - self.drains = 0 - - async def sync_once(self) -> int: - return await self.drain_once() - - async def drain_once(self) -> int: - self.drains += 1 - took = min(self.queued, self.bound) - self.queued -= took - return took - - -async def test_drain_fully_keeps_draining_past_the_worker_bound() -> None: - orchestrator = _BoundedOrchestrator(queued=5003, bound=5000) - assert await _drain_fully(orchestrator) == 5003 # type: ignore[arg-type] - assert orchestrator.queued == 0 - assert orchestrator.drains == 3 # 5000, 3, then the empty pass that stops it - - -async def test_drain_fully_stops_on_an_empty_queue() -> None: - orchestrator = _BoundedOrchestrator(queued=0, bound=5000) - assert await _drain_fully(orchestrator) == 0 # type: ignore[arg-type] - assert orchestrator.drains == 2 # the scan's drain and one confirming pass diff --git a/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py b/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py index b5972fd82..022d97ac9 100644 --- a/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py +++ b/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py @@ -181,6 +181,9 @@ async def test_force_enqueue_all_requeues_every_row_and_keeps_its_mtime( assert row.change_type == "modified" assert row.mtime == mtime assert row.retry_count == 0 + after = {r.md_path: r for r in await repo.list_all()} + assert after[done].lsn > max(r.lsn for r in before) + assert after[failed].lsn == after[done].lsn + 1 # oldest first, one LSN each async def test_list_all_returns_every_status_in_lsn_order( diff --git a/tests/unit/test_memory/test_cascade/test_orchestrator.py b/tests/unit/test_memory/test_cascade/test_orchestrator.py index 22fb3a14e..eedc91b56 100644 --- a/tests/unit/test_memory/test_cascade/test_orchestrator.py +++ b/tests/unit/test_memory/test_cascade/test_orchestrator.py @@ -2,6 +2,7 @@ from __future__ import annotations +import dataclasses from collections.abc import AsyncIterator from pathlib import Path from unittest.mock import patch @@ -296,3 +297,115 @@ async def test_watcher_can_be_disabled_without_losing_the_worker( assert orch._worker is not None assert orch._scanner is not None await orch.stop() + + +class _SharedQueue: + """A ledger two drainers claim from: this orchestrator's passes and, inside + a server, the background worker. ``held`` is the batch the worker has + claimed (``processing``), which ``pending`` counts too.""" + + def __init__(self, rows: int, bound: int = 5000) -> None: + self.rows, self.bound, self.held = rows, bound, 0 + self.drains: list[int] = [] + self.summaries: list[int] = [] + + async def drain(self, **_: object) -> int: + took = min(self.rows, self.bound) + self.rows -= took + self.drains.append(took) + return took + + async def summary(self) -> object: + pending = self.rows + self.held + self.summaries.append(pending) + return type("S", (), {"pending": pending})() + + +def _patched(orch: CascadeOrchestrator, q: _SharedQueue) -> tuple: + return ( + patch.object(orch, "sync_once", side_effect=q.drain), + patch.object(orch, "drain_once", side_effect=q.drain), + patch.object(orch, "queue_summary", side_effect=q.summary), + ) + + +async def test_drain_fully_keeps_draining_past_the_worker_bound( + runtime: MemoryRoot, +) -> None: + """The worker bounds one drain (``max_passes`` x batch size). The ledger + the one-time reconcile queues after an upgrade can exceed that bound, and + a caller that stops there reports success with rows still pending. + """ + orch = _make_orchestrator(runtime) + q = _SharedQueue(rows=5003) + a, b, c = _patched(orch, q) + with a, b, c: + assert await orch.drain_fully() == 5003 + assert q.rows == 0 + assert q.drains == [ + 5000, + 3, + ] # the scan's drain, then the rest; the ledger says when to stop + + +async def test_drain_fully_waits_for_a_batch_another_drainer_holds( + runtime: MemoryRoot, +) -> None: + """Inside a server the background worker claims batches too. 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 + (real host, 12000-row ledger: 50 rows left behind). So the ledger, not + our own pass, says when the drain is done. + """ + orch = _make_orchestrator(runtime) + q = _SharedQueue(rows=5050) + a, b, c = _patched(orch, q) + with a, b, c: + + async def first_pass(**_: object) -> int: + took = await q.drain() # 5000 by us; the worker claims the last 50 + q.rows, q.held = 0, 50 + return took + + async def worker_finishes_later(**_: object) -> int: + # our pass finds nothing pending; the worker's batch lands two polls later + if len(q.drains) >= 3: + q.held = 0 + q.drains.append(0) + return 0 + + orch.sync_once.side_effect = first_pass + orch.drain_once.side_effect = worker_finishes_later + orch._config = dataclasses.replace( + orch._config, worker_poll_interval_seconds=0.01 + ) + processed = await orch.drain_fully() + + assert q.summaries and q.summaries[-1] == 0, ( + "returned while the worker held a batch" + ) + assert processed == 5000 + + +async def test_drain_fully_gives_up_when_the_ledger_stops_moving( + runtime: MemoryRoot, +) -> None: + """A row stuck in ``processing`` (a hung handler) must not hang the caller + forever; after the idle limit the drain returns and the caller reports + what is left.""" + orch = _make_orchestrator(runtime) + q = _SharedQueue(rows=0) + q.held = 1 + a, b, c = _patched(orch, q) + with ( + a, + b, + c, + patch("everos.memory.cascade.orchestrator._DRAIN_IDLE_LIMIT_SECONDS", 0.05), + ): + orch._config = dataclasses.replace( + orch._config, worker_poll_interval_seconds=0.01 + ) + assert await orch.drain_fully() == 0 + assert q.summaries[-1] == 1 # gave up with the row still held, and said so