diff --git a/src/lemoncrow/gateway/adapters/mcp_server.py b/src/lemoncrow/gateway/adapters/mcp_server.py index 6dec547b7..ea23e010f 100644 --- a/src/lemoncrow/gateway/adapters/mcp_server.py +++ b/src/lemoncrow/gateway/adapters/mcp_server.py @@ -8879,6 +8879,7 @@ def _memory_summary(session_id: str) -> dict[str, Any]: # from the one before it. _code_index_freshness_for_current_call: threading.local = threading.local() + # Process-level engine cache keyed by resolved repo path. # Reusing the same engine across tool calls avoids re-opening the SQLite DB # and restarting autosync threads on every invocation — critical for both @@ -8889,7 +8890,17 @@ def _memory_summary(session_id: str) -> dict[str, Any]: # still serving it at version 23, returning empty results for every query with # no error of any kind. VersionedEngineCache stamps each entry and rebuilds on a # bump, and raises IndexRebuilding mid-reindex rather than answering with []. -_code_engine_cache = VersionedEngineCache("code_engine") +# +# A rebuild must also stop the engine it replaces. Each engine runs its own +# autosync thread, which keeps it alive, so a superseded engine kept polling +# and spawning reindexes: one more loop per index bump, until the daemon ran a +# reindex storm that held the index-write lock nearly all the time and +# code_search answered "index is being rebuilt" for most calls. +def _retire_code_engine(engine: Any) -> None: + engine.stop_autosync() + + +_code_engine_cache = VersionedEngineCache("code_engine", on_evict=_retire_code_engine) # ``cache_key -> (capability, engine_it_was_built_from)``. # diff --git a/src/lemoncrow/infra/code_intel/freshness.py b/src/lemoncrow/infra/code_intel/freshness.py index 9019fd620..ae07356b8 100644 --- a/src/lemoncrow/infra/code_intel/freshness.py +++ b/src/lemoncrow/infra/code_intel/freshness.py @@ -267,6 +267,13 @@ class VersionedEngineCache: Rebuilds happen under a lock with a double check, so N concurrent callers arriving at a version bump together produce one rebuild, not N. + + Dropping an entry does not end whatever the value started. *on_evict* is + called with every value the cache lets go of -- superseded, discarded or + cleared -- so its owner can stop it. The code engine is why this exists: its + background autosync thread references the engine, so an evicted engine never + died. Every index bump left one more loop polling the tree and spawning its + own reindex, and those reindexes bumped the version again. """ def __init__( @@ -274,9 +281,11 @@ def __init__( name: str, recheck_seconds: float = DEFAULT_RECHECK_SECONDS, clock: Callable[[], float] = time.monotonic, + on_evict: Callable[[Any], None] | None = None, ) -> None: self.name = name self.recheck_seconds = float(recheck_seconds) + self.on_evict = on_evict self.evictions = 0 self._clock = clock self._lock = threading.Lock() @@ -332,7 +341,19 @@ def get(self, key: str, repo_root: Path | str, build: Callable[[], Any]) -> tupl self.evictions += 1 value = build() self._entries[key] = _Entry(value=value, index_version=state.index_version) - return value, FRESHNESS_FRESH if superseded is None else FRESHNESS_REBUILT + # Outside the lock: retiring may join threads, and every other caller of + # this cache would wait on it. + if entry is not None: + self._retire(entry.value) + return value, FRESHNESS_FRESH if superseded is None else FRESHNESS_REBUILT + + def _retire(self, value: Any) -> None: + if self.on_evict is None: + return + try: + self.on_evict(value) + except Exception: + logger.warning("%s: retiring an evicted entry failed", self.name, exc_info=True) def peek(self, key: str) -> Any | None: """The cached value for *key* without probing or building.""" @@ -346,13 +367,18 @@ def version_of(self, key: str) -> int | None: def discard(self, key: str) -> None: with self._lock: - self._entries.pop(key, None) + entry = self._entries.pop(key, None) + if entry is not None: + self._retire(entry.value) def clear(self) -> None: with self._lock: + dropped = [entry.value for entry in self._entries.values()] self._entries.clear() self._probes.clear() self.evictions = 0 + for value in dropped: + self._retire(value) def __len__(self) -> int: return len(self._entries) diff --git a/src/lemoncrow/pro/capabilities/code_context/engine.py b/src/lemoncrow/pro/capabilities/code_context/engine.py index cb5884256..97fb75393 100644 --- a/src/lemoncrow/pro/capabilities/code_context/engine.py +++ b/src/lemoncrow/pro/capabilities/code_context/engine.py @@ -14793,6 +14793,17 @@ def _stop_autosync_worker(self) -> None: self._autosync_stop.set() self._stop_file_watcher() + def stop_autosync(self) -> None: + """Stop this engine's background index refresh, permanently. + + Queries keep working; this instance just stops polling the tree and + spawning reindexes. Call it when replacing the engine: the worker thread + references the engine, so dropping every other reference does not stop + it, and a replaced engine otherwise refreshes the index for the life of + the process. + """ + self._stop_autosync_worker() + # --- File watcher (event-driven via watchdog) --- def _start_file_watcher(self) -> None: diff --git a/tests/gateway/test_code_engine_cache_invalidation.py b/tests/gateway/test_code_engine_cache_invalidation.py index 6612a9e60..741d675d8 100644 --- a/tests/gateway/test_code_engine_cache_invalidation.py +++ b/tests/gateway/test_code_engine_cache_invalidation.py @@ -49,10 +49,14 @@ def __init__(self, root: Path) -> None: type(self).instances += 1 self.root = root self.db_path = str(root) + self.stopped = False def index_ready(self) -> bool: return True + def stop_autosync(self) -> None: + self.stopped = True + @pytest.fixture def indexed_repo(tmp_path: Path) -> Path: @@ -110,7 +114,11 @@ def _isolated_caches(monkeypatch: pytest.MonkeyPatch) -> None: _FakeEngine.instances = 0 monkeypatch.setattr(code_context, "CodeContextEngine", _FakeEngine) - monkeypatch.setattr(mcp_server, "_code_engine_cache", VersionedEngineCache("test", recheck_seconds=0.0)) + monkeypatch.setattr( + mcp_server, + "_code_engine_cache", + VersionedEngineCache("test", recheck_seconds=0.0, on_evict=mcp_server._code_engine_cache.on_evict), + ) monkeypatch.setattr(mcp_server, "_scoped_context_cache", {}) mcp_server._code_index_freshness_for_current_call.value = None mcp_server._code_engine_for_current_call.value = None @@ -140,6 +148,21 @@ def test_index_version_bump_rebuilds_the_engine(indexed_repo: Path) -> None: assert mcp_server._code_index_freshness_for_current_call.value == FRESHNESS_REBUILT +def test_a_rebuilt_engine_stops_the_one_it_replaced(indexed_repo: Path) -> None: + """A superseded engine left the cache but kept running. + + Its autosync thread held it alive, so each index bump added one more loop + polling the tree and spawning its own reindex -- a storm that kept the + index-write lock held and made code_search answer "being rebuilt". + """ + first = mcp_server._code_context_engine(str(indexed_repo)) + _bump(indexed_repo, 2) + second = mcp_server._code_context_engine(str(indexed_repo)) + + assert first.stopped, "the replaced engine was left running" + assert not second.stopped + + class _FakeScoped: """Stands in for the compiled ScopedContextCapability.""" @@ -281,11 +304,12 @@ def test_freshness_does_not_leak_into_the_next_call(indexed_repo: Path) -> None: def test_runtime_cache_reset_clears_the_engine_cache(indexed_repo: Path) -> None: - mcp_server._code_context_engine(str(indexed_repo)) + first = mcp_server._code_context_engine(str(indexed_repo)) assert len(mcp_server._code_engine_cache) == 1 mcp_server._code_engine_cache.clear() assert len(mcp_server._code_engine_cache) == 0 + assert first.stopped, "a cleared engine was left running" mcp_server._code_context_engine(str(indexed_repo)) assert _FakeEngine.instances == 2 diff --git a/tests/infra/code_intel/test_code_engine_autosync_retirement.py b/tests/infra/code_intel/test_code_engine_autosync_retirement.py new file mode 100644 index 000000000..341aa4c5a --- /dev/null +++ b/tests/infra/code_intel/test_code_engine_autosync_retirement.py @@ -0,0 +1,70 @@ +"""A daemon runs one autosync loop per repo, however often the index moves. + +Measured on a live daemon: the symphony-alpha index-write lock was held for the +whole of a sampled window, and code_search answered "index is being rebuilt" for +most calls. Every index bump had left the replaced engine's autosync loop +running, and each leaked loop spawned its own reindex per change (reproduced: six +engines, one edited file, six `lc code index` processes). The fake-engine tests +pin the wiring; this one drives the real engine through the production cache, so +the live thread count is the evidence. +""" + +from __future__ import annotations + +import sqlite3 +import subprocess +import threading +import time +from pathlib import Path + +import pytest + +from lemoncrow.gateway.adapters import mcp_server +from lemoncrow.infra.code_intel.freshness import VersionedEngineCache +from lemoncrow.infra.code_intel.store import CODE_CONTEXT_DB, workspace_dir +from lemoncrow.pro.capabilities.code_context import CodeContextEngine + + +def _live_autosync_threads(name: str) -> list[threading.Thread]: + return [t for t in threading.enumerate() if t.name == name and t.is_alive()] + + +def _bump(root: Path) -> None: + conn = sqlite3.connect(workspace_dir(root) / CODE_CONTEXT_DB) + try: + conn.execute("UPDATE engine_state SET value = CAST(value AS INTEGER) + 1 WHERE key = 'index_version'") + conn.commit() + finally: + conn.close() + + +def test_index_bumps_leave_exactly_one_autosync_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = tmp_path / "repo" + (repo / "pkg").mkdir(parents=True) + for i in range(3): + (repo / "pkg" / f"mod{i}.py").write_text(f"def f{i}():\n return {i}\n") + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + CodeContextEngine(repo, autosync_enabled=False).index_repo(force=True) + + # The suite disables autosync globally; this test is about autosync. + monkeypatch.setenv("LEMONCROW_CODE_AUTOSYNC", "1") + monkeypatch.setenv("LEMONCROW_CODE_FILE_WATCHER", "0") + cache = VersionedEngineCache("test", recheck_seconds=0.0, on_evict=mcp_server._code_engine_cache.on_evict) + + engines: list[CodeContextEngine] = [] + try: + for _ in range(4): + engine, _freshness = cache.get(str(repo), repo, lambda: CodeContextEngine(repo)) + engines.append(engine) + _bump(repo) + assert len({id(e) for e in engines}) == 4, "the index bumps did not rebuild the engine" + + name = f"lemoncrow-code-autosync-{engines[0].repo_id[:8]}" + deadline = time.monotonic() + 10 + while len(_live_autosync_threads(name)) > 1 and time.monotonic() < deadline: + time.sleep(0.05) + + assert len(_live_autosync_threads(name)) == 1, "replaced engines kept their autosync loops" + finally: + for engine in engines: + engine.stop_autosync() diff --git a/tests/infra/code_intel/test_freshness.py b/tests/infra/code_intel/test_freshness.py index 67c4d0b3d..a52fdc0e3 100644 --- a/tests/infra/code_intel/test_freshness.py +++ b/tests/infra/code_intel/test_freshness.py @@ -198,6 +198,53 @@ def test_version_bump_evicts_and_rebuilds(make_workspace: WorkspaceFactory) -> N assert cache.evictions == 1 +def test_a_superseded_entry_is_retired_once(make_workspace: WorkspaceFactory) -> None: + """Eviction has to end what the evicted value started, not just forget it.""" + root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=1) + retired: list[object] = [] + cache = VersionedEngineCache("test", recheck_seconds=0.0, on_evict=retired.append) + + first, _ = cache.get("k", root, object) + cache.get("k", root, object) + assert retired == [], "a stable index retired its own live entry" + + _set_index_version(root, 2) + second, _ = cache.get("k", root, object) + + assert retired == [first] + assert second is not first + + +def test_discard_and_clear_retire_what_they_drop(make_workspace: WorkspaceFactory) -> None: + root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=1) + retired: list[object] = [] + cache = VersionedEngineCache("test", recheck_seconds=0.0, on_evict=retired.append) + + discarded, _ = cache.get("a", root, object) + cache.discard("a") + cleared, _ = cache.get("b", root, object) + cache.clear() + + assert retired == [discarded, cleared] + + +def test_a_failing_retirement_does_not_fail_the_lookup(make_workspace: WorkspaceFactory) -> None: + """The rebuilt value is already cached; a stop hook that raises must not lose it.""" + root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=1) + + def refuse(_value: object) -> None: + raise RuntimeError("stop failed") + + cache = VersionedEngineCache("test", recheck_seconds=0.0, on_evict=refuse) + first, _ = cache.get("k", root, object) + _set_index_version(root, 2) + second, freshness = cache.get("k", root, object) + + assert second is not first + assert freshness == FRESHNESS_REBUILT + assert cache.peek("k") is second + + def test_first_build_is_fresh_not_rebuilt(make_workspace: WorkspaceFactory) -> None: """Nothing was superseded on a cold cache; only an eviction is 'rebuilt'.""" root = make_workspace(files=_FILES, symbols=_SYMBOLS)