fix(serve): stop a graph reload from stalling every MCP request

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:

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