From abeb4b87c8cb15f2ce3703dba4c12ce5b80bc3c3 Mon Sep 17 00:00:00 2001 From: ayushcodes10 Date: Thu, 12 Nov 2026 19:29:36 +0530 Subject: [PATCH 1/2] Stop a graph reload from stalling every MCP request _GraphContextCache.load() ran the whole rebuild, parsing graph.json and building its trigram index, inside the cache's single lock. That one lock covers the pinned default graph and every project graph, so while a changed graph.json was reloaded every request on the server waited, including requests for a different graph and requests that could have been answered from the graph already sitting in memory. On a large graph the rebuild alone can take upward of fifteen seconds. The rebuild now runs outside the lock, single flighted per resolved path so concurrent callers never start a second, redundant rebuild of the same path. A caller that already holds a stale entry for that path gets it back immediately while someone else's rebuild is in flight, rather than queuing behind it. A caller with no entry yet, the path's first ever load, has nothing stale to serve and waits for the one in flight build instead of racing a second one. Since the lock is held only for the stat and the small dict bookkeeping around it, a slow rebuild of one path no longer blocks a request for an unrelated one. --- graphify/serve.py | 50 +++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 46 insertions(+), 4 deletions(-) diff --git a/graphify/serve.py b/graphify/serve.py index c39872091a..ddfe94a880 100644 --- a/graphify/serve.py +++ b/graphify/serve.py @@ -146,6 +146,10 @@ def __init__(self, max_contexts: int): self._entries: OrderedDict[str, dict] = OrderedDict() self._pinned: dict[str, dict] = {} self._lock = threading.Lock() + # One in-flight build per (resolved_path, pinned); keyed separately + # from `_entries`/`_pinned` so a rebuild in progress for one path + # never blocks a lookup for another (#5177). + self._loading: dict[tuple[str, bool], threading.Event] = {} def _load_entry(self, resolved_path: str, key: tuple[int, int]) -> dict: """Build one entry for an already-resolved path and known file key. @@ -177,27 +181,65 @@ def load(self, resolved_path: str, *, pinned: bool = False) -> tuple[nx.Graph, d ``pinned=True`` is reserved for the server's configured default graph; it remains warm without consuming a project-cache slot. + + The actual rebuild (``_load_entry``, which parses graph.json and builds + its trigram index) runs outside ``_lock`` so a slow reload of one path + never blocks a request for another (#5177). While a rebuild for THIS + path is already in flight, a caller that still has an old entry for it + gets that stale entry back immediately rather than queuing behind the + rebuild; a caller with no entry yet (the path's first ever load) has + nothing stale to serve and waits for the one in-flight build instead + of starting a redundant second one. """ + entries = self._pinned if pinned else self._entries + loading_key = (resolved_path, pinned) with self._lock: try: stat_result = Path(resolved_path).stat() except FileNotFoundError: raise FileNotFoundError(f"graph.json not found: {resolved_path}") from None key = (stat_result.st_mtime_ns, stat_result.st_size) - entries = self._pinned if pinned else self._entries entry = entries.get(resolved_path) if entry is not None and entry["key"] == key: if not pinned: self._entries.move_to_end(resolved_path) return entry["G"], entry["communities"] - entry = self._load_entry(resolved_path, key) - entries[resolved_path] = entry + in_flight = self._loading.get(loading_key) + if entry is not None and in_flight is not None: + if not pinned: + self._entries.move_to_end(resolved_path) + return entry["G"], entry["communities"] + + if in_flight is not None: + wait_event, build_here = in_flight, False + else: + wait_event = threading.Event() + self._loading[loading_key] = wait_event + build_here = True + + if not build_here: + wait_event.wait() + # The build this thread waited on may belong to a key that was + # already superseded by a newer write; recurse to either pick up + # its result or (if it failed, or was already stale on arrival) + # take a fresh run at loading/building it. + return self.load(resolved_path, pinned=pinned) + + try: + new_entry = self._load_entry(resolved_path, key) + finally: + with self._lock: + self._loading.pop(loading_key, None) + wait_event.set() + + with self._lock: + entries[resolved_path] = new_entry if not pinned: self._entries.move_to_end(resolved_path) while len(self._entries) > self._max_contexts: self._entries.popitem(last=False) - return entry["G"], entry["communities"] + return new_entry["G"], new_entry["communities"] def _node_arg(arguments: dict) -> str: From a19d426310fcc92014858a2e163a90bfacec5e64 Mon Sep 17 00:00:00 2001 From: ayushcodes10 Date: Thu, 12 Nov 2026 19:29:53 +0530 Subject: [PATCH 2/2] Add concurrency tests for the graph reload lock Covers a caller getting its stale entry back immediately while another thread's reload of the same path is in flight, a reload of one path never blocking a request for a different path, concurrent first time loads of a brand new path still single flighting to one build rather than each paying for their own, and a later call seeing the new graph once an in flight rebuild finishes. --- tests/test_graph_context_cache_reload.py | 168 +++++++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 tests/test_graph_context_cache_reload.py diff --git a/tests/test_graph_context_cache_reload.py b/tests/test_graph_context_cache_reload.py new file mode 100644 index 0000000000..6b7aa97bac --- /dev/null +++ b/tests/test_graph_context_cache_reload.py @@ -0,0 +1,168 @@ +"""`_GraphContextCache.load` under concurrent access (#5177). + +The whole rebuild used to run inside the cache's single lock, so a slow +reload of one graph stalled every request on the server, including requests +for a different graph and requests that could have been answered from the +graph already in memory. The rebuild now runs outside the lock, is +single-flighted per path, and a caller that already has a (stale) entry gets +it back immediately instead of queuing behind someone else's reload. +""" +from __future__ import annotations + +import json +import os +import threading +import time +from pathlib import Path + +from graphify.serve import _GraphContextCache + +GRAPH = { + "directed": True, + "nodes": [{"id": "a", "label": "A", "community": 0}, {"id": "b", "label": "B", "community": 0}], + "edges": [{"source": "a", "target": "b", "relation": "calls", "confidence": "EXTRACTED"}], +} + + +def _write_graph(path: Path) -> None: + path.write_text(json.dumps(GRAPH)) + + +def _gate(cache: _GraphContextCache, delay: float, slow_for=None): + """Make `cache._load_entry` sleep `delay`s, counting every call. + + `slow_for` restricts the delay to paths whose resolved string contains it; + every other path still uses the real (fast) loader. + """ + original = cache._load_entry + calls: list[tuple[str, tuple[int, int]]] = [] + + def gated(resolved_path, key): + calls.append((resolved_path, key)) + if slow_for is None or slow_for in resolved_path: + time.sleep(delay) + return original(resolved_path, key) + + cache._load_entry = gated + return calls + + +def test_stale_entry_served_while_a_reload_is_in_flight(tmp_path): + graph_path = tmp_path / "graph.json" + _write_graph(graph_path) + path = str(graph_path) + + cache = _GraphContextCache(8) + cache.load(path) # warm, unpatched + + calls = _gate(cache, delay=0.3) + os.utime(path, (time.time() + 10, time.time() + 10)) # make the warm entry stale + + results: dict[int, float] = {} + + def caller(i): + t0 = time.time() + cache.load(path) + results[i] = time.time() - t0 + + threads = [threading.Thread(target=caller, args=(i,)) for i in range(3)] + for t in threads: + t.start() + time.sleep(0.05) + for t in threads: + t.join() + + # Exactly one caller pays the rebuild cost; the others, which still had + # the old entry, return almost immediately with it rather than blocking. + slow = [dt for dt in results.values() if dt > 0.2] + fast = [dt for dt in results.values() if dt <= 0.2] + assert len(slow) == 1, results + assert len(fast) == 2, results + assert len(calls) == 1, "a second, redundant reload was started while one was already in flight" + + +def test_concurrent_first_load_of_new_path_builds_once(tmp_path): + """Control: with no stale entry to fall back to, concurrent first-time + loads of the same brand-new path must still single-flight to one build, + not each start (and redundantly pay for) their own.""" + graph_path = tmp_path / "graph.json" + _write_graph(graph_path) + path = str(graph_path) + + cache = _GraphContextCache(8) + calls = _gate(cache, delay=0.3) + + results: dict[int, float] = {} + + def caller(i): + t0 = time.time() + cache.load(path) + results[i] = time.time() - t0 + + threads = [threading.Thread(target=caller, args=(i,)) for i in range(4)] + for t in threads: + t.start() + time.sleep(0.02) + for t in threads: + t.join() + + assert len(calls) == 1, "every concurrent first-time caller triggered its own rebuild" + assert all(dt > 0.2 for dt in results.values()), results + + +def test_reload_of_one_path_does_not_block_another_path(tmp_path): + p1 = tmp_path / "p1" / "graph.json" + p2 = tmp_path / "p2" / "graph.json" + p1.parent.mkdir() + p2.parent.mkdir() + _write_graph(p1) + _write_graph(p2) + + cache = _GraphContextCache(8) + _gate(cache, delay=0.5, slow_for="p1") + + results: dict[str, float] = {} + + def load_p1(): + t0 = time.time() + cache.load(str(p1)) + results["p1"] = time.time() - t0 + + def load_p2(): + time.sleep(0.05) + t0 = time.time() + cache.load(str(p2)) + results["p2"] = time.time() - t0 + + t1 = threading.Thread(target=load_p1) + t2 = threading.Thread(target=load_p2) + t1.start() + t2.start() + t1.join() + t2.join() + + assert results["p1"] > 0.4, results + assert results["p2"] < 0.2, "an unrelated path's load was blocked by another path's slow rebuild" + + +def test_reload_eventually_serves_the_new_graph(tmp_path): + """Sanity: once an in-flight rebuild finishes, a later call sees the new + data, not the stale one forever.""" + graph_path = tmp_path / "graph.json" + _write_graph(graph_path) + path = str(graph_path) + + cache = _GraphContextCache(8) + G, _ = cache.load(path) + assert G.number_of_nodes() == 2 + + bigger = { + "directed": True, + "nodes": [{"id": n, "label": n, "community": 0} for n in "abc"], + "edges": [], + } + graph_path.write_text(json.dumps(bigger)) + os.utime(path, (time.time() + 10, time.time() + 10)) + + G2, _ = cache.load(path) + assert G2.number_of_nodes() == 3