From 687a180721b24f88edd72e1260ecf38073c9b849 Mon Sep 17 00:00:00 2001 From: pravit-amp <43916793+pravit-amp@users.noreply.github.com> Date: Wed, 12 Aug 2026 08:02:45 -0700 Subject: [PATCH] fix(context): take the lock in ContextGraph.to_dict() (#929) * fix(context): take the lock in ContextGraph.to_dict() to_dict() iterated self.nodes.values() and self.edges without holding self._lock, so a concurrent writer raised "RuntimeError: dictionary changed size during iteration". It was the only reader on the class that did not take the lock -- stats(), density(), find_nodes(), find_edges(), get_neighbors(), get_nodes_by_label(), state_at() and save_to_file() all hold it. Commit 1d1ae398 introduced the RLock and added 26 "with self._lock:" blocks; to_dict already existed and was not among them. save_to_file is safe only incidentally -- it holds the lock and builds its payload inline rather than delegating to to_dict, so it never reaches the unguarded loops. Beyond the RuntimeError, the unguarded body could also return a torn snapshot: the statistics block reads len(self.nodes)/len(self.edges) after building the node and edge lists, so a write landing in between yields counts that contradict the payload they describe. self._lock is an RLock, so this composes with the callers that already hold it (build_from_conversation and build_from_documents both return self.to_dict() from inside a locked block). Neither external caller -- agent_context's _capture_checkpoint_state nor triplet_store's knowledge-graph conversion -- defines a lock of its own, so there is no ordering inversion. Add tests/context/test_context_graph_thread_safety.py: a deterministic check that to_dict() blocks while another thread holds _lock (no race window needed), a reentrancy check, and three checks under concurrent writes covering the RuntimeError, statistics/payload agreement, and duplicate node ids. Four of the five fail against the unfixed method. Closes #923 * test(context): make to_dict lock tests deterministic and hang-proof Wait for the worker thread to actually start before asserting to_dict() blocks on _lock, and run the reentrancy check in a joined worker so a non-reentrant lock fails the test instead of hanging CI. * test(context): assert worker threads actually stopped after timed joins A join(timeout=...) on a daemon thread returns even if the thread is still running, so a deadlock would leak a live thread into subsequent tests instead of failing. Assert not is_alive() after each timed join. --------- Co-authored-by: Pravit Ampapathini Co-authored-by: Zohaib Hassnain <109234410+ZohaibHassan16@users.noreply.github.com> --- semantica/context/context_graph.py | 79 ++++---- .../test_context_graph_thread_safety.py | 168 ++++++++++++++++++ 2 files changed, 208 insertions(+), 39 deletions(-) create mode 100644 tests/context/test_context_graph_thread_safety.py diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index 06419759..944e1f6a 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -1769,47 +1769,48 @@ class ContextGraph: def to_dict(self) -> Dict[str, Any]: """Export graph to dictionary format.""" - nodes_out = [] - for n in self.nodes.values(): - entry: Dict[str, Any] = { - "id": n.node_id, - "type": n.node_type, - "content": n.content, - "properties": n.properties, - "metadata": n.metadata, - } - if n.valid_from is not None: - entry["valid_from"] = n.valid_from - if n.valid_until is not None: - entry["valid_until"] = n.valid_until - nodes_out.append(entry) + with self._lock: + nodes_out = [] + for n in self.nodes.values(): + entry: Dict[str, Any] = { + "id": n.node_id, + "type": n.node_type, + "content": n.content, + "properties": n.properties, + "metadata": n.metadata, + } + if n.valid_from is not None: + entry["valid_from"] = n.valid_from + if n.valid_until is not None: + entry["valid_until"] = n.valid_until + nodes_out.append(entry) - edges_out = [] - for e in self.edges: - entry = { - "id": e.edge_id, - "familyId": e.family_id or e.edge_id, - "source": e.source_id, - "target": e.target_id, - "type": e.edge_type, - "weight": e.weight, - } - if e.metadata: - entry["metadata"] = e.metadata - if e.valid_from is not None: - entry["valid_from"] = e.valid_from - if e.valid_until is not None: - entry["valid_until"] = e.valid_until - edges_out.append(entry) + edges_out = [] + for e in self.edges: + entry = { + "id": e.edge_id, + "familyId": e.family_id or e.edge_id, + "source": e.source_id, + "target": e.target_id, + "type": e.edge_type, + "weight": e.weight, + } + if e.metadata: + entry["metadata"] = e.metadata + if e.valid_from is not None: + entry["valid_from"] = e.valid_from + if e.valid_until is not None: + entry["valid_until"] = e.valid_until + edges_out.append(entry) - return { - "nodes": nodes_out, - "edges": edges_out, - "statistics": { - "node_count": len(self.nodes), - "edge_count": len(self.edges), - }, - } + return { + "nodes": nodes_out, + "edges": edges_out, + "statistics": { + "node_count": len(self.nodes), + "edge_count": len(self.edges), + }, + } def from_dict(self, graph_dict: Dict[str, Any]) -> None: """Load graph from dictionary format.""" diff --git a/tests/context/test_context_graph_thread_safety.py b/tests/context/test_context_graph_thread_safety.py new file mode 100644 index 00000000..0d521d7e --- /dev/null +++ b/tests/context/test_context_graph_thread_safety.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +"""Regression tests for ``ContextGraph.to_dict()`` thread safety. + +``ContextGraph`` guards its state with ``self._lock`` (an ``RLock``), and every +reader on the class takes it -- ``stats``, ``density``, ``find_nodes``, +``find_edges``, ``get_neighbors``, ``get_nodes_by_label``, ``state_at`` and +``save_to_file`` all do. ``to_dict`` was the one exception: it iterated +``self.nodes.values()`` and ``self.edges`` unguarded, so a concurrent writer +raised ``RuntimeError: dictionary changed size during iteration``. + +``save_to_file`` was safe only incidentally -- it holds the lock and builds its +payload inline rather than delegating to ``to_dict``. +""" + +import threading +import time + +from semantica.context.context_graph import ContextGraph + + +def _seeded_graph(node_count: int = 200) -> ContextGraph: + graph = ContextGraph(advanced_analytics=False) + for i in range(node_count): + graph.add_node(f"seed{i}", "seed") + return graph + + +class TestToDictHoldsTheLock: + """``to_dict`` must take ``_lock``, like every sibling reader.""" + + def test_to_dict_waits_for_the_lock(self): + """Deterministic proof the lock is held -- no race window needed. + + With the lock held elsewhere, ``to_dict`` must block. Without the fix it + returns immediately, since it never asks for the lock at all. + """ + graph = _seeded_graph(10) + started = threading.Event() + finished = threading.Event() + + def snapshot(): + started.set() + graph.to_dict() + finished.set() + + with graph._lock: + worker = threading.Thread(target=snapshot, daemon=True) + worker.start() + assert started.wait(timeout=5.0), "the worker thread never started running" + # The worker is now running and cannot finish while this thread + # owns the lock. + assert not finished.wait(timeout=0.5), ( + "to_dict() completed while another thread held _lock, so it is " + "reading graph state unguarded" + ) + + assert finished.wait(timeout=5.0), "to_dict() did not complete after _lock was released" + worker.join(timeout=5.0) + assert not worker.is_alive(), "the worker thread is still running after to_dict() finished" + + def test_to_dict_is_reentrant_for_a_caller_holding_the_lock(self): + """``_lock`` is an ``RLock``, so lock-holding callers must not deadlock. + + The nested acquisition runs in a daemon worker joined with a timeout so + that a non-reentrant lock fails the test instead of hanging it. + """ + graph = _seeded_graph(10) + result = {} + + def nested_snapshot(): + with graph._lock: + result["snapshot"] = graph.to_dict() + + worker = threading.Thread(target=nested_snapshot, daemon=True) + worker.start() + worker.join(timeout=5.0) + + assert not worker.is_alive(), ( + "to_dict() deadlocked when called by a thread already holding " + "_lock -- the lock is no longer reentrant" + ) + assert len(result["snapshot"]["nodes"]) == 10 + + +class TestToDictUnderConcurrentWrites: + """The reported race: snapshot one thread, mutate from another.""" + + def _run_race(self, graph: ContextGraph, reader, duration: float = 1.0): + """Hammer ``reader`` while a writer adds nodes. Returns (errors, reads).""" + stop = threading.Event() + errors = [] + reads = [] + + def writer(): + i = 0 + while not stop.is_set(): + try: + graph.add_node(f"w{i}", "written") + except Exception as exc: # pragma: no cover - writer must stay healthy + errors.append(exc) + return + i += 1 + + def reader_loop(): + while not stop.is_set(): + try: + reads.append(reader()) + except Exception as exc: + errors.append(exc) + stop.set() + return + + threads = [ + threading.Thread(target=writer, daemon=True), + threading.Thread(target=reader_loop, daemon=True), + ] + for thread in threads: + thread.start() + time.sleep(duration) + stop.set() + for thread in threads: + thread.join(timeout=5.0) + assert not thread.is_alive(), ( + "a worker thread was still running 5s after the stop signal -- " + "a hang here would otherwise leak into subsequent tests" + ) + + return errors, reads + + def test_to_dict_does_not_raise_during_concurrent_writes(self): + graph = _seeded_graph() + errors, reads = self._run_race(graph, graph.to_dict) + + assert not errors, f"to_dict() raised under concurrent writes: {errors[0]!r}" + assert reads, "the reader thread never completed a to_dict() call" + + def test_to_dict_snapshot_is_internally_consistent(self): + """The reported statistics must describe the payload actually emitted. + + ``to_dict`` builds ``nodes``/``edges`` and then reads ``len(self.nodes)`` + and ``len(self.edges)`` for its ``statistics`` block. Unguarded, a write + landing between those steps yields counts that contradict the lists. + """ + graph = _seeded_graph() + errors, reads = self._run_race(graph, graph.to_dict) + + assert not errors, f"to_dict() raised under concurrent writes: {errors[0]!r}" + assert reads, "the reader thread never completed a to_dict() call" + for snapshot in reads: + stats = snapshot["statistics"] + assert stats["node_count"] == len(snapshot["nodes"]), ( + f"statistics.node_count={stats['node_count']} contradicts the " + f"{len(snapshot['nodes'])} nodes in the same snapshot" + ) + assert stats["edge_count"] == len(snapshot["edges"]), ( + f"statistics.edge_count={stats['edge_count']} contradicts the " + f"{len(snapshot['edges'])} edges in the same snapshot" + ) + + def test_snapshot_node_ids_are_unique(self): + """A torn read can emit the same node twice; a locked one cannot.""" + graph = _seeded_graph() + errors, reads = self._run_race(graph, graph.to_dict) + + assert not errors, f"to_dict() raised under concurrent writes: {errors[0]!r}" + for snapshot in reads: + ids = [node["id"] for node in snapshot["nodes"]] + assert len(ids) == len(set(ids)), "to_dict() emitted duplicate node ids"