diff --git a/codebase_rag/cli.py b/codebase_rag/cli.py index 9a91d84e4c..03af8c47ed 100644 --- a/codebase_rag/cli.py +++ b/codebase_rag/cli.py @@ -81,6 +81,8 @@ from .sync_guard import clear_sync_incomplete as _clear_sync_incomplete from .sync_guard import partial_graph_refusal as _partial_graph_refusal from .sync_guard import project_owner_refusal as _project_owner_refusal +from .sync_guard import retire_stranded_runs as _retire_stranded_runs +from .sync_guard import stranded_runs as _stranded_runs from .sync_guard import sync_marker_params as _sync_marker_params from .sync_guard import sync_owner as _sync_owner from .sync_lock import SyncLockError, repo_sync_lock @@ -1107,16 +1109,17 @@ def _sweep_glosses_after_failed_delete( def _run_updater_deferring_interrupt( - updater: "GraphUpdater", + updater: "GraphUpdater", *, rebuild: bool = False ) -> KeyboardInterrupt | None: """Run the sync's graph update; return a Ctrl+C that landed after its commit. The run stamps the sync time as part of its commit (#2444), and the caller clears its marker before re-raising what this returns; a Ctrl+C - before the commit raises `SyncInterrupted` instead. + before the commit raises `SyncInterrupted` instead. `rebuild` forces a + full re-index over runs another writer left part-way. """ try: - updater.run() + updater.run(force=rebuild) except ex.EmbeddingsInterrupted as stop: # Raised only after the run committed, so the graph is whole and # the sync is recorded like any other; the interrupt then ends @@ -1186,6 +1189,15 @@ def _run_graph_sync( # hits and can map onto unrelated nodes in the rebuilt graph. clear_all_embeddings() + # Read before this sync marks: the runs whose markers its rebuild + # supersedes (review of PR #3493). + stranded = _stranded_runs(ingestor, project_name, cs.CLI_SYNC_RUN_ID) + if stranded: + logger.warning( + ls.REBUILD_OVER_STRANDED_RUNS.format( + project=project_name, count=len(stranded) + ) + ) # After the wipe, which would delete the marker with everything else, # and before `ensure_constraints`, whose migration can already purge. _mark_sync_incomplete(ingestor, project_name, owner, owned_roots) @@ -1205,8 +1217,9 @@ def _run_graph_sync( capture=_capture_selection(capture), skip_embeddings=skip_embeddings, ) - interrupted = _run_updater_deferring_interrupt(updater) + interrupted = _run_updater_deferring_interrupt(updater, rebuild=bool(stranded)) _clear_sync_incomplete(ingestor, project_name) + _retire_stranded_runs(ingestor, project_name, stranded) # Taken before the export: counting it made a no-op sync of a small # repo report the time to dump the whole shared graph (issue #2440). elapsed = time.monotonic() - elapsed diff --git a/codebase_rag/constants/graph.py b/codebase_rag/constants/graph.py index 8736fb5720..bc3ee0a31a 100644 --- a/codebase_rag/constants/graph.py +++ b/codebase_rag/constants/graph.py @@ -283,7 +283,9 @@ class GoImplementersUnknown(StrEnum): # Who wrote the CLI sync's incomplete-run marker (PR #2532 review): the host # (machine and pid namespace) and pid of its process, and a token unique to # that one sync, so `delete-project` clears it only once that sync provably -# stopped, and only if no newer sync has re-marked it since. +# stopped, and only if no newer sync has re-marked it since. The MCP +# server's and `cgr check`'s markers name the host and pid too, so a full +# rebuild retires one only once its process exited (review of PR #3496). KEY_OWNER_HOST = "owner_host" KEY_OWNER_PID = "owner_pid" KEY_OWNER_TOKEN = "owner_token" diff --git a/codebase_rag/cypher_queries.py b/codebase_rag/cypher_queries.py index 4560b0d1bd..b53a6eb25f 100644 --- a/codebase_rag/cypher_queries.py +++ b/codebase_rag/cypher_queries.py @@ -198,6 +198,13 @@ CYPHER_MARK_CLI_SYNC_INCOMPLETE = ( CYPHER_MARK_PROJECT_INCOMPLETE + ", " + _CLI_SYNC_MARKER_OWNER_SET ) +# The MCP server's runs and `cgr check` name their process the same way +# (review of PR #3496), so a full rebuild can tell a run whose process exited +# from one still writing. No token: each of these runs has an id of its own. +CYPHER_MARK_RUN_INCOMPLETE = ( + CYPHER_MARK_PROJECT_INCOMPLETE + + ", m.owner_host = $owner_host, m.owner_pid = $owner_pid" +) CYPHER_CLI_SYNC_MARKER_OWNER = ( "MATCH (m:IncompleteRun {project: $project_name, run_id: $run_id}) " "RETURN m.owner_host AS owner_host, m.owner_pid AS owner_pid, " @@ -260,6 +267,29 @@ "RETURN coalesce(m.run_incomplete, false) AS run_incomplete, " "coalesce(m.writing, true) AS writing" ) +# The runs a full rebuild of the project supersedes (review of PR #3493): no +# other path removes a marker its run left after its first write, so one +# whose process died stayed down for good. Read before the rebuild marks, +# with the process each marker names: `sync_guard.stranded_runs` keeps only +# the runs whose process provably exited, since the checkout's sync lock +# keeps out no writer of another checkout or CGR_HOME under the same project +# name (review of PR #3496). A marker from before run ids existed is left +# alone, as `CYPHER_CLEAR_LEGACY_PROJECT_INCOMPLETE` leaves it. +CYPHER_STRANDED_RUN_IDS = ( + "MATCH (m:IncompleteRun {project: $project_name}) " + "WHERE m.run_id IS NOT NULL AND m.run_id <> $run_id " + "AND coalesce(m.writing, true) " + "RETURN m.run_id AS run_id, m.owner_host AS owner_host, " + "m.owner_pid AS owner_pid" +) +# Deletes one run `CYPHER_STRANDED_RUN_IDS` returned, once the forced rebuild +# that supersedes it has committed and cleared its own marker, and only while +# the marker still names the exited process: the CLI sync's run id is shared, +# so a CLI writer that started since has marked it under its own. +CYPHER_RETIRE_STRANDED_RUN = ( + "MATCH (m:IncompleteRun {project: $project_name, run_id: $run_id}) " + "WHERE m.owner_host = $owner_host AND m.owner_pid = $owner_pid DELETE m" +) # Every project with an outstanding marker, whichever path wrote it: `cgr # status` reports them so a sync that was interrupted is visible there, not diff --git a/codebase_rag/logs.py b/codebase_rag/logs.py index 098305b9fe..1f36adc403 100644 --- a/codebase_rag/logs.py +++ b/codebase_rag/logs.py @@ -1206,6 +1206,23 @@ "The sync of {project} finished, but its incomplete-run marker could not " "be cleared: {error}. The graph is complete; the next sync clears it." ) +REBUILD_OVER_STRANDED_RUNS = ( + "{count} earlier run(s) that wrote to project {project} stopped part-way, " + "so its graph may be partial. Re-indexing the whole repository instead of " + "only what changed." +) +STRANDED_RUNS_RETIRED = ( + "Cleared {count} incomplete-run marker(s) the rebuild of {project} superseded." +) +STRANDED_RUNS_UNREADABLE = ( + "Could not read which earlier runs left project {project} partial " + "({error}); their markers stay until a later full sync." +) +STRANDED_RUNS_NOT_RETIRED = ( + "The rebuild of {project} finished, but the incomplete-run markers it " + "superseded could not be cleared: {error}. The graph is complete; the next " + "full sync clears them." +) CLI_DELETE_MARKER_NOT_CLEARED = ( "Project {project} was deleted, but its incomplete-run marker could not be " "cleared: {error}. cgr status may list it as interrupted until it is synced " diff --git a/codebase_rag/mcp/tools.py b/codebase_rag/mcp/tools.py index ab0dd08e9a..8207d2420c 100644 --- a/codebase_rag/mcp/tools.py +++ b/codebase_rag/mcp/tools.py @@ -32,6 +32,7 @@ run_to_text_answer, ) from codebase_rag.services.provenance import head_commit +from codebase_rag.sync_guard import retire_stranded_runs, run_owner, stranded_runs from codebase_rag.sync_lock import repo_sync_lock from codebase_rag.tools import tool_descriptions as td from codebase_rag.tools.ast_grep_service import AstGrepService @@ -1505,6 +1506,11 @@ def _index_repository_sync(self) -> str: # a marker saying so lets a fresh process clear it and carry on rather # than refuse until a full update (#1705 review). The phase advances # to writing immediately before `delete_project`. + # + # The runs whose markers this re-index supersedes are read first + # (review of PR #3493). It deletes the project before rebuilding it, + # so it is already the full rebuild they need. + stranded = stranded_runs(self.ingestor, project_name, self._run_id) if (refusal := self._require_marker(project_name, writing=False)) is not None: raise RuntimeError(refusal) self._cleanup_embeddings_then_begin_writing(project_name) @@ -1536,6 +1542,7 @@ def _index_repository_sync(self) -> str: # (#1705 review). if (stuck := self._require_marker_cleared(project_name)) is not None: raise RuntimeError(stuck) + retire_stranded_runs(self.ingestor, project_name, stranded) self._live_updater = updater return cs.MCP_INDEX_SUCCESS_PROJECT.format( @@ -1604,6 +1611,10 @@ def _update_repository_sync(self) -> str: # Aborts on failure because the marker is the only protection that # survives a restart; nothing has been touched yet, so stopping costs # nothing. + # + # The runs whose markers this update supersedes are read first, and + # force a full re-index (review of PR #3493). + stranded = stranded_runs(self.ingestor, project_name, self._run_id) if (refusal := self._require_marker(project_name)) is not None: raise RuntimeError(refusal) self.ingestor.ensure_constraints() @@ -1619,7 +1630,13 @@ def _update_repository_sync(self) -> str: exclude_paths=exclude_paths, project_name=project_name, ) - updater.run() + if stranded: + logger.warning( + lg.REBUILD_OVER_STRANDED_RUNS.format( + project=project_name, count=len(stranded) + ) + ) + updater.run(force=bool(stranded)) self.ingestor.flush_all() # Cleared only after the flush: the marker must outlive any failure # that leaves batches uncommitted. @@ -1632,6 +1649,7 @@ def _update_repository_sync(self) -> str: # (#1705 review). if (stuck := self._require_marker_cleared(project_name)) is not None: raise RuntimeError(stuck) + retire_stranded_runs(self.ingestor, project_name, stranded) self._live_updater = updater return cs.MCP_UPDATE_SUCCESS.format(path=self.project_root) @@ -1696,7 +1714,10 @@ def _persist_incomplete( cs.KEY_RUN_ID: self._run_id, } if incomplete: - query = cq.CYPHER_MARK_PROJECT_INCOMPLETE + # Naming this process lets a full rebuild retire the marker once + # it has exited without clearing it (review of PR #3496). + query = cq.CYPHER_MARK_RUN_INCOMPLETE + params |= run_owner() params[cs.KEY_WRITING] = writing else: query = cq.CYPHER_CLEAR_PROJECT_INCOMPLETE diff --git a/codebase_rag/structural_check.py b/codebase_rag/structural_check.py index 92ae41f19e..b12b816277 100644 --- a/codebase_rag/structural_check.py +++ b/codebase_rag/structural_check.py @@ -38,6 +38,7 @@ from .graph_updater import GraphUpdater, _load_exclusion_state, _load_project_stamps from .services import QueryingIngestorProtocol from .structural_delta import StructuralDelta, normalise_paths, observe +from .sync_guard import run_owner from .types_defs import LanguageQueries, PropertyDict, ReingestReport from .utils.path_utils import derive_project_name, should_skip_path @@ -393,7 +394,8 @@ def _measure_then_restore( cs.KEY_RUN_ID: uuid.uuid4().hex, } store.execute_write( - cq.CYPHER_MARK_PROJECT_INCOMPLETE, {**marker, cs.KEY_WRITING: True} + cq.CYPHER_MARK_RUN_INCOMPLETE, + {**marker, **run_owner(), cs.KEY_WRITING: True}, ) try: return measure(lambda: apply(lambda: guard.capture(updater.reingest_scope))) diff --git a/codebase_rag/sync_guard.py b/codebase_rag/sync_guard.py index 96079dc8e3..b048f41f86 100644 --- a/codebase_rag/sync_guard.py +++ b/codebase_rag/sync_guard.py @@ -24,8 +24,12 @@ from . import cypher_queries as cq from . import logs as ls from .services import QueryProtocol -from .types_defs import PropertyValue -from .utils.process_owner import process_host +from .types_defs import PropertyValue, ResultRow +from .utils.process_owner import ( + pid_has_exited, + process_host, + process_table_identified, +) class SyncRefusedError(RuntimeError): @@ -282,3 +286,98 @@ def partial_graph_refusal(ingestor: QueryProtocol, project_name: str) -> str | N ): return cs.CLI_ERR_EDIT_GRAPH_PARTIAL.format(project=project_name) return None + + +class StrandedRun(NamedTuple): + """A run whose marker a full rebuild retires, and the exited process it names.""" + + run_id: str + owner_host: str + owner_pid: int + + +def run_owner() -> dict[str, PropertyValue]: + """The process an MCP or `cgr check` run's marker names (review of PR #3496).""" + return {cs.KEY_OWNER_HOST: process_host(), cs.KEY_OWNER_PID: os.getpid()} + + +def _stopped_run(row: ResultRow) -> StrandedRun | None: + """The run a marker names if its process provably exited. + + Only a process in this process table can be checked, and only where the + host's name tells this table from every other (`process_table_identified`): + a hostname alone may be another machine's, whose pid is no process here. + A marker from another host, from a live process (this one included), or + from before markers named their process is kept: it may guard a graph its + run is still writing, as `cli._stopped_sync_token` reasons. + """ + run_id = row.get(cs.KEY_RUN_ID) + pid = row.get(cs.KEY_OWNER_PID) + here = process_host() + if ( + not isinstance(run_id, str) + or not process_table_identified() + or row.get(cs.KEY_OWNER_HOST) != here + or not isinstance(pid, int) + or pid == os.getpid() + or not pid_has_exited(pid) + ): + return None + return StrandedRun(run_id, here, pid) + + +def stranded_runs( + ingestor: QueryProtocol, project_name: str, own_run_id: str +) -> list[StrandedRun]: + """The other runs whose markers a full rebuild of the project supersedes. + + Asked before the rebuild marks. A run counts only once its process has + exited, whichever checkout it wrote from: the checkout's sync lock keeps + out only the writers of this checkout under this CGR_HOME, and a run of + the same project elsewhere may still be writing (review of PR #3496). The + rebuild must then be forced: a stopped re-ingest may have deleted an + unchanged dependent's definitions, which an incremental run skips by its + saved hash, as `cgr watch`'s `_marked_run` says. A store that cannot + answer yields none, so the rebuild runs as before and those markers stay + for the next one. + """ + try: + rows = ingestor.fetch_all( + cq.CYPHER_STRANDED_RUN_IDS, + {cs.KEY_PROJECT_NAME: project_name, cs.KEY_RUN_ID: own_run_id}, + ) + except Exception as exc: + logger.warning( + ls.STRANDED_RUNS_UNREADABLE.format(project=project_name, error=exc) + ) + return [] + return [run for row in rows if (run := _stopped_run(row)) is not None] + + +def retire_stranded_runs( + ingestor: QueryProtocol, project_name: str, runs: list[StrandedRun] +) -> None: + """Take off the markers of `runs` once a forced rebuild committed. + + Never fails the rebuild, which is complete by now: a marker that stays + only refuses until the next full sync retires it. + """ + if not runs: + return + try: + for run in runs: + ingestor.execute_write( + cq.CYPHER_RETIRE_STRANDED_RUN, + { + cs.KEY_PROJECT_NAME: project_name, + cs.KEY_RUN_ID: run.run_id, + cs.KEY_OWNER_HOST: run.owner_host, + cs.KEY_OWNER_PID: run.owner_pid, + }, + ) + except Exception as exc: + logger.warning( + ls.STRANDED_RUNS_NOT_RETIRED.format(project=project_name, error=exc) + ) + return + logger.info(ls.STRANDED_RUNS_RETIRED.format(project=project_name, count=len(runs))) diff --git a/codebase_rag/tests/test_check_isolation_marker.py b/codebase_rag/tests/test_check_isolation_marker.py index c7e9bf028e..86de352e16 100644 --- a/codebase_rag/tests/test_check_isolation_marker.py +++ b/codebase_rag/tests/test_check_isolation_marker.py @@ -72,7 +72,7 @@ def test_the_marker_is_cleared_once_graph_and_cache_are_back(tmp_path: Path) -> _run(tmp_path, store) assert store.writes == [ - cq.CYPHER_MARK_PROJECT_INCOMPLETE, + cq.CYPHER_MARK_RUN_INCOMPLETE, cq.CYPHER_CLEAR_PROJECT_INCOMPLETE, ] @@ -90,4 +90,4 @@ def block_the_cache() -> None: with pytest.raises(OSError): _run(tmp_path, store, during=block_the_cache) - assert store.writes == [cq.CYPHER_MARK_PROJECT_INCOMPLETE] + assert store.writes == [cq.CYPHER_MARK_RUN_INCOMPLETE] diff --git a/codebase_rag/tests/test_cli_sync_incomplete_marker.py b/codebase_rag/tests/test_cli_sync_incomplete_marker.py index 66ab42dc44..e375016a7e 100644 --- a/codebase_rag/tests/test_cli_sync_incomplete_marker.py +++ b/codebase_rag/tests/test_cli_sync_incomplete_marker.py @@ -9,6 +9,8 @@ import os import re +import subprocess +import sys from collections.abc import Generator from pathlib import Path from unittest.mock import MagicMock, patch @@ -61,7 +63,7 @@ def write(query: str, params: dict | None = None) -> None: @pytest.fixture def updater(events: list[str]) -> MagicMock: instance = MagicMock() - instance.run.side_effect = lambda: events.append("run") + instance.run.side_effect = lambda force=False: events.append("run") instance.skipped_because_in_sync = False # A bare MagicMock attribute is truthy, which would read every interrupt # as one that landed after the run committed. @@ -132,7 +134,7 @@ def test_a_sync_marks_before_writing_and_clears_after_running( def test_a_failed_sync_keeps_its_marker( tmp_path: Path, sync_env: MagicMock, updater: MagicMock, events: list[str] ) -> None: - def interrupted() -> None: + def interrupted(force: bool = False) -> None: events.append("run") raise KeyboardInterrupt @@ -143,6 +145,148 @@ def interrupted() -> None: assert events == ["mark", "constraints", "run"], events +def _gone_pid() -> int: + """The pid of a process that has exited and been reaped.""" + child = subprocess.Popen([sys.executable, "-c", ""]) + child.wait() + return child.pid + + +@pytest.fixture +def identified_process_table() -> Generator[None, None, None]: + """A host whose name tells its process table apart, as Linux's does. + + On macOS and Windows the name is the hostname alone, and no run is + stranded there (review of PR #3496). + """ + with patch("codebase_rag.sync_guard.process_table_identified", return_value=True): + yield + + +def _strand(ingestor: MagicMock, events: list[str], runs: list[dict]) -> None: + """Markers of other writers' runs that got past their first write (PR #3493). + + The MCP server marks with a run id of its own, and nothing but that run + cleared its marker, so one whose process died left the project refused + for good. + """ + marker_write = ingestor.execute_write.side_effect + + def read(query: str, params: dict | None = None) -> list[dict]: + if query == cq.CYPHER_STRANDED_RUN_IDS: + events.append("stranded") + return runs + return [] + + def write(query: str, params: dict | None = None) -> None: + if query == cq.CYPHER_RETIRE_STRANDED_RUN: + events.append("retire") + marker_write(query, params) + + ingestor.fetch_all.side_effect = read + ingestor.execute_write.side_effect = write + + +def _run(run_id: str, pid: int, host: str | None = None) -> dict: + return { + cs.KEY_RUN_ID: run_id, + cs.KEY_OWNER_HOST: process_host() if host is None else host, + cs.KEY_OWNER_PID: pid, + } + + +@pytest.mark.usefixtures("identified_process_table") +def test_a_sync_over_a_run_whose_process_exited_rebuilds_then_retires_it( + tmp_path: Path, + sync_env: MagicMock, + ingestor: MagicMock, + updater: MagicMock, + events: list[str], +) -> None: + dead = _gone_pid() + _strand(ingestor, events, [_run("mcp-dead", dead)]) + + _sync(tmp_path) + + # Read before this sync marks; retired only once its own marker is off. + assert events == ["stranded", "mark", "constraints", "run", "clear", "retire"] + # Forced: a stopped re-ingest may have deleted an unchanged dependent's + # definitions, which an incremental run skips by its saved hash. + updater.run.assert_called_once_with(force=True) + (read,) = [ + call.args[1] + for call in ingestor.fetch_all.call_args_list + if call.args[0] == cq.CYPHER_STRANDED_RUN_IDS + ] + assert read == {cs.KEY_PROJECT_NAME: "proj", cs.KEY_RUN_ID: cs.CLI_SYNC_RUN_ID} + # Deleted only while it still names that process. + assert _marker_writes(ingestor, cq.CYPHER_RETIRE_STRANDED_RUN) == [ + {cs.KEY_PROJECT_NAME: "proj"} | _run("mcp-dead", dead) + ] + + +@pytest.mark.usefixtures("identified_process_table") +def test_a_sync_with_nothing_stranded_stays_incremental( + tmp_path: Path, + sync_env: MagicMock, + ingestor: MagicMock, + updater: MagicMock, + events: list[str], +) -> None: + _strand(ingestor, events, []) + + _sync(tmp_path) + + updater.run.assert_called_once_with(force=False) + assert "retire" not in events + + +@pytest.mark.usefixtures("identified_process_table") +@pytest.mark.parametrize( + "run", + [ + # Review of PR #3496: a run of the same project from another checkout, + # or under another CGR_HOME, is not held off by this checkout's lock + # and may still be writing. + pytest.param(_run("mcp-live", os.getppid()), id="live-process"), + pytest.param(_run("mcp-mine", os.getpid()), id="this-process"), + pytest.param(_run("mcp-far", 4242, host="elsewhere"), id="other-host"), + pytest.param({cs.KEY_RUN_ID: "mcp-old"}, id="names-no-process"), + ], +) +def test_a_sync_leaves_a_run_that_may_still_be_writing( + tmp_path: Path, + sync_env: MagicMock, + ingestor: MagicMock, + updater: MagicMock, + events: list[str], + run: dict, +) -> None: + _strand(ingestor, events, [run]) + + _sync(tmp_path) + + assert "retire" not in events + updater.run.assert_called_once_with(force=False) + + +@pytest.mark.usefixtures("identified_process_table") +def test_a_sync_that_stops_short_retires_nothing( + tmp_path: Path, + sync_env: MagicMock, + ingestor: MagicMock, + updater: MagicMock, + events: list[str], +) -> None: + _strand(ingestor, events, [_run("mcp-dead", _gone_pid())]) + updater.run.side_effect = RuntimeError("killed mid-sync") + + with pytest.raises(RuntimeError): + _sync(tmp_path) + + assert "retire" not in events + + def test_a_clean_sync_marks_after_the_wipe( tmp_path: Path, sync_env: MagicMock, events: list[str] ) -> None: diff --git a/codebase_rag/tests/test_process_owner.py b/codebase_rag/tests/test_process_owner.py index efaa60ec3d..241da390d5 100644 --- a/codebase_rag/tests/test_process_owner.py +++ b/codebase_rag/tests/test_process_owner.py @@ -10,12 +10,13 @@ import subprocess import sys from collections.abc import Iterator +from pathlib import Path from unittest.mock import patch import pytest from codebase_rag import constants as cs -from codebase_rag.utils.process_owner import pid_has_exited +from codebase_rag.utils.process_owner import pid_has_exited, process_table_identified def _gone_pid() -> int: @@ -78,3 +79,53 @@ def test_windows_never_signals_the_process_it_asks_about(running_pid: int) -> No assert not pid_has_exited(running_pid) assert not pid_has_exited(os.getpid()) assert pid_has_exited(_gone_pid()) + + +# Review of PR #3496: a full sync retires another run's marker once that +# run's pid has exited here, which says something only when the host's name +# cannot also be another machine's or another pid namespace's. + + +@pytest.fixture +def fresh_identity() -> Iterator[None]: + process_table_identified.cache_clear() + yield + process_table_identified.cache_clear() + + +@pytest.mark.usefixtures("fresh_identity") +@pytest.mark.parametrize("missing", ["boot-id", "pid-namespace"]) +def test_a_host_name_without_both_parts_identifies_no_process_table( + missing: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Negative: the hostname alone, or a container hiding its pid namespace. + boot_id = tmp_path / "boot_id" + boot_id.write_text("6b1c2f0e\n", encoding="utf-8") + absent = str(tmp_path / "absent") + monkeypatch.setattr( + cs, "PROCESS_BOOT_ID_PATH", absent if missing == "boot-id" else str(boot_id) + ) + if missing == "pid-namespace": + monkeypatch.setattr(cs, "PROCESS_PID_NAMESPACE_PATH", absent) + + assert not process_table_identified() + + +@pytest.mark.usefixtures("fresh_identity") +@pytest.mark.skipif( + sys.platform == cs.PLATFORM_WINDOWS, + reason="creating a symbolic link takes a privilege on Windows", +) +def test_a_boot_id_and_a_pid_namespace_identify_the_process_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Controlled paths, as Linux lays them out: a container that hides the + # real ones is not identified, and must not fail this test. + boot_id = tmp_path / "boot_id" + boot_id.write_text("6b1c2f0e\n", encoding="utf-8") + namespace = tmp_path / "pid" + namespace.symlink_to("pid:[4026531836]") + monkeypatch.setattr(cs, "PROCESS_BOOT_ID_PATH", str(boot_id)) + monkeypatch.setattr(cs, "PROCESS_PID_NAMESPACE_PATH", str(namespace)) + + assert process_table_identified() diff --git a/codebase_rag/tests/test_start_output_scope.py b/codebase_rag/tests/test_start_output_scope.py index 0657057caf..ba84aba181 100644 --- a/codebase_rag/tests/test_start_output_scope.py +++ b/codebase_rag/tests/test_start_output_scope.py @@ -76,7 +76,7 @@ def build(**kwargs: str) -> MagicMock: # The name the updater writes on the Project node. updater.project_name = kwargs["project_name"].strip() updater.skipped_because_in_sync = in_sync - updater.run.side_effect = lambda: clock.advance(SYNC_SECONDS) + updater.run.side_effect = lambda force=False: clock.advance(SYNC_SECONDS) return updater return build diff --git a/codebase_rag/tests/test_status_syncs_from_graph.py b/codebase_rag/tests/test_status_syncs_from_graph.py index 6422512385..2222e2f6b6 100644 --- a/codebase_rag/tests/test_status_syncs_from_graph.py +++ b/codebase_rag/tests/test_status_syncs_from_graph.py @@ -141,6 +141,25 @@ def fetch_all( if claimed: self._mark(params) return claimed + if params and query == cq.CYPHER_STRANDED_RUN_IDS: + rows: list[ResultRow] = [] + for run in sorted(self.markers): + project, run_id = run + if ( + project != params[cs.KEY_PROJECT_NAME] + or run_id == params[cs.KEY_RUN_ID] + or not self.writing.get(run, True) + ): + continue + owner = self.owners.get(run, {}) + rows.append( + { + cs.KEY_RUN_ID: run_id, + cs.KEY_OWNER_HOST: owner.get(cs.KEY_OWNER_HOST), + cs.KEY_OWNER_PID: owner.get(cs.KEY_OWNER_PID), + } + ) + return rows return super().fetch_all(query, params) def _release_project_claim(self, params: PropertyDict) -> None: @@ -162,8 +181,10 @@ def _mark(self, params: PropertyDict) -> None: self.writing[run] = self.writing.get(run, False) or bool( params.get(cs.KEY_WRITING) ) - if cs.KEY_OWNER_TOKEN in params: - self.owners[run] = {key: params[key] for key in _OWNER_KEYS} + if cs.KEY_OWNER_PID in params: + self.owners[run] = { + key: params[key] for key in _OWNER_KEYS if key in params + } def _clear(self, run: tuple[str, str]) -> None: self.markers.discard(run) @@ -174,6 +195,7 @@ def execute_write(self, query: str, params: PropertyDict | None = None) -> None: if params and query in ( cq.CYPHER_MARK_PROJECT_INCOMPLETE, cq.CYPHER_MARK_CLI_SYNC_INCOMPLETE, + cq.CYPHER_MARK_RUN_INCOMPLETE, ): self._mark(params) return @@ -186,6 +208,15 @@ def execute_write(self, query: str, params: PropertyDict | None = None) -> None: if token == params[cs.KEY_OWNER_TOKEN]: self._clear(run) return + if params and query == cq.CYPHER_RETIRE_STRANDED_RUN: + run = self._run(params) + owner = self.owners.get(run, {}) + if all( + owner.get(key) == params[key] + for key in (cs.KEY_OWNER_HOST, cs.KEY_OWNER_PID) + ): + self._clear(run) + return if params and query == cq.CYPHER_RECOVER_PROJECT_INCOMPLETE: project = str(params[cs.KEY_PROJECT_NAME]) for run in [ @@ -355,7 +386,7 @@ def _fail_sync( ) -> None: """A sync that stops short, after doing `mid_sync` while it still runs.""" - def run(_updater: GraphUpdater) -> None: + def run(_updater: GraphUpdater, force: bool = False) -> None: if mid_sync is not None: mid_sync() raise RuntimeError("killed mid-sync") @@ -398,6 +429,113 @@ def _gone_pid() -> int: return child.pid +def _mcp_marker( + graph: _Memgraph, project: str, run_id: str, *, writing: bool, pid: int | None +) -> None: + """The marker an MCP server's run put down under its own run id. + + It names the run's process, as the server does since review of PR #3496; + without a `pid` it is one an older release wrote, naming none. + """ + params: PropertyDict = { + cs.KEY_PROJECT_NAME: project, + cs.KEY_RUN_ID: run_id, + cs.KEY_WRITING: writing, + } + if pid is None: + graph.execute_write(cq.CYPHER_MARK_PROJECT_INCOMPLETE, params) + return + params |= {cs.KEY_OWNER_HOST: process_host(), cs.KEY_OWNER_PID: pid} + graph.execute_write(cq.CYPHER_MARK_RUN_INCOMPLETE, params) + + +class TestAFullSyncRetiresRunsItSupersedes: + # Review of PR #3493: an MCP run that died after its first write left its + # marker under a run id nothing else clears, so `cgr status` and every + # later edit kept refusing however many times the project was re-synced. + + @pytest.fixture(autouse=True) + def _identified_process_table(self) -> Generator[None, None, None]: + # As on Linux. On macOS and Windows the host's name is the hostname + # alone, so nothing is retired there (review of PR #3496). + with patch( + "codebase_rag.sync_guard.process_table_identified", return_value=True + ): + yield + + def test_a_crashed_mcp_runs_marker_goes_with_the_next_full_sync( + self, graphs: dict[int, _Memgraph], tmp_path: Path + ) -> None: + _sync(tmp_path, "billing") + graph = graphs[THIS_GRAPH] + _mcp_marker(graph, "billing", "mcp-crashed", writing=True, pid=_gone_pid()) + marker = f"({cs.CLI_STATUS_SYNC_INCOMPLETE})" + assert marker in _listed(_cgr("status"))["billing"] + + _sync(tmp_path, "billing") + + assert graph.markers == set() + assert marker not in _listed(_cgr("status"))["billing"] + + def test_a_failed_sync_leaves_the_crashed_runs_marker( + self, graphs: dict[int, _Memgraph], tmp_path: Path + ) -> None: + _sync(tmp_path, "billing") + graph = graphs[THIS_GRAPH] + _mcp_marker(graph, "billing", "mcp-crashed", writing=True, pid=_gone_pid()) + + _fail_sync(tmp_path, "billing") + + assert ("billing", "mcp-crashed") in graph.markers + + def test_a_run_that_never_wrote_is_left_to_its_own_recovery( + self, graphs: dict[int, _Memgraph], tmp_path: Path + ) -> None: + # It left the graph as it found it, so nothing needs rebuilding; the + # MCP server clears such a marker when it next reads the project. + _sync(tmp_path, "billing") + graph = graphs[THIS_GRAPH] + _mcp_marker(graph, "billing", "mcp-idle", writing=False, pid=_gone_pid()) + + _sync(tmp_path, "billing") + + assert graph.markers == {("billing", "mcp-idle")} + + @pytest.mark.parametrize( + "pid", + [ + # Review of PR #3496: a run of the project from another checkout, + # which this checkout's lock does not hold off, may still be + # writing; so may one that names no process. + pytest.param(os.getppid(), id="process-still-running"), + pytest.param(None, id="names-no-process"), + ], + ) + def test_a_run_that_may_still_be_writing_keeps_its_marker( + self, graphs: dict[int, _Memgraph], tmp_path: Path, pid: int | None + ) -> None: + _sync(tmp_path, "billing") + graph = graphs[THIS_GRAPH] + _mcp_marker(graph, "billing", "mcp-elsewhere", writing=True, pid=pid) + + _sync(tmp_path, "billing") + + assert graph.markers == {("billing", "mcp-elsewhere")} + + def test_only_the_synced_projects_runs_are_retired( + self, graphs: dict[int, _Memgraph], tmp_path: Path + ) -> None: + _sync(tmp_path, "billing") + _sync(tmp_path, "ledger") + graph = graphs[THIS_GRAPH] + _mcp_marker(graph, "billing", "mcp-crashed", writing=True, pid=_gone_pid()) + _mcp_marker(graph, "ledger", "mcp-crashed", writing=True, pid=_gone_pid()) + + _sync(tmp_path, "billing") + + assert graph.markers == {("ledger", "mcp-crashed")} + + class TestDeletingAProjectAfterAFailedSync: # Review of PR #2532: the failed sync's marker sits on its own node, out # of the delete's reach, so the deleted project stayed listed as diff --git a/codebase_rag/tests/test_stranded_run_markers.py b/codebase_rag/tests/test_stranded_run_markers.py new file mode 100644 index 0000000000..babd7260fd --- /dev/null +++ b/codebase_rag/tests/test_stranded_run_markers.py @@ -0,0 +1,301 @@ +# A full rebuild retires the incomplete-run markers of runs it supersedes +# (review of PR #3493). An MCP server marks with a run id of its own and +# only that run cleared it, so one whose process died after its first write +# left the project refused for good: by its reads, by `cgr rename` and +# `cgr change-signature`, and in `cgr status`. The CLI sync's side is in +# test_cli_sync_incomplete_marker.py and test_status_syncs_from_graph.py. + +from __future__ import annotations + +import os +import subprocess +import sys +from collections.abc import Iterator +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from codebase_rag import constants as cs +from codebase_rag import cypher_queries as cq +from codebase_rag.mcp.tools import MCPToolsRegistry +from codebase_rag.sync_guard import StrandedRun, retire_stranded_runs, stranded_runs +from codebase_rag.types_defs import PropertyDict, ResultRow +from codebase_rag.utils.path_utils import derive_project_name +from codebase_rag.utils.process_owner import process_host + +DEAD_RUN = "mcp-dead" + + +@pytest.fixture(autouse=True) +def _identified_process_table() -> Iterator[None]: + """A host whose name tells its process table apart, as Linux's does. + + On macOS and Windows the name is the hostname alone, so nothing is + stranded there; these tests are about what follows once a run is. + """ + with patch("codebase_rag.sync_guard.process_table_identified", return_value=True): + yield + + +def _gone_pid() -> int: + """The pid of a process that has exited and been reaped.""" + child = subprocess.Popen([sys.executable, "-c", ""]) + child.wait() + return child.pid + + +def _marker(run_id: str, pid: int | None, host: str | None = None) -> ResultRow: + """A marker row as `CYPHER_STRANDED_RUN_IDS` returns it.""" + return { + cs.KEY_RUN_ID: run_id, + cs.KEY_OWNER_HOST: process_host() if host is None else host, + cs.KEY_OWNER_PID: pid, + } + + +class Store: + """The ingestor double: the other runs' markers, and every write recorded.""" + + def __init__(self, stranded: list[ResultRow]) -> None: + self.stranded = stranded + self.reads: list[PropertyDict] = [] + self.writes: list[tuple[str, PropertyDict]] = [] + self.ingestor = MagicMock() + self.ingestor.fetch_all.side_effect = self.fetch_all + self.ingestor.execute_write.side_effect = self.execute_write + + def fetch_all( + self, query: str, params: PropertyDict | None = None + ) -> list[ResultRow]: + if query == cq.CYPHER_STRANDED_RUN_IDS: + self.reads.append(dict(params or {})) + return self.stranded + return [] + + def execute_write(self, query: str, params: PropertyDict | None = None) -> None: + self.writes.append((query, dict(params or {}))) + + def queries(self) -> list[str]: + return [query for query, _params in self.writes] + + +@pytest.fixture +def project_root(tmp_path: Path) -> Path: + (tmp_path / "app.py").write_text("def main(): pass\n", encoding="utf-8") + return tmp_path + + +def _registry(root: Path, store: Store) -> MCPToolsRegistry: + registry = MCPToolsRegistry( + project_root=str(root), ingestor=store.ingestor, cypher_gen=MagicMock() + ) + store.ingestor.list_projects.return_value = [derive_project_name(root)] + return registry + + +# --- the MCP server's full update and re-index ---------------------------------- + + +async def test_an_update_over_a_stranded_run_rebuilds_then_retires_it( + project_root: Path, +) -> None: + dead = _gone_pid() + store = Store([_marker(DEAD_RUN, dead)]) + registry = _registry(project_root, store) + + with patch("codebase_rag.mcp.tools.GraphUpdater") as updater_cls: + result = await registry.update_repository() + + assert "Error" not in result, result + # Forced: an incremental run skips the unchanged dependents a stopped + # re-ingest may have deleted. + updater_cls.return_value.run.assert_called_once_with(force=True) + # Its own run id is not stranded: that marker is this update's to clear. + assert store.reads == [ + { + cs.KEY_PROJECT_NAME: derive_project_name(project_root), + cs.KEY_RUN_ID: registry._run_id, + } + ] + queries = store.queries() + assert queries.index(cq.CYPHER_RETIRE_STRANDED_RUN) > queries.index( + cq.CYPHER_CLEAR_PROJECT_INCOMPLETE + ) + # Deleted only while the marker still names the process that exited. + assert store.writes[-1] == ( + cq.CYPHER_RETIRE_STRANDED_RUN, + {cs.KEY_PROJECT_NAME: derive_project_name(project_root)} + | _marker(DEAD_RUN, dead), + ) + + +async def test_the_servers_marker_names_its_process(project_root: Path) -> None: + # Review of PR #3496: what lets a later rebuild tell a run whose process + # exited from one still writing. + store = Store([]) + registry = _registry(project_root, store) + + with patch("codebase_rag.mcp.tools.GraphUpdater"): + await registry.update_repository() + + marks = [ + params + for query, params in store.writes + if query == cq.CYPHER_MARK_RUN_INCOMPLETE + ] + assert marks + for params in marks: + assert params[cs.KEY_RUN_ID] == registry._run_id + assert params[cs.KEY_OWNER_HOST] == process_host() + assert params[cs.KEY_OWNER_PID] == os.getpid() + + +async def test_an_update_with_nothing_stranded_stays_incremental( + project_root: Path, +) -> None: + store = Store([]) + registry = _registry(project_root, store) + + with patch("codebase_rag.mcp.tools.GraphUpdater") as updater_cls: + await registry.update_repository() + + updater_cls.return_value.run.assert_called_once_with(force=False) + assert cq.CYPHER_RETIRE_STRANDED_RUN not in store.queries() + + +async def test_an_update_that_fails_retires_nothing(project_root: Path) -> None: + store = Store([_marker(DEAD_RUN, _gone_pid())]) + registry = _registry(project_root, store) + + with patch("codebase_rag.mcp.tools.GraphUpdater") as updater_cls: + updater_cls.return_value.run.side_effect = RuntimeError("update died") + result = await registry.update_repository() + + assert "update died" in result + assert cq.CYPHER_RETIRE_STRANDED_RUN not in store.queries() + + +async def test_an_update_leaves_a_run_whose_process_still_runs( + project_root: Path, +) -> None: + # Review of PR #3496: the checkout's lock holds off only this checkout's + # writers under this CGR_HOME. A run of the project from anywhere else may + # still be writing, and would lose the marker guarding what it leaves. + store = Store([_marker("mcp-elsewhere", os.getppid())]) + registry = _registry(project_root, store) + + with patch("codebase_rag.mcp.tools.GraphUpdater") as updater_cls: + await registry.update_repository() + + updater_cls.return_value.run.assert_called_once_with(force=False) + assert cq.CYPHER_RETIRE_STRANDED_RUN not in store.queries() + + +async def test_a_reindex_retires_the_runs_it_supersedes(project_root: Path) -> None: + store = Store([_marker(DEAD_RUN, _gone_pid())]) + registry = _registry(project_root, store) + + with ( + patch("codebase_rag.mcp.tools.GraphUpdater"), + patch("codebase_rag.mcp.tools.delete_project_embeddings"), + ): + result = await registry.index_repository() + + assert "Error" not in result, result + queries = store.queries() + assert queries.index(cq.CYPHER_RETIRE_STRANDED_RUN) > queries.index( + cq.CYPHER_CLEAR_PROJECT_INCOMPLETE + ) + + +async def test_a_reindex_that_fails_retires_nothing(project_root: Path) -> None: + store = Store([_marker(DEAD_RUN, _gone_pid())]) + registry = _registry(project_root, store) + + with ( + patch("codebase_rag.mcp.tools.GraphUpdater") as updater_cls, + patch("codebase_rag.mcp.tools.delete_project_embeddings"), + ): + updater_cls.return_value.run.side_effect = RuntimeError("index died") + await registry.index_repository() + + assert cq.CYPHER_RETIRE_STRANDED_RUN not in store.queries() + + +# --- the helpers ---------------------------------------------------------------- + + +def test_stranded_runs_are_the_other_runs_whose_process_exited() -> None: + dead = _gone_pid() + store = Store( + [ + _marker("dead", dead), + _marker("live", os.getppid()), + _marker("mine", os.getpid()), + _marker("far", dead, host="another-machine"), + {cs.KEY_RUN_ID: "old"}, + ] + ) + + assert stranded_runs(store.ingestor, "proj", "own") == [ + StrandedRun("dead", process_host(), dead) + ] + assert store.reads == [{cs.KEY_PROJECT_NAME: "proj", cs.KEY_RUN_ID: "own"}] + + +def test_where_the_process_table_cannot_be_told_apart_nothing_is_stranded() -> None: + # Review of PR #3496: under a hostname alone, a run on another machine of + # the same name reads as local, and its pid, absent here, as exited. + store = Store([_marker(DEAD_RUN, _gone_pid())]) + + with patch("codebase_rag.sync_guard.process_table_identified", return_value=False): + assert stranded_runs(store.ingestor, "proj", "own") == [] + + +def test_a_store_that_cannot_answer_strands_nothing() -> None: + # The rebuild then runs as it did before, and the markers stay for the + # next one rather than being cleared on a guess. + ingestor = MagicMock() + ingestor.fetch_all.side_effect = ConnectionError("graph unreachable") + + assert stranded_runs(ingestor, "proj", "own") == [] + + +def test_retiring_no_runs_writes_nothing() -> None: + ingestor = MagicMock() + + retire_stranded_runs(ingestor, "proj", []) + + ingestor.execute_write.assert_not_called() + + +def test_a_retire_that_fails_does_not_fail_the_rebuild() -> None: + ingestor = MagicMock() + ingestor.execute_write.side_effect = ConnectionError("graph unreachable") + + retire_stranded_runs(ingestor, "proj", [StrandedRun(DEAD_RUN, "host", 4242)]) + + ingestor.execute_write.assert_called_once() + + +def test_each_run_is_retired_against_the_process_it_named() -> None: + # The CLI sync's run id is shared by every CLI writer of the project, so + # one that started since re-marked it under its own process and keeps it. + store = Store([]) + runs = [StrandedRun("run-a", "host", 4242), StrandedRun("run-b", "host", 4243)] + + retire_stranded_runs(store.ingestor, "proj", runs) + + assert store.writes == [ + ( + cq.CYPHER_RETIRE_STRANDED_RUN, + { + cs.KEY_PROJECT_NAME: "proj", + cs.KEY_RUN_ID: run.run_id, + cs.KEY_OWNER_HOST: run.owner_host, + cs.KEY_OWNER_PID: run.owner_pid, + }, + ) + for run in runs + ] diff --git a/codebase_rag/utils/process_owner.py b/codebase_rag/utils/process_owner.py index d5ce2b6aec..ab5f0ee884 100644 --- a/codebase_rag/utils/process_owner.py +++ b/codebase_rag/utils/process_owner.py @@ -96,6 +96,20 @@ def process_host() -> str: ) +@cache +def process_table_identified() -> bool: + """Whether `process_host` names this process table and no other. + + Only where both the boot id and the pid namespace can be read, as on + Linux. Elsewhere, or in a container that hides them, the name is the + hostname alone, which another machine can carry too, so a pid recorded + under it may be a process this table has never had (review of PR #3496). + """ + return bool(_read_text(cs.PROCESS_BOOT_ID_PATH)) and bool( + _read_link(cs.PROCESS_PID_NAMESPACE_PATH) + ) + + def pid_may_be_running(pid: int) -> bool: """False only when no process with this pid exists in this process table. diff --git a/docs/guide/realtime-updates.md b/docs/guide/realtime-updates.md index bffa876337..dc8d7283da 100644 --- a/docs/guide/realtime-updates.md +++ b/docs/guide/realtime-updates.md @@ -104,7 +104,19 @@ stopped part-way through a write therefore shows in `cgr status` as an unfinished sync, and the MCP server does not trust the partial graph, until the next sync of the project finishes. A change that arrives while another sync of the project has left its marker down re-indexes the whole repository -instead of applying that change alone. +instead of applying that change alone. A full sync (`cgr start +--update-graph`, or the MCP server's `update_repository` and +`index_repository`) also takes off the marker an MCP server's run left when +it stopped part-way through a write, which nothing else clears: it rebuilds +the whole repository first, then retires that marker with its own. It does so +only once the process that wrote the marker has exited on this machine. A run +still going, even one from another checkout under the same project name, keeps +its marker, and so does a run on another machine or in another container. A +marker an older release wrote names no process, so it stays too. The check +needs a machine whose processes cannot be mistaken for another machine's: +Linux, where the boot id and the pid namespace tell them apart. On macOS and +Windows, which name the machine by its hostname alone, such a marker stays, +as it did before. ### One sync of a checkout at a time