From 24c45a01423fe1c9d93cca7d65ce6caf6abbe3eb Mon Sep 17 00:00:00 2001 From: Gal Shubeli Date: Tue, 25 Aug 2026 11:31:24 +0300 Subject: [PATCH] fix(falkordb): make the graph-DB push converge, batch its writes, add a repo-keyed delta (#3057) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `push_to_falkordb` is MERGE-only, so once `graphify global add` prunes a repo out of the global graph, a full re-push leaves every one of those nodes in the target permanently: the database diverges from graph.json and never converges back. @Azeem1985 measured +1,250 nodes / +1,151 edges of surplus on a 75k-node global graph, with 25 of 25 sampled pruned ids still present after a full re-push. The same code path also sent one query per node and one per edge, and matched edge endpoints with a label-free `MATCH (a {id: $src})` that no index can serve. On 20 repos / 20k nodes a full push took 79.7s; batched UNWIND against an indexed `:Entity` label takes 2.9s — 27x — and the labelling change is what makes the index reachable at all, since nodes previously carried only their file-type label. What is new: - Batched UNWIND writes, an `:Entity` label indexed on `id` alongside the existing file-type label, and a one-time backfill so MERGE adopts nodes written by an older graphify instead of duplicating them. This is also @kingjin94's #2258 diagnosis. - `--graph-name`. The parameter always existed on the writer; the CLI never passed it, so every CLI push landed on the `graphify` key with no way to aim it elsewhere. That is what turned a mistake into a 1.2M-node deletion. - `--prune` converges the target: nodes AND edges. DETACH DELETE takes a pruned node's edges with it, but an edge dropped between two surviving endpoints needs its own sweep, which is the surplus-edge half of the report. Opt-in, because `--graph-name` never existed and some `graphify` keys hold several projects merged together, where a converging default would either silently delete or hard-refuse on a routine re-push. - An add-only push now reports the drift it cannot fix, so the divergence stops being silent even when nobody passes `--prune`. - Repo-keyed delta (`graphify global push`, default). Mirrors global_add's own contract: the repo is the unit of change, keyed on the manifest source_hash global_add already records and already uses for its own skip. 5.0% of the rows for a 1-of-20-repo change; 0 rows when nothing moved. - Drift repair. Push state lives in the target as :GraphifyPushState nodes, cross-checked against the target's own per-repo counts. A local ledger still reads clean after a wipe or a half-landed run and never repairs it. - Cross-repo edges survive a delta. global_add remaps external-library nodes onto whichever repo first contributed them, so a B->A edge is owned by neither alone; the delta re-sends every edge incident to a rewritten repo, not only its internal ones. - Deletes are capped at 20% of the target unless `--allow-shrink` (the #479 rule applied to the push) and paged with the LIMIT inside a WITH, because FalkorDB's LIMIT does not short-circuit an eager DELETE. - `graphify global push`. `export --push` resolves its source from a project output directory, so the global graph had no CLI push route at all. The Neo4j writer is untouched — same defects, but a fix cannot be verified without a Neo4j instance. The new flags are refused on `export neo4j` rather than silently ignored, since accepting `--prune` there would report a converged push that converged nothing. 15 new integration tests against a live FalkorDB; they fail on this commit's parent and pass here. No CHANGELOG entry — this repo folds those at release. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 7 + graphify/cli.py | 132 ++++++++- graphify/exporters/graphdb.py | 460 ++++++++++++++++++++++++++--- graphify/global_graph.py | 36 +++ tests/test_falkordb_integration.py | 345 ++++++++++++++++++++++ 5 files changed, 932 insertions(+), 48 deletions(-) diff --git a/README.md b/README.md index 0c14d207c9..15e9853f75 100644 --- a/README.md +++ b/README.md @@ -758,10 +758,17 @@ graphify export callflow-html --max-sections 8 # cap generated architecture graphify export callflow-html --output docs/arch.html graphify export callflow-html ./some-repo/graphify-out +graphify export falkordb --push falkordb://localhost:6379 # push this project's graph +graphify export falkordb --push falkordb://localhost:6379 --graph-name stg # choose the target graph +graphify export falkordb --push falkordb://localhost:6379 --prune # mirror: delete what the source dropped + graphify global add graphify-out/graph.json --as myrepo # register a project graph into ~/.graphify/global-graph.json graphify global remove myrepo # remove a project from the global graph graphify global list # show all registered repos + node/edge counts graphify global path # print path to the global graph file +graphify global push falkordb://localhost:6379 # push the global graph to FalkorDB (delta by default) +graphify global push falkordb://host:6379 --graph-name staging # pick the target graph in the instance +graphify global push falkordb://host:6379 --full --prune # re-send every repo and converge the target graphify prs # PR dashboard: CI, review, worktree, graph impact graphify prs 42 # deep dive on PR #42 diff --git a/graphify/cli.py b/graphify/cli.py index 5b73397266..5b22312387 100644 --- a/graphify/cli.py +++ b/graphify/cli.py @@ -2616,8 +2616,11 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": print(" graphml [--graph PATH]", file=sys.stderr) print(" neo4j [--graph PATH] [--push URI] [--user U] [--password P]", file=sys.stderr) print(" (or set NEO4J_PASSWORD instead of --password to keep it off argv)", file=sys.stderr) - print(" falkordb [--graph PATH] [--push URI] [--user U] [--password P]", file=sys.stderr) + print(" falkordb [--graph PATH] [--push URI] [--user U] [--password P] [--graph-name NAME]", file=sys.stderr) + print(" [--prune] [--allow-shrink]", file=sys.stderr) print(" (or set FALKORDB_PASSWORD instead of --password to keep it off argv)", file=sys.stderr) + print(" --graph-name selects the target graph in the instance (default \"graphify\");", file=sys.stderr) + print(" --prune deletes what the source no longer has so the target mirrors it.", file=sys.stderr) sys.exit(1) # Parse shared args @@ -2651,6 +2654,12 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": os.environ.get("FALKORDB_PASSWORD") if subcmd == "falkordb" else os.environ.get("NEO4J_PASSWORD") ) or None + # Target selection inside the server (falkordb only). Never exposed + # before, so every CLI push landed on the "graphify" key regardless of + # what that key already held (#3057). + push_graph_name = "graphify" # falkordb: named graph in the instance + push_prune = False # falkordb: delete what the source no longer has + push_allow_shrink = False # falkordb: override the prune size guard i = 0 while i < len(args): a = args[i] @@ -2706,6 +2715,12 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": push_user = args[i + 1]; i += 2 elif a == "--password" and i + 1 < len(args): push_password = args[i + 1]; i += 2 + elif a == "--graph-name" and i + 1 < len(args): + push_graph_name = args[i + 1]; i += 2 + elif a == "--prune": + push_prune = True; i += 1 + elif a == "--allow-shrink": + push_allow_shrink = True; i += 1 elif subcmd == "callflow-html" and not a.startswith("-") and not graph_path_explicit: candidate = Path(a) if candidate.name == "graph.json" or candidate.suffix.lower() == ".json": @@ -2883,6 +2898,24 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": print(f"graph.graphml written - open in Gephi, yEd, or any GraphML tool") elif subcmd == "neo4j": + # --graph-name/--prune/--allow-shrink only exist on the FalkorDB + # writer. Refuse rather than ignore: silently accepting --prune here + # would report a converged push that never deleted anything. + _falkor_only = [ + name for name, given in ( + ("--graph-name", push_graph_name != "graphify"), + ("--prune", push_prune), + ("--allow-shrink", push_allow_shrink), + ) if given + ] + if _falkor_only: + print( + f"error: {', '.join(_falkor_only)} " + f"{'is' if len(_falkor_only) == 1 else 'are'} supported only by " + f"`graphify export falkordb`.", + file=sys.stderr, + ) + sys.exit(1) if push_uri: from graphify.export import push_to_neo4j as _push if push_password is None: @@ -2899,9 +2932,27 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": elif subcmd == "falkordb": if push_uri: from graphify.export import push_to_falkordb as _push - result = _push(G, uri=push_uri, user=push_user, - password=push_password, communities=communities) - print(f"Pushed to FalkorDB: {result['nodes']} nodes, {result['edges']} edges") + try: + result = _push(G, uri=push_uri, user=push_user, + password=push_password, communities=communities, + graph_name=push_graph_name, prune=push_prune, + allow_shrink=push_allow_shrink) + except ValueError as exc: # prune size guard + print(f"error: {exc}", file=sys.stderr) + sys.exit(1) + _summary = f"{result['nodes']} nodes, {result['edges']} edges" + if push_prune: + _summary += ( + f" (pruned {result['deleted']} nodes, " + f"{result['deleted_edges']} edges)" + ) + print(f"Pushed to FalkorDB [{push_graph_name}]: {_summary}") + if not push_prune and result.get("target_surplus"): + print( + f" note: target has {result['target_surplus']} node(s) the " + f"source does not; --prune converges it.", + file=sys.stderr, + ) else: from graphify.export import to_cypher as _to_cypher _to_cypher(G, str(out_dir / "cypher.txt")) @@ -2934,6 +2985,7 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": global_remove as _global_remove, global_list as _global_list, global_path as _global_path, + global_push as _global_push, ) if subcmd == "add": # graphify global add [--as ] @@ -2961,6 +3013,76 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": f"-{result['nodes_removed']} pruned. Global: {_global_path()}") except Exception as exc: print(f"error: {exc}", file=sys.stderr); sys.exit(1) + elif subcmd == "push": + # graphify global push [--graph-name N] [--full] [--prune] + # [--allow-shrink] [--user U] [--password P] + args = sys.argv[3:] + uri = None + g_name = "graphify" + g_user = None + g_password = os.environ.get("FALKORDB_PASSWORD") or None + g_delta = True + g_prune = False + g_allow_shrink = False + i = 0 + while i < len(args): + a = args[i] + if a == "--graph-name" and i + 1 < len(args): + g_name = args[i + 1]; i += 2 + elif a == "--user" and i + 1 < len(args): + g_user = args[i + 1]; i += 2 + elif a == "--password" and i + 1 < len(args): + g_password = args[i + 1]; i += 2 + elif a == "--full": + g_delta = False; i += 1 + elif a == "--prune": + g_prune = True; i += 1 + elif a == "--allow-shrink": + g_allow_shrink = True; i += 1 + elif not uri and not a.startswith("-"): + uri = a; i += 1 + else: + i += 1 + if not uri: + print( + "Usage: graphify global push [--graph-name NAME] [--full] " + "[--prune] [--allow-shrink]\n" + " Delta by default: only repos whose source changed (or whose " + "count in the target drifted) are re-sent.\n" + " --full re-sends every repo; add --prune to make a full push " + "converge instead of only adding.", + file=sys.stderr, + ) + sys.exit(1) + try: + res = _global_push( + uri, graph_name=g_name, user=g_user, password=g_password, + delta=g_delta, prune=g_prune, allow_shrink=g_allow_shrink, + ) + except (ValueError, FileNotFoundError) as exc: + print(f"error: {exc}", file=sys.stderr); sys.exit(1) + if g_delta: + pushed = res.get("repos_pushed", []) + removed = res.get("repos_removed", []) + skipped = res.get("repos_skipped", []) + if not pushed and not removed: + print(f"Global graph [{g_name}]: up to date, {len(skipped)} repo(s) unchanged.") + else: + print( + f"Global graph [{g_name}]: {res['nodes']} nodes, " + f"{res['edges']} edges across {len(pushed)} repo(s); " + f"{len(skipped)} unchanged, {res['deleted']} nodes pruned." + ) + for tag in pushed: + print(f" re-pushed {tag} ({res.get('reasons', {}).get(tag, 'changed')})") + for tag in removed: + print(f" removed {tag} (no longer in the manifest)") + else: + line = f"Global graph [{g_name}]: {res['nodes']} nodes, {res['edges']} edges" + if g_prune: + line += (f" (pruned {res['deleted']} nodes, " + f"{res['deleted_edges']} edges)") + print(line) elif subcmd == "remove": tag = sys.argv[3] if len(sys.argv) > 3 else "" if not tag: @@ -2981,7 +3103,7 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": elif subcmd == "path": print(_global_path()) else: - print("Usage: graphify global [add|remove|list|path]", file=sys.stderr); sys.exit(1) + print("Usage: graphify global [add|remove|list|path|push]", file=sys.stderr); sys.exit(1) elif cmd == "extract": # Headless full-pipeline extraction for CI / scripts (#698). diff --git a/graphify/exporters/graphdb.py b/graphify/exporters/graphdb.py index 14c47f0d5c..7daf3a1c96 100644 --- a/graphify/exporters/graphdb.py +++ b/graphify/exporters/graphdb.py @@ -1,9 +1,90 @@ -"""graphdb — moved verbatim from graphify/export.py.""" +"""graphdb — direct push of a graphify graph into Neo4j / FalkorDB. + +The FalkorDB writer batches its writes and keys every node on a shared +``:Entity`` label indexed on ``id``, plus the node's file-type label for +display (``:Entity:Python``). That buys two things the previous per-row writer +did not have: + + - edge-endpoint MATCHes resolve through an index instead of scanning every + node in the graph once per edge (#2258); + - writes go out as batched ``UNWIND`` statements rather than one round trip + per node and one per edge. + +Convergence (#3057). By default a push only adds and updates, so anything the +source has since pruned survives in the target forever and the two silently +diverge — a `global add` that prunes a repo, followed by any number of full +re-pushes, never removes those nodes. ``prune=True`` makes the push *converge*: +every node and edge this push did not write is deleted, so the target ends up +an exact mirror of the source. Edges need their own sweep, not just the nodes: +``DETACH DELETE`` takes a pruned node's edges with it, but an edge dropped +between two endpoints that both survive would otherwise linger forever. + +Because pruning is destructive it is opt-in, and it refuses to run when the +deletion would exceed ``shrink_limit`` of the target unless +``allow_shrink=True`` — the same "refuse to SILENTLY drop nodes" rule as the +#479 build guard. + +Pruning deletes by *absence from this push*, not by repo. Point a push at a +graph holding anything you did not push and ``prune=True`` will remove it; use +``graph_name`` to give each source its own target graph. + +The Neo4j writer is unchanged. It has the same per-row and unindexed-MATCH +problems, but a fix cannot be verified without a Neo4j instance, so it keeps +its old add-only behaviour and the new options are refused rather than +silently ignored on that path. +""" from __future__ import annotations -from graphify.analyze import _node_community_map import networkx as nx import re +import time + +from graphify.analyze import _node_community_map + +# Rows per UNWIND batch. +_BATCH = 1000 +# Rows deleted per convergence page. +_DELETE_PAGE = 10_000 +# Refuse a prune that would delete more than this fraction of the target. +_DEFAULT_SHRINK_LIMIT = 0.20 +# Stamped on every node and edge a push writes; convergence deletes whatever +# does not carry the current value. +_EPOCH_PROP = "graphify_push_epoch" + + +def _safe_rel(relation: str) -> str: + return re.sub(r"[^A-Z0-9_]", "_", relation.upper().replace(" ", "_").replace("-", "_")) or "RELATED_TO" + + +def _safe_label(label: str) -> str: + """Sanitize a node label to prevent Cypher injection.""" + sanitized = re.sub(r"[^A-Za-z0-9_]", "", label) + return sanitized if sanitized else "Entity" + + +def _scalar_props(data: dict) -> dict: + return { + k: v for k, v in data.items() + if isinstance(v, (str, int, float, bool)) and not k.startswith("_") + } + + +def _new_epoch() -> int: + """Identifier for one push. Millisecond clock, so two pushes into the same + target never collide and the value is meaningful when read back.""" + return int(time.time() * 1000) + + +def _chunked(iterable, size: int): + """Yield lists of at most `size` items, holding one chunk at a time.""" + chunk = [] + for item in iterable: + chunk.append(item) + if len(chunk) >= size: + yield chunk + chunk = [] + if chunk: + yield chunk def push_to_neo4j( @@ -77,6 +158,293 @@ def _safe_label(label: str) -> str: driver.close() return {"nodes": nodes_pushed, "edges": edges_pushed} + +# --------------------------------------------------------------------------- +# FalkorDB batched writer +# --------------------------------------------------------------------------- + +def _ensure_schema(graph) -> None: + """Index the shared label, and adopt nodes written before it existed. + + A graph pushed by an older graphify carries only its file-type label, so + MERGE on :Entity would create a duplicate beside every one of them. The + backfill is idempotent and a no-op on a graph this writer already owns. + """ + try: + graph.query("CREATE INDEX FOR (n:Entity) ON (n.id)") + except Exception: + pass # already exists, or an engine without the syntax — push still works + graph.query("MATCH (n) WHERE n.id IS NOT NULL AND NOT n:Entity SET n:Entity") + + +def _write_nodes(graph, rows, node_community: dict, epoch: int) -> int: + """Batched node upsert. `rows` yields (node_id, attrs).""" + pushed = 0 + for chunk in _chunked(rows, _BATCH): + by_label: dict[str, list[dict]] = {} + for node_id, data in chunk: + props = _scalar_props(data) + props["id"] = node_id + cid = node_community.get(node_id) + if cid is not None: + props["community"] = cid + props[_EPOCH_PROP] = epoch + ftype = _safe_label(str(data.get("file_type", "Entity")).capitalize()) + by_label.setdefault(ftype, []).append(props) + for ftype, batch in by_label.items(): + graph.query( + f"UNWIND $rows AS row MERGE (n:Entity {{id: row.id}}) " + f"SET n:{ftype} SET n += row", + {"rows": batch}, + ) + pushed += len(batch) + return pushed + + +def _write_edges(graph, rows, epoch: int) -> int: + """Batched edge upsert. `rows` yields (u, v, attrs).""" + pushed = 0 + for chunk in _chunked(rows, _BATCH): + by_rel: dict[str, list[dict]] = {} + for u, v, data in chunk: + rel = _safe_rel(data.get("relation", "RELATED_TO")) + props = _scalar_props(data) + props[_EPOCH_PROP] = epoch + by_rel.setdefault(rel, []).append({"src": u, "tgt": v, "props": props}) + for rel, batch in by_rel.items(): + graph.query( + f"UNWIND $rows AS row " + f"MATCH (a:Entity {{id: row.src}}), (b:Entity {{id: row.tgt}}) " + f"MERGE (a)-[r:{rel}]->(b) SET r += row.props", + {"rows": batch}, + ) + pushed += len(batch) + return pushed + + +# --------------------------------------------------------------------------- +# Convergence +# --------------------------------------------------------------------------- + +_STALE_NODES = f"MATCH (n:Entity) WHERE n.{_EPOCH_PROP} IS NULL OR n.{_EPOCH_PROP} <> $epoch" +_STALE_EDGES = ( + f"MATCH (:Entity)-[r]->(:Entity) " + f"WHERE r.{_EPOCH_PROP} IS NULL OR r.{_EPOCH_PROP} <> $epoch" +) + + +def _count(graph, cypher: str, params: dict) -> int: + return int(graph.query(cypher, params).result_set[0][0]) + + +def _delete_paged(graph, stale_match: str, var: str, params: dict, expected: int) -> int: + """Delete `stale_match` in pages until none remain. Returns rows deleted. + + LIMIT does not page a DELETE in FalkorDB: its known-limitations doc notes + LIMIT "does not currently short-circuit eager operations like CREATE, SET, + or DELETE", so `... DELETE n LIMIT $page` deletes everything matched rather + than one page. The LIMIT has to sit in a WITH that precedes the DELETE. + """ + verb = "DETACH DELETE" if var == "n" else "DELETE" + page = f"{stale_match} WITH {var} LIMIT {_DELETE_PAGE} {verb} {var}" + remaining = expected + while remaining > 0: + graph.query(page, params) + after = _count(graph, f"{stale_match} RETURN count({var})", params) + if after >= remaining: + raise RuntimeError( + f"graphify: prune stalled with {after} stale rows remaining (no " + f"progress in one page). Target may be read-only, or the delete " + f"may be racing another writer." + ) + remaining = after + return expected - remaining + + +def _converge(graph, epoch: int, allow_shrink: bool, shrink_limit: float) -> tuple[int, int]: + """Delete every node and edge this push did not write.""" + params = {"epoch": epoch} + stale_n = _count(graph, f"{_STALE_NODES} RETURN count(n)", params) + stale_e = _count(graph, f"{_STALE_EDGES} RETURN count(r)", params) + if stale_n <= 0 and stale_e <= 0: + return 0, 0 + + total_n = _count(graph, "MATCH (n:Entity) RETURN count(n)", {}) + total_e = _count(graph, "MATCH (:Entity)-[r]->(:Entity) RETURN count(r)", {}) + if not allow_shrink: + for kind, stale, total in (("nodes", stale_n, total_n), ("edges", stale_e, total_e)): + if total > 0 and (stale / total) > shrink_limit: + raise ValueError( + f"graphify: push --prune would delete {stale} of {total} " + f"{kind} ({stale / total:.0%}) from the target graph, over " + f"the {shrink_limit:.0%} safety limit. That usually means " + f"the push is aimed at the wrong graph — check --graph-name. " + f"Pass --allow-shrink if the removal is intended. Nothing " + f"was deleted; the additive part of this push has already " + f"been applied." + ) + + # Nodes first: DETACH DELETE takes their edges with them, so the edge sweep + # has less to do and its count is settled by the time it runs. + nodes_deleted = _delete_paged(graph, _STALE_NODES, "n", params, stale_n) if stale_n else 0 + remaining_e = _count(graph, f"{_STALE_EDGES} RETURN count(r)", params) + edges_deleted = ( + _delete_paged(graph, _STALE_EDGES, "r", params, remaining_e) if remaining_e else 0 + ) + return nodes_deleted, edges_deleted + + +# --------------------------------------------------------------------------- +# Repo-keyed delta +# +# `global add` already treats the repo as the unit of change: it prunes a repo +# whole, re-adds it whole, records a per-repo source_hash in the global +# manifest, and returns skipped=True when that hash has not moved. The delta +# push mirrors that contract instead of inventing one, so a 226-repo global +# graph with one changed repo sends one repo's rows rather than all of them. +# +# The "what did I last push" state lives in the TARGET database, not in a local +# ledger: a ledger cannot notice that the database was wiped or that a run +# half-landed — it still reads clean and the delta never repairs the drift. +# Reading the target's own per-repo node counts and re-pushing any repo whose +# count disagrees with the manifest turns silent permanent drift into automatic +# repair. +# --------------------------------------------------------------------------- + +_STATE_LABEL = "GraphifyPushState" + + +def _read_push_state(graph) -> dict[str, dict]: + rows = graph.query( + f"MATCH (s:{_STATE_LABEL}) RETURN s.repo, s.source_hash, s.node_count", {} + ).result_set or [] + return {r[0]: {"source_hash": r[1], "node_count": int(r[2] or 0)} for r in rows if r[0]} + + +def _target_repo_counts(graph) -> dict[str, int]: + """The target's OWN per-repo node counts — the check a ledger cannot do.""" + rows = graph.query( + "MATCH (n:Entity) WHERE n.repo IS NOT NULL RETURN n.repo, count(n)", {} + ).result_set or [] + return {r[0]: int(r[1]) for r in rows if r[0]} + + +def _plan_delta(manifest_repos: dict, state: dict, live_counts: dict): + """Decide which repos to re-push and which to delete.""" + changed, reasons = [], {} + for tag, info in manifest_repos.items(): + known = state.get(tag) + want_count = int(info.get("node_count") or 0) + live = live_counts.get(tag, 0) + if known is None: + changed.append(tag); reasons[tag] = "not present in target" + elif known.get("source_hash") != info.get("source_hash"): + changed.append(tag); reasons[tag] = "source changed" + elif live != want_count: + changed.append(tag) + reasons[tag] = f"target drift ({live} nodes in target, manifest says {want_count})" + removed = list(dict.fromkeys( + [t for t in list(state) + list(live_counts) if t not in manifest_repos] + )) + return changed, removed, reasons + + +def _index_repos(G: nx.Graph, tags): + """Bucket a global graph's nodes and edges by repo, in two passes. + + Edges are bucketed by *either* endpoint's repo, not by both. `global add` + remaps external-library nodes onto whichever repo first contributed them, + so a cross-repo edge B->A is owned by neither B nor A alone. Pruning repo A + drops that edge with A's node; re-adding only A's internal edges would not + bring it back and the target would quietly lose cross-repo connectivity on + every delta. + """ + wanted = set(tags) + nodes_by: dict[str, list] = {t: [] for t in wanted} + for nid, data in G.nodes(data=True): + tag = data.get("repo") + if tag in wanted: + nodes_by[tag].append((nid, data)) + edges_by: dict[str, list] = {t: [] for t in wanted} + for u, v, data in G.edges(data=True): + for tag in {G.nodes[u].get("repo"), G.nodes[v].get("repo")} & wanted: + edges_by[tag].append((u, v, data)) + return nodes_by, edges_by + + +def _prune_repo_paged(graph, tag: str) -> int: + """Delete one repo's nodes, paged. Returns the count removed.""" + scoped = "MATCH (n:Entity {repo: $t})" + params = {"t": tag} + n = _count(graph, f"{scoped} RETURN count(n)", params) + if n: + _delete_paged(graph, scoped, "n", params, n) + return n + + +def _push_delta( + G, graph, node_community: dict, epoch: int, manifest_repos: dict, + allow_shrink: bool, shrink_limit: float, +) -> dict: + state = _read_push_state(graph) + live_counts = _target_repo_counts(graph) + changed, removed, reasons = _plan_delta(manifest_repos, state, live_counts) + + # Size guard, before anything is deleted: a manifest that does not belong to + # this database looks exactly like a genuine mass removal. Same rule as the + # #479 build guard, applied to the push. + # + # Only NET removal counts. A re-pushed repo is pruned and immediately + # re-added, so its nodes are not lost — charging them here would refuse any + # delta touching more than shrink_limit of a small global graph. A repo that + # comes back SMALLER is a partial removal, so charge the difference: that is + # what catches "the manifest says 2 nodes, the target holds 50,000". + total_nodes = _count(graph, "MATCH (n:Entity) RETURN count(n)", {}) + doomed = sum(live_counts.get(t, 0) for t in removed) + doomed += sum( + max(0, live_counts.get(t, 0) - int(manifest_repos.get(t, {}).get("node_count") or 0)) + for t in changed + ) + if not allow_shrink and total_nodes > 0 and (doomed / total_nodes) > shrink_limit: + raise ValueError( + f"graphify: delta push would remove {doomed} of {total_nodes} nodes " + f"({doomed / total_nodes:.0%}) in the target graph, over the " + f"{shrink_limit:.0%} safety limit. That usually means this manifest " + f"does not belong to this database — check --graph-name. Pass " + f"--allow-shrink if it is intended. Nothing was changed." + ) + + nodes_by, edges_by = _index_repos(G, changed) if changed else ({}, {}) + + nodes_pushed = edges_pushed = deleted = 0 + for tag in changed: + deleted += _prune_repo_paged(graph, tag) + nodes_pushed += _write_nodes(graph, nodes_by.get(tag, []), node_community, epoch) + edges_pushed += _write_edges(graph, edges_by.get(tag, []), epoch) + info = manifest_repos.get(tag, {}) + graph.query( + f"MERGE (s:{_STATE_LABEL} {{repo: $repo}}) " + f"SET s.source_hash = $h, s.node_count = $n, s.epoch = $e", + {"repo": tag, "h": info.get("source_hash"), + "n": int(info.get("node_count") or 0), "e": epoch}, + ) + + for tag in removed: + deleted += _prune_repo_paged(graph, tag) + graph.query(f"MATCH (s:{_STATE_LABEL} {{repo: $repo}}) DELETE s", {"repo": tag}) + + return { + "nodes": nodes_pushed, + "edges": edges_pushed, + "deleted": deleted, + "deleted_edges": 0, # repo prune is DETACH DELETE; edges go with the nodes + "repos_pushed": changed, + "repos_removed": removed, + "repos_skipped": [t for t in manifest_repos if t not in changed], + "reasons": reasons, + } + + def push_to_falkordb( G: nx.Graph, uri: str, @@ -84,13 +452,16 @@ def push_to_falkordb( password: str | None = None, communities: dict[int, list[str]] | None = None, graph_name: str = "graphify", + prune: bool = False, + allow_shrink: bool = False, + shrink_limit: float = _DEFAULT_SHRINK_LIMIT, + repo_manifest: dict | None = None, ) -> dict[str, int]: """Push graph directly to a running FalkorDB instance via the Python SDK. Requires: pip install falkordb - FalkorDB is OpenCypher-compatible, so the MERGE/SET upsert queries are - identical to push_to_neo4j. Differences from the Neo4j path: + FalkorDB is OpenCypher-compatible. Differences from the Neo4j path: - connects with FalkorDB(host, port, username, password) instead of a bolt driver; only the host/port are read from the URI, so the scheme is informational - "falkordb://localhost:6379", "redis://localhost:6379" @@ -102,8 +473,23 @@ def push_to_falkordb( and password may be None. - no APOC: the Neo4j path does not use APOC either, so nothing to port. - Uses MERGE so re-running is safe - nodes and edges are upserted, not - duplicated. Returns a dict with counts of nodes and edges pushed. + Writes are batched UNWIND upserts against an indexed ``:Entity`` label, so + re-running is safe - nodes and edges are upserted, not duplicated. + + graph_name: which named graph in the instance to write. FalkorDB keys each + graph by name, so this is the difference between a staging graph and + production - set it explicitly for anything that matters. + prune: delete nodes and edges this push did not write, so the target + converges on the source instead of accumulating. See the module + docstring. + repo_manifest: the global manifest's ``repos`` dict. Switches the push into + repo-keyed DELTA mode - only repos whose ``source_hash`` moved (or whose + node count in the target has drifted from the manifest) are re-sent, and + repos the manifest no longer lists are deleted. Convergence is implied, + so ``prune`` is not needed with it. + + Returns a dict with counts of nodes and edges pushed, nodes and edges + deleted, and in delta mode the repos re-pushed / skipped / removed. """ try: from falkordb import FalkorDB @@ -115,14 +501,7 @@ def push_to_falkordb( from urllib.parse import urlparse node_community = _node_community_map(communities) if communities else {} - - def _safe_rel(relation: str) -> str: - return re.sub(r"[^A-Z0-9_]", "_", relation.upper().replace(" ", "_").replace("-", "_")) or "RELATED_TO" - - def _safe_label(label: str) -> str: - """Sanitize a FalkorDB node label to prevent Cypher injection.""" - sanitized = re.sub(r"[^A-Za-z0-9_]", "", label) - return sanitized if sanitized else "Entity" + epoch = _new_epoch() parsed = urlparse(uri if "://" in uri else f"redis://{uri}") # FalkorDB auth is optional. Only send credentials when a password is @@ -138,36 +517,31 @@ def _safe_label(label: str) -> str: password=connect_password, ) graph = db.select_graph(graph_name) - nodes_pushed = 0 - edges_pushed = 0 + _ensure_schema(graph) - for node_id, data in G.nodes(data=True): - props = { - k: v for k, v in data.items() - if isinstance(v, (str, int, float, bool)) and not k.startswith("_") - } - props["id"] = node_id - cid = node_community.get(node_id) - if cid is not None: - props["community"] = cid - ftype = _safe_label(data.get("file_type", "Entity").capitalize()) - graph.query( - f"MERGE (n:{ftype} {{id: $id}}) SET n += $props", - {"id": node_id, "props": props}, + if repo_manifest is not None: + return _push_delta( + G, graph, node_community, epoch, repo_manifest, allow_shrink, shrink_limit ) - nodes_pushed += 1 - for u, v, data in G.edges(data=True): - rel = _safe_rel(data.get("relation", "RELATED_TO")) - props = { - k: v for k, v in data.items() - if isinstance(v, (str, int, float, bool)) and not k.startswith("_") - } - graph.query( - f"MATCH (a {{id: $src}}), (b {{id: $tgt}}) " - f"MERGE (a)-[r:{rel}]->(b) SET r += $props", - {"src": u, "tgt": v, "props": props}, - ) - edges_pushed += 1 + nodes_pushed = _write_nodes(graph, G.nodes(data=True), node_community, epoch) + edges_pushed = _write_edges(graph, G.edges(data=True), epoch) - return {"nodes": nodes_pushed, "edges": edges_pushed} + deleted = deleted_edges = 0 + surplus = 0 + if prune: + deleted, deleted_edges = _converge(graph, epoch, allow_shrink, shrink_limit) + else: + # An add-only push cannot converge, and #3057's whole point is that the + # divergence is SILENT. Report it: the same "not stamped by this push" + # count the prune path would delete tells the caller exactly how far the + # target has drifted, so a one-line notice can replace the silence. + surplus = _count(graph, f"{_STALE_NODES} RETURN count(n)", {"epoch": epoch}) + + return { + "nodes": nodes_pushed, + "edges": edges_pushed, + "deleted": deleted, + "deleted_edges": deleted_edges, + "target_surplus": surplus, + } diff --git a/graphify/global_graph.py b/graphify/global_graph.py index eddd0c92a4..370dad5aaf 100644 --- a/graphify/global_graph.py +++ b/graphify/global_graph.py @@ -175,6 +175,42 @@ def global_remove(repo_tag: str) -> int: return removed +def global_push( + uri: str, + graph_name: str = "graphify", + *, + user: str | None = None, + password: str | None = None, + delta: bool = True, + prune: bool = False, + allow_shrink: bool = False, +) -> dict: + """Push the global graph to a FalkorDB target. + + `export --push` resolves its source from a project's output directory, so + the global graph — the one graph the push/prune contract is actually about — + had no CLI route at all (#3057). + + delta: repo-keyed incremental push (default). Only repos whose manifest + source_hash moved, or whose node count in the target has drifted, are + re-sent; repos the manifest no longer lists are deleted. Set False for a + full push, in which case `prune` decides whether it converges. + """ + from graphify.exporters.graphdb import push_to_falkordb + + G = _load_global_graph() + if G.number_of_nodes() == 0: + raise FileNotFoundError( + "global graph is empty — add a project with `graphify global add` first" + ) + manifest = _load_manifest().get("repos", {}) + return push_to_falkordb( + G, uri=uri, user=user, password=password, graph_name=graph_name, + prune=prune, allow_shrink=allow_shrink, + repo_manifest=manifest if delta else None, + ) + + def global_list() -> dict: """Return the manifest repos dict.""" return _load_manifest().get("repos", {}) diff --git a/tests/test_falkordb_integration.py b/tests/test_falkordb_integration.py index 649e1d2749..6d69e98adf 100644 --- a/tests/test_falkordb_integration.py +++ b/tests/test_falkordb_integration.py @@ -89,3 +89,348 @@ def test_push_to_falkordb_is_idempotent(db): assert node_count == G.number_of_nodes() assert edge_count == G.number_of_edges() + + +def _counts(db, name=GRAPH_NAME): + graph = db.select_graph(name) + n = graph.query("MATCH (n) RETURN count(n)").result_set[0][0] + e = graph.query("MATCH ()-[r]->() RETURN count(r)").result_set[0][0] + return n, e + + +def _entity_counts(db, name=GRAPH_NAME): + """Nodes/edges excluding the :GraphifyPushState bookkeeping nodes.""" + graph = db.select_graph(name) + n = graph.query("MATCH (n:Entity) RETURN count(n)").result_set[0][0] + e = graph.query("MATCH (:Entity)-[r]->(:Entity) RETURN count(r)").result_set[0][0] + return n, e + + +def _fixture_graph(): + from graphify.build import build_from_json + + return build_from_json(json.loads((FIXTURES / "extraction.json").read_text())) + + +def test_pushed_nodes_carry_the_entity_label(db): + """Every node must carry the shared :Entity label. Labelling by file_type + alone leaves edge-endpoint MATCHes unable to use any index (#2258), and any + consumer keyed on a single label sees an empty graph.""" + from graphify.export import push_to_falkordb + + G = _fixture_graph() + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + + graph = db.select_graph(GRAPH_NAME) + total = graph.query("MATCH (n) RETURN count(n)").result_set[0][0] + entities = graph.query("MATCH (n:Entity) RETURN count(n)").result_set[0][0] + assert entities == total > 0 + + +def test_pushed_nodes_keep_their_file_type_label_too(db): + """The file-type label is what existing queries key on, so :Entity is added + alongside it, not instead of it.""" + from graphify.export import push_to_falkordb + + G = _fixture_graph() + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + + graph = db.select_graph(GRAPH_NAME) + labels = { + r[0] for r in graph.query( + "MATCH (n:Entity) UNWIND labels(n) AS l RETURN DISTINCT l" + ).result_set + } + assert "Entity" in labels + assert labels - {"Entity"}, f"no file-type labels survived: {labels}" + + +def test_push_without_prune_still_never_deletes(db): + """The default stays add-only, so the old contract is unchanged.""" + from graphify.export import push_to_falkordb + + G = _fixture_graph() + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + db.select_graph(GRAPH_NAME).query("CREATE (:Entity {id: 'stale-node-1'})") + before, _ = _counts(db) + + result = push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + + assert _counts(db)[0] == before + assert result["deleted"] == 0 + + +def test_prune_removes_what_the_source_no_longer_has(db): + """#3057: with prune a re-push converges on the source instead of keeping + every node the source has since pruned, forever.""" + from graphify.export import push_to_falkordb + + G = _fixture_graph() + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + ground_truth = _counts(db) + + graph = db.select_graph(GRAPH_NAME) + for i in range(5): + graph.query(f"CREATE (:Entity {{id: 'pruned-{i}'}})") + assert _counts(db)[0] == ground_truth[0] + 5 + + result = push_to_falkordb( + G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True, allow_shrink=True + ) + + assert result["deleted"] == 5 + assert _counts(db) == ground_truth + assert graph.query( + "MATCH (n:Entity) WHERE n.id STARTS WITH 'pruned-' RETURN count(n)" + ).result_set[0][0] == 0 + + +def test_prune_removes_an_edge_the_source_dropped(db): + """Convergence has to cover edges, not just nodes. DETACH DELETE takes a + pruned node's edges with it, but an edge dropped from the source whose two + endpoints both survive needs its own sweep — that is the surplus-edge half + of #3057 (+1,151 edges alongside +1,250 nodes in the report).""" + import networkx as nx + from graphify.export import push_to_falkordb + + G = nx.Graph() + for i in range(10): + G.add_node(f"n{i}", label=f"s{i}", file_type="python") + for i in range(9): + G.add_edge(f"n{i}", f"n{i + 1}", relation="calls") + + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True) + assert _counts(db) == (10, 9) + + G.remove_edge("n0", "n1") # both endpoints survive + + result = push_to_falkordb( + G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True, allow_shrink=True + ) + + assert result["deleted"] == 0 + assert result["deleted_edges"] == 1 + assert _counts(db) == (10, 8) + + +def test_prune_is_idempotent_when_nothing_is_stale(db): + from graphify.export import push_to_falkordb + + G = _fixture_graph() + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True) + baseline = _counts(db) + + result = push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True) + + assert result["deleted"] == 0 and result["deleted_edges"] == 0 + assert _counts(db) == baseline + + +def test_prune_refuses_a_mass_deletion_without_allow_shrink(db): + """The #479 rule applied to the push: a source aimed at the wrong graph + looks exactly like a genuine mass removal, so refuse it by default.""" + from graphify.export import push_to_falkordb + + G = _fixture_graph() + graph = db.select_graph(GRAPH_NAME) + for i in range(200): + graph.query(f"CREATE (:Entity {{id: 'someone-elses-{i}'}})") + + with pytest.raises(ValueError, match="safety limit"): + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True) + + assert graph.query( + "MATCH (n:Entity) WHERE n.id STARTS WITH 'someone-elses-' RETURN count(n)" + ).result_set[0][0] == 200 + + +def test_graph_name_isolates_targets(db): + """Before #3057 the CLI could not name a target, so every push landed on + the `graphify` key. Two names must not touch each other.""" + from graphify.export import push_to_falkordb + + other = f"{GRAPH_NAME}_other" + try: + G = _fixture_graph() + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=other) + db.select_graph(other).query("CREATE (:Entity {id: 'only-in-other'})") + + push_to_falkordb( + G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True, allow_shrink=True + ) + + assert db.select_graph(other).query( + "MATCH (n:Entity {id: 'only-in-other'}) RETURN count(n)" + ).result_set[0][0] == 1 + finally: + try: + db.select_graph(other).delete() + except Exception: + pass + + +# -------------------------------------------------------------------------- +# Repo-keyed delta (#3057) +# -------------------------------------------------------------------------- + +@pytest.fixture() +def global_src(): + """A two-repo global graph shaped the way `global add` leaves one.""" + import networkx as nx + + G = nx.Graph() + for tag in ("repoA", "repoB"): + for i in range(20): + G.add_node(f"{tag}::n{i}", label=f"{tag}{i}", file_type="python", repo=tag) + for i in range(19): + G.add_edge(f"{tag}::n{i}", f"{tag}::n{i + 1}", relation="calls") + # cross-repo edge: B depends on a node owned by A + G.add_edge("repoB::n0", "repoA::n0", relation="imports") + return G + + +def _manifest(G, tags=("repoA", "repoB"), hashes=None): + hashes = hashes or {} + out = {} + for t in tags: + n = sum(1 for _, d in G.nodes(data=True) if d.get("repo") == t) + out[t] = {"source_hash": hashes.get(t, f"hash-{t}-v1"), "node_count": n} + return out + + +def test_delta_skips_repos_whose_hash_has_not_moved(db, global_src): + """The delta mirrors global_add's own contract: an unmoved source_hash is a + skip, so a 226-repo graph with one changed repo sends one repo's rows.""" + from graphify.export import push_to_falkordb + + m = _manifest(global_src) + first = push_to_falkordb( + global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m + ) + assert sorted(first["repos_pushed"]) == ["repoA", "repoB"] + + second = push_to_falkordb( + global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m + ) + assert second["repos_pushed"] == [] + assert sorted(second["repos_skipped"]) == ["repoA", "repoB"] + assert second["nodes"] == 0 and second["edges"] == 0 + + +def test_delta_resends_only_the_changed_repo(db, global_src): + from graphify.export import push_to_falkordb + + m = _manifest(global_src) + push_to_falkordb(global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m) + before = _entity_counts(db) + + m["repoA"]["source_hash"] = "hash-repoA-v2" + result = push_to_falkordb( + global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m + ) + + assert result["repos_pushed"] == ["repoA"] + assert result["repos_skipped"] == ["repoB"] + assert result["nodes"] == 20 # repoA only, not all 40 + assert _entity_counts(db) == before # shape unchanged + + +def test_delta_preserves_the_cross_repo_edge_when_a_repo_is_rewritten(db, global_src): + """Pruning repoA drops the B->A edge with A's node. Re-adding only edges + whose BOTH endpoints are in repoA would lose it silently.""" + from graphify.export import push_to_falkordb + + m = _manifest(global_src) + push_to_falkordb(global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m) + graph = db.select_graph(GRAPH_NAME) + q = ("MATCH (:Entity {id:'repoB::n0'})-[r:IMPORTS]-(:Entity {id:'repoA::n0'}) " + "RETURN count(r)") + assert graph.query(q).result_set[0][0] == 1 + + m["repoA"]["source_hash"] = "hash-repoA-v2" + push_to_falkordb(global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m) + + assert graph.query(q).result_set[0][0] == 1, "cross-repo edge lost by the delta" + + +def test_delta_deletes_a_repo_the_manifest_no_longer_lists(db, global_src): + from graphify.export import push_to_falkordb + + m = _manifest(global_src) + push_to_falkordb(global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m) + assert _entity_counts(db)[0] == 40 + + del m["repoB"] + result = push_to_falkordb( + global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, + repo_manifest=m, allow_shrink=True, + ) + + assert result["repos_removed"] == ["repoB"] + assert result["deleted"] == 20 + assert db.select_graph(GRAPH_NAME).query( + "MATCH (n:Entity {repo:'repoB'}) RETURN count(n)" + ).result_set[0][0] == 0 + + +def test_delta_repairs_drift_a_ledger_would_miss(db, global_src): + """A 'what I last pushed' ledger still reads clean after the database is + wiped or a run half-lands. Reading the target's OWN per-repo counts turns + that silent permanent drift into automatic repair.""" + from graphify.export import push_to_falkordb + + m = _manifest(global_src) + push_to_falkordb(global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m) + + db.select_graph(GRAPH_NAME).query( + "MATCH (n:Entity {repo:'repoB'}) WITH n LIMIT 8 DETACH DELETE n" + ) + assert _entity_counts(db)[0] == 32 + + # Hashes have NOT moved — a ledger-only delta would skip both repos here. + result = push_to_falkordb( + global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m + ) + + assert result["repos_pushed"] == ["repoB"] + assert "drift" in result["reasons"]["repoB"] + assert _entity_counts(db)[0] == 40 + + +def test_delta_refuses_a_manifest_aimed_at_the_wrong_database(db, global_src): + """The 1,211,189-node accident: a test manifest pointed at a live graph + looks exactly like a genuine mass removal.""" + from graphify.export import push_to_falkordb + + m = _manifest(global_src) + push_to_falkordb(global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=m) + + foreign = {"someone-elses-repo": {"source_hash": "x", "node_count": 1}} + with pytest.raises(ValueError, match="safety limit"): + push_to_falkordb( + global_src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, repo_manifest=foreign + ) + + assert _entity_counts(db)[0] == 40 # nothing touched + + +def test_add_only_push_reports_the_drift_it_cannot_fix(db): + """#3057's divergence is silent. An add-only push can't converge, but it + can say how far the target has drifted (@Azeem1985's review request).""" + from graphify.export import push_to_falkordb + + G = _fixture_graph() + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + db.select_graph(GRAPH_NAME).query("CREATE (:Entity {id: 'left-behind'})") + + result = push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + assert result["deleted"] == 0 # still add-only + assert result["target_surplus"] == 1 # but no longer silent about it + + converged = push_to_falkordb( + G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True, allow_shrink=True + ) + assert converged["deleted"] == 1 + after = push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + assert after["target_surplus"] == 0