diff --git a/CHANGELOG.md b/CHANGELOG.md index ae58e4c560..a211cac13c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,7 @@ Full release notes with details on each version: [GitHub Releases](https://githu ## 0.9.61 (2026-09-12) +- Fix: `graphify global add` now offsets each repo's community ids into a shared id space and links identically declared types across repos, the same #3014/#3007 fixes `merge-graphs` already had — a store built through the incremental add path previously fused unrelated repos' communities into one at id 0 and never linked a single shared type. `graphify global add`/`global remove` also now hold an exclusive lock across their whole load, mutate, save cycle — without it, two concurrent calls could each read the same snapshot and the second save would silently discard the first's work entirely, losing an entire repo's added nodes (#3100, thanks @ayushcodes10). - Fix: `graphify.serve` now imports cleanly on Python 3.12 and 3.13. The `chinese` extra pins `jieba-py` from 3.12 onward (0.9.60 mistakenly kept the old `jieba` until 3.14, and its invalid regex escapes are a hard error on 3.12+), and the jieba import now suppresses the tokenizer's `SyntaxWarning` regardless of message or line so it never escalates under `-W error`. - Fix: the git hook's rebuild-root guard now rejects a symlink-loop or dangling `.graphify_root` on Python 3.13, whose `Path.resolve()` no longer raises on a loop — the saved root must resolve to a real directory inside the repo before it is adopted. diff --git a/graphify/cli.py b/graphify/cli.py index feecee3841..80cc0d7fe0 100644 --- a/graphify/cli.py +++ b/graphify/cli.py @@ -3145,6 +3145,9 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": if result.get("cross_repo_calls"): print(f" resolved {result['cross_repo_calls']} " f"member call(s) across repos") + if result.get("shared_type_links"): + print(f" linked {result['shared_type_links']} " + f"type declaration(s) shared across repos") except Exception as exc: print(f"error: {exc}", file=sys.stderr); sys.exit(1) elif subcmd == "remove": diff --git a/graphify/global_graph.py b/graphify/global_graph.py index cd1ea19244..acdb9b811d 100644 --- a/graphify/global_graph.py +++ b/graphify/global_graph.py @@ -1,4 +1,5 @@ from __future__ import annotations +import contextlib import json import hashlib import sys @@ -12,6 +13,45 @@ _GLOBAL_MANIFEST = _GLOBAL_DIR / "global-manifest.json" +@contextlib.contextmanager +def _global_store_lock(): + """Exclusive advisory lock around a read-modify-write cycle on the + global graph store. + + global_add/global_remove each load the graph and manifest, mutate them + in memory, then save both back. Two concurrent calls would both read + the same pre-write snapshot, compute conflicting results from it, and + the second save would silently discard the first's work entirely (not + just a duplicated community id, which the narrower #3100 fix already + closes). Blocks until acquired rather than failing, since a caller + should wait its turn; released automatically if the process is killed + (fcntl.flock), so no stale-lock cleanup is needed. + + No-op on platforms without fcntl (Windows) -- matching the same + fallback already used by watch.py's per-repo rebuild lock, since a lost + update there is a pre-existing risk this does not newly introduce. + """ + try: + import fcntl + except ImportError: + yield + return + # Read _GLOBAL_DIR fresh rather than a module-level path computed from it + # once at import time, so a caller (tests included) that points the + # store elsewhere by rebinding _GLOBAL_DIR is honored here too. + _GLOBAL_DIR.mkdir(parents=True, exist_ok=True) + fh = open(_GLOBAL_DIR / ".global-graph.lock", "a+", encoding="utf-8") + try: + fcntl.flock(fh.fileno(), fcntl.LOCK_EX) + yield + finally: + try: + fcntl.flock(fh.fileno(), fcntl.LOCK_UN) + except OSError: + pass + fh.close() + + def _load_manifest() -> dict: if _GLOBAL_MANIFEST.exists(): try: @@ -88,101 +128,134 @@ def global_add(source_path: Path, repo_tag: str) -> dict: if not source_path.exists(): raise FileNotFoundError(f"graph not found: {source_path}") - manifest = _load_manifest() src_hash = _file_hash(source_path) - existing = manifest["repos"].get(repo_tag, {}) - existing_path = existing.get("source_path", "") - if existing_path and existing_path != str(source_path.resolve()): - print( - f"[graphify global] warning: repo tag '{repo_tag}' previously pointed to " - f"{existing_path!r}, now updating to {str(source_path.resolve())!r}. " - f"Use --as to give it a different name.", - file=sys.stderr, - ) - if existing.get("source_hash") == src_hash: - return {"repo_tag": repo_tag, "nodes_added": 0, "nodes_removed": 0, "skipped": True, - "cross_repo_calls": 0} - - # Load source graph - from graphify.security import check_graph_file_size_cap - check_graph_file_size_cap(source_path) - data = json.loads(source_path.read_text(encoding="utf-8")) - if "links" not in data and "edges" in data: - data = dict(data, links=data["edges"]) - try: - src_G = _jg.node_link_graph(data, edges="links") - except TypeError: - src_G = _jg.node_link_graph(data) - - # Prefix IDs for cross-project isolation - prefixed = prefix_graph_for_global(src_G, repo_tag) - - # Load global graph and prune stale nodes for this repo - G = _load_global_graph() - removed = prune_repo_from_graph(G, repo_tag) - - # Merge external-library nodes (no source_file) by label to avoid duplication - external_labels = { - d.get("label", ""): n - for n, d in G.nodes(data=True) - if not d.get("source_file") and d.get("label") - } - # Map each deduplicated external onto the existing global node so that - # edges incident to it can be rewired instead of dropped. - remap = {} - for node, data in prefixed.nodes(data=True): - if not data.get("source_file") and data.get("label") in external_labels: - remap[node] = external_labels[data["label"]] - - # Compose: add prefixed nodes (except deduplicated externals) into global graph - for node, data in prefixed.nodes(data=True): - if node not in remap: - G.add_node(node, **data) - for u, v, data in prefixed.edges(data=True): - u = remap.get(u, u) - v = remap.get(v, v) - if u != v: # don't introduce self-loops via remapping - G.add_edge(u, v, **data) - - added = prefixed.number_of_nodes() - len(remap) - # A member call parked on a caller node (#3152) may be answered by a repo - # already in the global graph, or by this one for a repo added earlier. The - # pass recomputes its own output, so adding repos one at a time lands where a - # single merge-graphs of the same inputs would. - from graphify.cross_repo_calls import link_cross_repo_member_calls - - cross_repo_calls = link_cross_repo_member_calls(G) - _save_global_graph(G) - - manifest["repos"][repo_tag] = { - "added_at": datetime.now(timezone.utc).isoformat(), - "source_path": str(source_path.resolve()), - "node_count": added, - "edge_count": prefixed.number_of_edges(), - "source_hash": src_hash, - } - _save_manifest(manifest) + with _global_store_lock(): + manifest = _load_manifest() + + existing = manifest["repos"].get(repo_tag, {}) + existing_path = existing.get("source_path", "") + if existing_path and existing_path != str(source_path.resolve()): + print( + f"[graphify global] warning: repo tag '{repo_tag}' previously pointed to " + f"{existing_path!r}, now updating to {str(source_path.resolve())!r}. " + f"Use --as to give it a different name.", + file=sys.stderr, + ) + if existing.get("source_hash") == src_hash: + return {"repo_tag": repo_tag, "nodes_added": 0, "nodes_removed": 0, "skipped": True, + "cross_repo_calls": 0, "shared_type_links": 0} + + # Load source graph + from graphify.security import check_graph_file_size_cap + check_graph_file_size_cap(source_path) + data = json.loads(source_path.read_text(encoding="utf-8")) + if "links" not in data and "edges" in data: + data = dict(data, links=data["edges"]) + try: + src_G = _jg.node_link_graph(data, edges="links") + except TypeError: + src_G = _jg.node_link_graph(data) + + # Load global graph and prune stale nodes for this repo, before prefixing + # the incoming one: the offset below reads the store's community ids as + # they stand once this repo's own stale entries are already gone, so + # re-adding the same repo cannot inflate it forever. + G = _load_global_graph() + removed = prune_repo_from_graph(G, repo_tag) + + # Offset the incoming repo's community ids past every other repo's + # already in the store (#3014, #3100): every graph.json numbers its own + # communities from 0, and the CLI merge-graphs command already offsets + # for exactly this reason, but the incremental add path here kept + # prefixing each new repo's communities from 0 too, colliding with + # whatever id another repo already occupied. + community_offset = 0 + existing_cids = [ + d["community"] for _, d in G.nodes(data=True) if isinstance(d.get("community"), int) + ] + if existing_cids: + community_offset = max(existing_cids) + 1 + + # Prefix IDs for cross-project isolation + prefixed = prefix_graph_for_global(src_G, repo_tag, community_offset=community_offset) + + # Merge external-library nodes (no source_file) by label to avoid duplication + external_labels = { + d.get("label", ""): n + for n, d in G.nodes(data=True) + if not d.get("source_file") and d.get("label") + } + # Map each deduplicated external onto the existing global node so that + # edges incident to it can be rewired instead of dropped. + remap = {} + for node, data in prefixed.nodes(data=True): + if not data.get("source_file") and data.get("label") in external_labels: + remap[node] = external_labels[data["label"]] + + # Compose: add prefixed nodes (except deduplicated externals) into global graph + for node, data in prefixed.nodes(data=True): + if node not in remap: + G.add_node(node, **data) + for u, v, data in prefixed.edges(data=True): + u = remap.get(u, u) + v = remap.get(v, v) + if u != v: # don't introduce self-loops via remapping + G.add_edge(u, v, **data) + + added = prefixed.number_of_nodes() - len(remap) + # A member call parked on a caller node (#3152) may be answered by a repo + # already in the global graph, or by this one for a repo added earlier. The + # pass recomputes its own output, so adding repos one at a time lands where a + # single merge-graphs of the same inputs would. + from graphify.cross_repo_calls import link_cross_repo_member_calls + + cross_repo_calls = link_cross_repo_member_calls(G) + + # A type both this repo and an earlier one declare arrives as two + # unconnected nodes, since every id is repo prefixed. The CLI batch + # merge command already links them so a traversal can cross the repo + # boundary (#3007); global_add never called this pass, so an + # incrementally built store held zero same_type_as edges no matter how + # many repos actually shared a type. Re-run over the whole store on + # every add, same as the member call pass above: it only adds an edge + # where none exists yet, so repeated calls across successive adds stay + # cheap and cannot double an edge. + from graphify.cross_repo_types import link_shared_type_declarations + + shared_type_links = link_shared_type_declarations(G) + _save_global_graph(G) + + manifest["repos"][repo_tag] = { + "added_at": datetime.now(timezone.utc).isoformat(), + "source_path": str(source_path.resolve()), + "node_count": added, + "edge_count": prefixed.number_of_edges(), + "source_hash": src_hash, + } + _save_manifest(manifest) return {"repo_tag": repo_tag, "nodes_added": added, "nodes_removed": removed, - "skipped": False, "cross_repo_calls": cross_repo_calls} + "skipped": False, "cross_repo_calls": cross_repo_calls, + "shared_type_links": shared_type_links} def global_remove(repo_tag: str) -> int: """Remove all nodes for repo_tag from the global graph. Returns count removed.""" from graphify.build import prune_repo_from_graph - manifest = _load_manifest() - if repo_tag not in manifest["repos"]: - raise KeyError(f"repo '{repo_tag}' not in global graph") + with _global_store_lock(): + manifest = _load_manifest() + if repo_tag not in manifest["repos"]: + raise KeyError(f"repo '{repo_tag}' not in global graph") - G = _load_global_graph() - removed = prune_repo_from_graph(G, repo_tag) - _save_global_graph(G) + G = _load_global_graph() + removed = prune_repo_from_graph(G, repo_tag) + _save_global_graph(G) - del manifest["repos"][repo_tag] - _save_manifest(manifest) - return removed + del manifest["repos"][repo_tag] + _save_manifest(manifest) + return removed def global_list() -> dict: diff --git a/tests/test_global_graph.py b/tests/test_global_graph.py index ad5da774d2..690a67e5c6 100644 --- a/tests/test_global_graph.py +++ b/tests/test_global_graph.py @@ -198,6 +198,75 @@ def test_global_add_two_repos_no_collision(tmp_path): assert G.number_of_nodes() == 2 # no silent merge +# ── #3100: community offset and shared type linking parity with merge-graphs ── + +def test_global_add_offsets_community_ids_across_repos(tmp_path): + """merge-graphs already offsets each input's community ids into a shared + id space (#3014); global_add builds the same kind of store with the same + prefixer but kept the default (no) offset, so two repos both numbering + their own communities from 0 collided in the merged store -- worst of + all at id 0, which every repo starts numbering from.""" + g1 = tmp_path / "graph1.json" + g2 = tmp_path / "graph2.json" + G1 = _make_graph([ + {"id": "a", "label": "A", "source_file": "a.py", "community": 0}, + {"id": "b", "label": "B", "source_file": "b.py", "community": 1}, + ]) + G2 = _make_graph([ + {"id": "c", "label": "C", "source_file": "c.py", "community": 0}, + {"id": "d", "label": "D", "source_file": "d.py", "community": 1}, + ]) + _graph_to_json(G1, g1) + _graph_to_json(G2, g2) + + global_dir = tmp_path / ".graphify" + with patch("graphify.global_graph._GLOBAL_DIR", global_dir), \ + patch("graphify.global_graph._GLOBAL_GRAPH", global_dir / "global-graph.json"), \ + patch("graphify.global_graph._GLOBAL_MANIFEST", global_dir / "global-manifest.json"): + from graphify.global_graph import global_add, _load_global_graph + global_add(g1, "repoA") + global_add(g2, "repoB") + G = _load_global_graph() + + by_repo: dict[str, set[int]] = {} + for _, data in G.nodes(data=True): + by_repo.setdefault(data["repo"], set()).add(data["community"]) + assert by_repo["repoA"].isdisjoint(by_repo["repoB"]), ( + f"community ids collide across repos: {by_repo}" + ) + + +def test_global_add_links_shared_type_declarations(tmp_path): + """merge-graphs already links identically declared types across repos so + a traversal can cross the repo boundary (#3007); global_add never called + that pass, so an incrementally built store held zero same_type_as edges + no matter how many repos actually shared a type.""" + g1 = tmp_path / "graph1.json" + g2 = tmp_path / "graph2.json" + shared = { + "id": "contracttype", "label": "ContractType", "source_file": "models.cs", + "_callable_class": True, "metadata": {"namespace": "Acme.Contracts"}, + } + G1 = _make_graph([shared]) + G2 = _make_graph([shared]) + _graph_to_json(G1, g1) + _graph_to_json(G2, g2) + + global_dir = tmp_path / ".graphify" + with patch("graphify.global_graph._GLOBAL_DIR", global_dir), \ + patch("graphify.global_graph._GLOBAL_GRAPH", global_dir / "global-graph.json"), \ + patch("graphify.global_graph._GLOBAL_MANIFEST", global_dir / "global-manifest.json"): + from graphify.global_graph import global_add, _load_global_graph + first = global_add(g1, "repoA") + second = global_add(g2, "repoB") + G = _load_global_graph() + + assert first["shared_type_links"] == 0 # nothing to link against yet + assert second["shared_type_links"] == 1 + assert G.has_edge("repoA::contracttype", "repoB::contracttype") + assert G["repoA::contracttype"]["repoB::contracttype"]["relation"] == "same_type_as" + + def test_global_remove(tmp_path): src_graph = tmp_path / "graph.json" G = _make_graph([{"id": "userservice", "label": "UserService", "source_file": "src/user.py"}]) @@ -385,3 +454,87 @@ def test_global_add_rejects_oversized_source_graph(monkeypatch, tmp_path): from graphify.global_graph import global_add with pytest.raises(ValueError, match="exceeds"): global_add(src_graph, "repoA") + + +def test_global_store_lock_serializes_concurrent_critical_sections(tmp_path): + """Review finding: global_add/global_remove each load-mutate-save the + shared store with no locking, so two concurrent calls read the same + pre-write snapshot, compute conflicting results, and the second save + silently discards the first's work entirely. The lock added to close + this must actually provide mutual exclusion -- verified directly by + running several threads through the critical section and confirming + at most one is ever inside it at once, rather than through global_add + itself (racing its real logic reliably, without deadlocking a test + that also needs to pass once the lock works, is far harder to get + right than testing the lock's own guarantee in isolation).""" + import threading + import time + from graphify import global_graph as gg_mod + + global_dir = tmp_path / ".graphify" + active = {"count": 0} + max_active = {"value": 0} + counter_lock = threading.Lock() + + def worker(): + with gg_mod._global_store_lock(): + with counter_lock: + active["count"] += 1 + max_active["value"] = max(max_active["value"], active["count"]) + time.sleep(0.05) + with counter_lock: + active["count"] -= 1 + + with patch("graphify.global_graph._GLOBAL_DIR", global_dir): + threads = [threading.Thread(target=worker) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + + assert max_active["value"] == 1, ( + f"the lock let {max_active['value']} threads into the critical section at once" + ) + + +def test_global_add_concurrent_calls_both_survive(tmp_path): + """End-to-end: two different repos added via genuinely concurrent + global_add calls must both still be present afterward -- the lock + from the finding above must be held across the real load/mutate/save + cycle, not just demonstrated in isolation.""" + import threading + from graphify import global_graph as gg_mod + + src_a = tmp_path / "a.json" + src_b = tmp_path / "b.json" + _graph_to_json( + _make_graph([{"id": "a1", "label": "A1", "source_file": "a.py"}]), src_a + ) + _graph_to_json( + _make_graph([{"id": "b1", "label": "B1", "source_file": "b.py"}]), src_b + ) + + global_dir = tmp_path / ".graphify" + errors: list[Exception] = [] + + def add(src, tag): + try: + gg_mod.global_add(src, tag) + except Exception as exc: # pragma: no cover - surfaced via assertion below + errors.append(exc) + + with patch("graphify.global_graph._GLOBAL_DIR", global_dir), \ + patch("graphify.global_graph._GLOBAL_GRAPH", global_dir / "global-graph.json"), \ + patch("graphify.global_graph._GLOBAL_MANIFEST", global_dir / "global-manifest.json"): + t1 = threading.Thread(target=add, args=(src_a, "repoA")) + t2 = threading.Thread(target=add, args=(src_b, "repoB")) + t1.start() + t2.start() + t1.join(timeout=10) + t2.join(timeout=10) + + assert not errors, errors + manifest = json.loads((global_dir / "global-manifest.json").read_text()) + assert set(manifest["repos"]) == {"repoA", "repoB"}, ( + f"a concurrent add lost the other's update, got {set(manifest['repos'])}" + )