From dc5183c58360043ee8f4bdb6d856d49f9180af05 Mon Sep 17 00:00:00 2001 From: Gal Shubeli Date: Tue, 25 Aug 2026 10:36:55 +0300 Subject: [PATCH 1/2] fix(falkordb): make the graph-DB push converge, and add a repo-keyed delta (#3057) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `push_to_falkordb` was MERGE-only, so once `global add` pruned a repo out of the global graph a full re-push left every one of those nodes in the target forever: the database diverged from the source and never converged back. The same code path also wrote one query per node and one per edge, matched edge endpoints without a label (so no index could serve them), and labelled nodes by file_type only. That last part was its own bug: graphify's own store matches on `:Entity`, so a graph produced by `export falkordb --push` was unreadable by `graphify query`/`serve` — measured at 4000 nodes pushed, 0 read back. The FalkorDB writer now goes through GraphStore, the same batched UNWIND writer graphify uses for its own graphs, which supplies the `:Entity` label, the `n.id` index and the batching for free and deletes the duplicated writer. Pre-existing nodes are labelled on the way in so the MERGE switch does not duplicate them. Convergence is opt-in via --prune, covering 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. Opt-in rather than default because --graph-name never existed, so existing users may have several projects merged into the one `graphify` key and a default prune would delete all but the last. Delta mode (repo_manifest=) mirrors global_add's own contract: the repo is the unit of change, keyed on the manifest's per-repo source_hash. Only repos whose hash moved are re-sent. The "what I last pushed" state lives in the target as :GraphifyPushState nodes and is cross-checked against the target's own per-repo counts, so a wipe or a half-landed run is repaired instead of silently skipped. Cross-repo edges are restored by selecting every edge incident to a re-pushed repo, not only edges internal to it. Deletes are guarded at 20% of the target unless --allow-shrink, the same "refuse to SILENTLY drop nodes" rule as the #479 build guard, and paged with the LIMIT inside a WITH — FalkorDB's LIMIT does not short-circuit an eager DELETE, so `... DETACH DELETE n LIMIT $page` would delete the whole label. Also adds `graphify global push`: the global graph is a named FalkorDB graph with no output directory, so `export --push` (which resolves its source from a directory's falkordb.json pointer) could never reach the one graph this whole contract is about. Measured against a local FalkorDB, 20 repos / 20k nodes: full push, nothing changed 1.46s 39999 rows delta, nothing changed 0.01s 0 rows delta, 1 of 20 repos changed 0.33s 2001 rows (5.0%) Delta output is identical to a from-scratch load on node and edge counts. The Neo4j writer is deliberately untouched and still has the same defects; its fix needs a Neo4j instance to verify. The new flags are refused on that subcommand rather than silently ignored. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 7 + graphify/cli.py | 126 +++++++- graphify/exporters/graphdb.py | 501 ++++++++++++++++++++++++----- graphify/global_graph.py | 37 +++ tests/test_falkordb_integration.py | 341 ++++++++++++++++++++ 5 files changed, 931 insertions(+), 81 deletions(-) diff --git a/README.md b/README.md index e171cd0a1a..14e4d6d14d 100644 --- a/README.md +++ b/README.md @@ -747,10 +747,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 a FalkorDB target (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 6922746e50..451863dd53 100644 --- a/graphify/cli.py +++ b/graphify/cli.py @@ -2046,8 +2046,11 @@ def dispatch_command(cmd: str) -> None: 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 @@ -2081,6 +2084,12 @@ def dispatch_command(cmd: str) -> None: 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] @@ -2136,6 +2145,12 @@ def dispatch_command(cmd: str) -> None: 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": @@ -2288,6 +2303,24 @@ def dispatch_command(cmd: str) -> None: 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: @@ -2304,9 +2337,21 @@ def dispatch_command(cmd: str) -> None: 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}") else: from graphify.export import to_cypher as _to_cypher _to_cypher(G, str(out_dir / "cypher.txt")) @@ -2339,6 +2384,7 @@ def dispatch_command(cmd: str) -> None: 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 ] @@ -2366,6 +2412,76 @@ def dispatch_command(cmd: str) -> None: 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: @@ -2386,7 +2502,7 @@ def dispatch_command(cmd: str) -> None: 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 5a4a0449cb..e6346308ef 100644 --- a/graphify/exporters/graphdb.py +++ b/graphify/exporters/graphdb.py @@ -1,8 +1,287 @@ -"""graphdb — moved verbatim from graphify/export.py.""" +"""graphdb — direct push of a graphify graph into Neo4j / FalkorDB. + +The FalkorDB writer goes through ``GraphStore`` (``graphify/store.py``), the +same batched writer graphify uses for its own graphs, so a pushed graph gets: + + - the shared ``:Entity`` label plus its file-type label (``:Entity:Python``), + which is what makes a pushed graph readable by ``graphify query`` / ``serve`` + — they match on ``:Entity`` and saw nothing at all in a pushed graph before; + - the ``n.id`` index, so edge-endpoint MATCHes resolve through an index + instead of scanning every node once per edge (#2258); + - batched ``UNWIND`` writes rather than one round trip per node and 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. ``prune=True`` makes the push *converge*: every node this push did not +write is deleted — nodes *and* edges, so an edge dropped between two surviving +endpoints goes too — and the target ends up an exact mirror of the source. Because +that 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 deliberately untouched: it has the same per-row and +unindexed-MATCH problems, but neither the fix nor a convergence mode can be +tested here, so it keeps its old add-only behavior and the new flags are +refused rather than silently ignored on that path. +""" from __future__ import annotations -from graphify.analyze import _node_community_map import re +import time + +from graphify.analyze import _node_community_map + +# Rows per UNWIND batch. Matches GraphStore._BATCH so both writers behave the +# same way against the same server. +_BATCH = 1000 +# Nodes deleted per convergence page. Paged so a large prune is not one +# unbounded transaction. +_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 a push writes; convergence deletes whatever lacks the +# current value. A plain (non-underscore) name so GraphStore._scalar_props keeps +# it — underscore-prefixed properties are dropped on the way into the store. +_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 _new_epoch() -> int: + """Identifier for one push. Millisecond clock: 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 only one chunk at a time. + + The source is a streaming GraphStore view on this branch, so the push must + never materialize the whole graph — that is the OOM the reporter of #3057 + hit on a 1.87GB graph. + """ + chunk = [] + for item in iterable: + chunk.append(item) + if len(chunk) >= size: + yield chunk + chunk = [] + if chunk: + yield chunk + + +def _stamped_nodes(G, node_community: dict, epoch: int): + """Yield (node_id, attrs) with community and the push epoch merged in.""" + for node_id, data in G.nodes(data=True): + attrs = dict(data) + cid = node_community.get(node_id) + if cid is not None: + attrs["community"] = cid + attrs[_EPOCH_PROP] = epoch + yield (node_id, attrs) + + +def _stamped_edges(G, epoch: int): + """Yield (u, v, attrs) with the push epoch merged in, so the convergence + sweep can tell this push's edges from ones the source has since dropped.""" + for u, v, data in G.edges(data=True): + attrs = dict(data) + attrs[_EPOCH_PROP] = epoch + yield (u, v, attrs) + + +# A node this push did not write, and an edge this push did not write. Edges +# need their own sweep: DETACH DELETE on stale nodes takes their edges with +# them, but an edge dropped from the source whose two endpoints both survive +# would otherwise linger forever — the surplus-edge half of #3057. +_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 _delete_paged(run, count, 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 a page. The LIMIT has to sit in a WITH that precedes the DELETE, as + below. + """ + verb = "DETACH DELETE" if var == "n" else "DELETE" + page = f"{stale_match} WITH {var} LIMIT {_DELETE_PAGE} {verb} {var}" + remaining = expected + while remaining > 0: + run(page, params) + after = count(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 + + +# --------------------------------------------------------------------------- +# 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 — only repos whose hash +# moved are re-sent, so a 226-repo global graph with one changed repo sends one +# repo's rows rather than all of them (#3057). +# +# The "what did I last push" state lives in the TARGET database, not in a local +# ledger, as a :GraphifyPushState node per repo. 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(count_rows) -> dict[str, dict]: + """Per-repo {source_hash, node_count} the target believes it holds.""" + rows = count_rows( + f"MATCH (s:{_STATE_LABEL}) RETURN s.repo, s.source_hash, s.node_count", {} + ) + 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(count_rows) -> dict[str, int]: + """The target's OWN per-repo node counts — one indexed aggregate. This is + the check a ledger cannot do: it sees a wipe or a half-landed run.""" + rows = count_rows( + "MATCH (n:Entity) WHERE n.repo IS NOT NULL RETURN n.repo, count(n)", {} + ) + return {r[0]: int(r[1]) for r in rows if r[0]} + + +def _repo_nodes(G, tag: str): + """Stream one repo's nodes from the source global graph.""" + from graphify.store import _META_KEY + + for r in G._stream( + "MATCH (n:Entity {repo:$t}) RETURN n.id, properties(n)", "id(n)", {"t": tag} + ): + attrs = dict(r[1]) + attrs.pop(_META_KEY, None) + yield (r[0], attrs) + + +def _repo_edges(G, tag: str): + """Stream every edge incident to one repo's nodes — in EITHER direction. + + Not just edges whose both endpoints are in the repo: `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 own edges would not bring it + back, and the target would quietly lose cross-repo connectivity on every + delta. Selecting on `a.repo = t OR b.repo = t` restores it, and repo B's + nodes are never touched. + """ + for r in G._stream( + "MATCH (a:Entity)-[r]->(b:Entity) WHERE a.repo = $t OR b.repo = $t " + "RETURN a.id, b.id, properties(r)", "id(r)", {"t": tag}, + ): + yield (r[0], r[1], dict(r[2])) + + +def _prune_repo_paged(run, count, tag: str) -> int: + """Delete one repo's nodes, paged. Returns the count removed. + + Not GraphStore.prune_repo: that measures with two whole-graph + number_of_nodes() calls per repo, which on a 226-repo global graph is 452 + full scans. One scoped count over the indexed `repo` property does the same + job per repo. + """ + scoped = "MATCH (n:Entity {repo: $t})" + params = {"t": tag} + n = count(f"{scoped} RETURN count(n)", params) + if n: + _delete_paged(run, count, scoped, "n", params, n) + return n + + +def _plan_delta(manifest_repos: dict, state: dict, live_counts: dict) -> tuple[list, list, dict]: + """Decide which repos to re-push and which to delete. + + Returns (changed, removed, reasons). A repo is re-pushed when its manifest + hash moved, when the target never recorded it, or when the target's live + node count disagrees with what the manifest says it should hold (drift + repair). A repo the manifest no longer lists is deleted. + """ + changed, reasons = [], {} + for tag, info in manifest_repos.items(): + want_hash = info.get("source_hash") + want_count = int(info.get("node_count") or 0) + known = state.get(tag) + live = live_counts.get(tag, 0) + if known is None: + changed.append(tag); reasons[tag] = "not present in target" + elif known.get("source_hash") != want_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 = [t for t in list(state) + list(live_counts) if t not in manifest_repos] + # de-dup while keeping order + removed = list(dict.fromkeys(removed)) + return changed, removed, reasons + + +def _converge(run, count, epoch: int, allow_shrink: bool, shrink_limit: float) -> tuple[int, int]: + """Delete every node and edge this push did not write. + + Returns (nodes_deleted, edges_deleted). `run(cypher, params)` executes; + `count(cypher, params)` returns an int. + """ + params = {"epoch": epoch} + stale_n = count(f"{_STALE_NODES} RETURN count(n)", params) + stale_e = count(f"{_STALE_EDGES} RETURN count(r)", params) + if stale_n <= 0 and stale_e <= 0: + return 0, 0 + + total_n = count("MATCH (n:Entity) RETURN count(n)", {}) + total_e = count("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 + # that follows has less to do and its count is already settled. + nodes_deleted = _delete_paged(run, count, _STALE_NODES, "n", params, stale_n) if stale_n else 0 + remaining_e = count(f"{_STALE_EDGES} RETURN count(r)", params) + edges_deleted = ( + _delete_paged(run, count, _STALE_EDGES, "r", params, remaining_e) if remaining_e else 0 + ) + return nodes_deleted, edges_deleted def push_to_neo4j( @@ -76,97 +355,167 @@ def _safe_label(label: str) -> str: driver.close() return {"nodes": nodes_pushed, "edges": edges_pushed} + +def _push_delta( + G, target, node_community: dict, epoch: int, manifest_repos: dict, + run, count, rows, allow_shrink: bool, shrink_limit: float, +) -> dict: + """Repo-keyed delta push. See the module's delta section for the contract.""" + state = _read_push_state(rows) + live_counts = _target_repo_counts(rows) + 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 to the limit 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("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 " + f"{total_nodes} nodes ({doomed / total_nodes:.0%}) in the target " + f"graph, over the {shrink_limit:.0%} safety limit. That usually " + f"means this manifest does not belong to this database — check " + f"--graph-name. Pass --allow-shrink if it is intended. Nothing was " + f"changed." + ) + + nodes_pushed = edges_pushed = deleted = 0 + for tag in changed: + deleted += _prune_repo_paged(run, count, tag) + for chunk in _chunked(_repo_nodes(G, tag), _BATCH): + stamped = [] + for nid, attrs in chunk: + cid = node_community.get(nid) + if cid is not None: + attrs["community"] = cid + attrs[_EPOCH_PROP] = epoch + stamped.append((nid, attrs)) + target.add_nodes_from(stamped) + nodes_pushed += len(stamped) + for chunk in _chunked(_repo_edges(G, tag), _BATCH): + stamped = [(u, v, {**a, _EPOCH_PROP: epoch}) for u, v, a in chunk] + target.add_edges_from(stamped) + edges_pushed += len(stamped) + info = manifest_repos.get(tag, {}) + run( + 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(run, count, tag) + run(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, + G, uri: str, user: str | None = None, 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. + """Push graph directly to a running FalkorDB instance. 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: - - 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" - and a bare "localhost:6379" are all equivalent (default port 6379). - - a named graph is selected via db.select_graph(graph_name) (default - "graphify"); FalkorDB keys each graph by name in the same instance. - - queries run via graph.query(cypher, params) - there is no session object. - - auth is optional (FalkorDB runs without credentials by default), so user - 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. - """ - try: - from falkordb import FalkorDB - except ImportError as e: - raise ImportError( - "falkordb SDK not installed. Run: pip install falkordb" - ) from e + Writes through ``GraphStore``, the same batched ``UNWIND`` writer graphify + uses for its own graphs, so the pushed graph gets the ``:Entity`` label, the + ``n.id`` index, and a schema ``graphify query``/``serve`` can read back. + Only the host/port are read from the URI, so the scheme is informational — + "falkordb://localhost:6379", "redis://localhost:6379" and a bare + "localhost:6379" are all equivalent (default port 6379). Auth is optional + (FalkorDB runs without credentials by default), so user and password may be + None; credentials embedded in the URI take precedence. - from urllib.parse import urlparse + 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 — always 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. + """ + from graphify.store import GraphStore node_community = _node_community_map(communities) if communities else {} + epoch = _new_epoch() - def _safe_rel(relation: str) -> str: - return re.sub(r"[^A-Z0-9_]", "_", relation.upper().replace(" ", "_").replace("-", "_")) or "RELATED_TO" + # GraphStore.__init__ creates the :Entity(id) index and applies the same + # optional-auth rules as the old inline connection code. + target = GraphStore(graph_name=graph_name, uri=uri, user=user, password=password) - 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" + # See the Neo4j path: label pre-schema nodes so MERGE matches them instead + # of creating a duplicate beside each one. + target._g.query("MATCH (n) WHERE n.id IS NOT NULL AND NOT n:Entity SET n:Entity") - parsed = urlparse(uri if "://" in uri else f"redis://{uri}") - # FalkorDB auth is optional. Only send credentials when a password is - # provided; otherwise connect anonymously and ignore any bolt-style default - # username (e.g. Neo4j's "neo4j"), which FalkorDB rejects as an unknown ACL - # user. Credentials embedded in the URI take precedence over the args. - connect_user = parsed.username or (user if password else None) - connect_password = parsed.password or (password or None) - db = FalkorDB( - host=parsed.hostname or "localhost", - port=parsed.port or 6379, - username=connect_user, - password=connect_password, - ) - graph = db.select_graph(graph_name) - nodes_pushed = 0 - edges_pushed = 0 + def run(cypher, params): + target._g.query(cypher, params) - 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}, - ) - nodes_pushed += 1 + def count(cypher, params): + return int(target._g.query(cypher, params).result_set[0][0]) - 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}, + def rows(cypher, params): + return target._g.query(cypher, params).result_set or [] + + if repo_manifest is not None: + return _push_delta( + G, target, node_community, epoch, repo_manifest, + run, count, rows, allow_shrink, shrink_limit, ) - edges_pushed += 1 - return {"nodes": nodes_pushed, "edges": edges_pushed} + nodes_pushed = 0 + edges_pushed = 0 + for chunk in _chunked(_stamped_nodes(G, node_community, epoch), _BATCH): + target.add_nodes_from(chunk) + nodes_pushed += len(chunk) + for chunk in _chunked(_stamped_edges(G, epoch), _BATCH): + target.add_edges_from(chunk) + edges_pushed += len(chunk) + + deleted = deleted_edges = 0 + if prune: + deleted, deleted_edges = _converge(run, count, epoch, allow_shrink, shrink_limit) + + return { + "nodes": nodes_pushed, + "edges": edges_pushed, + "deleted": deleted, + "deleted_edges": deleted_edges, + } diff --git a/graphify/global_graph.py b/graphify/global_graph.py index ab9e419e74..ba5ece5646 100644 --- a/graphify/global_graph.py +++ b/graphify/global_graph.py @@ -158,6 +158,43 @@ 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 Neo4j-compatible / FalkorDB target. + + The global graph is a named FalkorDB graph with no output directory, so + `export --push` (which resolves its source from a directory's falkordb.json + pointer) could never reach it — the graph the whole push/prune contract is + 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..140e1541bd 100644 --- a/tests/test_falkordb_integration.py +++ b/tests/test_falkordb_integration.py @@ -89,3 +89,344 @@ 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 _build(extraction_overrides=None): + from graphify.build import build_from_json + + extraction = json.loads((FIXTURES / "extraction.json").read_text()) + if extraction_overrides: + extraction = extraction_overrides(extraction) + return build_from_json(extraction, graph_name="graphify_push_src", uri=f"{HOST}:{PORT}") + + +def test_pushed_nodes_carry_the_entity_label(db): + """A pushed graph must be readable by graphify's own store, which matches + on :Entity. Before #3057 the exporter labelled by file_type only, so every + `graphify query`/`serve` against a pushed graph came back empty.""" + from graphify.export import push_to_falkordb + + G = _build() + 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_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 = _build() + 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) + + after, _ = _counts(db) + assert after == 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 + accumulating the pruned nodes forever.""" + from graphify.export import push_to_falkordb + + G = _build() + push_to_falkordb(G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME) + ground_truth = _counts(db) + + # Simulate what `global add` leaves behind: nodes the source has dropped. + 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 + # Identical to a from-scratch load on both counts. + assert _counts(db) == ground_truth + survivors = graph.query( + "MATCH (n:Entity) WHERE n.id STARTS WITH 'pruned-' RETURN count(n)" + ).result_set[0][0] + assert survivors == 0 + + +def test_prune_is_idempotent_and_a_noop_when_nothing_is_stale(db): + from graphify.export import push_to_falkordb + + G = _build() + 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 + assert _counts(db) == baseline + + +def test_prune_refuses_a_mass_deletion_without_allow_shrink(db): + """The #479 rule applied to the push: a manifest 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 = _build() + graph = db.select_graph(GRAPH_NAME) + # A target dominated by nodes this push does not write. + for i in range(200): + graph.query(f"CREATE (:Entity {{id: 'someone-elses-{i}'}})") + before, _ = _counts(db) + + with pytest.raises(ValueError, match="safety limit"): + push_to_falkordb( + G, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True + ) + + # Refused means refused: nothing deleted. + survivors = graph.query( + "MATCH (n:Entity) WHERE n.id STARTS WITH 'someone-elses-' RETURN count(n)" + ).result_set[0][0] + assert survivors == 200 + assert _counts(db)[0] >= before + + +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 = _build() + 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 + ) + + still_there = db.select_graph(other).query( + "MATCH (n:Entity {id: 'only-in-other'}) RETURN count(n)" + ).result_set[0][0] + assert still_there == 1 + finally: + try: + db.select_graph(other).delete() + except Exception: + pass + + +def test_prune_removes_an_edge_the_source_dropped(db): + """Convergence has to cover edges, not just nodes. DETACH DELETE takes the + edges of a pruned node 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 (the report measured +1,151 edges alongside +1,250 nodes).""" + from graphify.export import push_to_falkordb + from graphify.store import GraphStore + + src = GraphStore(graph_name="graphify_push_src", uri=f"{HOST}:{PORT}") + src.clear() + src.add_nodes_from([(f"n{i}", {"label": f"s{i}", "file_type": "python"}) for i in range(10)]) + src.add_edges_from([(f"n{i}", f"n{i + 1}", {"relation": "calls"}) for i in range(9)]) + + push_to_falkordb(src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True) + assert _counts(db) == (10, 9) + + src.remove_edges([("n0", "n1")]) # both endpoints survive + + result = push_to_falkordb( + src, uri=f"{HOST}:{PORT}", graph_name=GRAPH_NAME, prune=True, allow_shrink=True + ) + + assert result["deleted"] == 0 # no node went away + assert result["deleted_edges"] == 1 # but the dropped edge did + assert _counts(db) == (10, 8) + + +# -------------------------------------------------------------------------- +# Repo-keyed delta (#3057) +# -------------------------------------------------------------------------- + +SRC_GLOBAL = "graphify_global_src" + + +@pytest.fixture() +def global_src(): + """A two-repo global graph shaped the way `global add` leaves one.""" + from graphify.store import GraphStore + + G = GraphStore(graph_name=SRC_GLOBAL, uri=f"{HOST}:{PORT}") + G.clear() + G.add_nodes_from( + [(f"repoA::n{i}", {"label": f"a{i}", "file_type": "python", "repo": "repoA"}) + for i in range(20)] + + [(f"repoB::n{i}", {"label": f"b{i}", "file_type": "python", "repo": "repoB"}) + for i in range(20)] + ) + G.add_edges_from( + [(f"repoA::n{i}", f"repoA::n{i + 1}", {"relation": "calls"}) for i in range(19)] + + [(f"repoB::n{i}", f"repoB::n{i + 1}", {"relation": "calls"}) for i in range(19)] + # cross-repo edge: B depends on a node owned by A + + [("repoB::n0", "repoA::n0", {"relation": "imports"})] + ) + yield G + try: + G.clear() + except Exception: + pass + + +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 _manifest(G, tags=("repoA", "repoB"), hashes=None): + hashes = hashes or {} + out = {} + for t in tags: + n = int(G._g.query( + "MATCH (n:Entity {repo:$t}) RETURN count(n)", {"t": t} + ).result_set[0][0]) + 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(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" # repoA changed, repoB did not + 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 # and the graph is unchanged in shape + + +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 — the delta has to + restore every edge incident to the repo.""" + 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): + """#3057's core defect, at the repo level: what the source pruned must go.""" + 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 + graph = db.select_graph(GRAPH_NAME) + assert graph.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) + + # Half-landed run: someone/something removed part of repoB from the target. + 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 # repaired + + +def test_delta_refuses_a_manifest_aimed_at_the_wrong_database(db, global_src): + """The 1,211,189-node accident: a 40-repo 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 From c1d0a7e48f8fea4adca039a6cb13e80800fc0166 Mon Sep 17 00:00:00 2001 From: Gal Shubeli Date: Tue, 25 Aug 2026 11:23:10 +0300 Subject: [PATCH 2/2] feat(falkordb): report add-only push drift instead of leaving it silent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up from @Azeem1985 on #3057. Keeping --prune opt-in is right, but that leaves the reported failure mode — the target quietly holding nodes the source pruned — still silent on the default path. An add-only push now counts the nodes it did not stamp (the same query the prune path deletes by) and returns it as `target_surplus`; the CLI prints one line naming the count and the flag that fixes it. Zero risk: nothing is deleted, and a converged target reports 0. Co-Authored-By: Claude Opus 5 (1M context) --- graphify/cli.py | 6 ++++++ graphify/exporters/graphdb.py | 8 ++++++++ tests/test_falkordb_integration.py | 23 +++++++++++++++++++++++ 3 files changed, 37 insertions(+) diff --git a/graphify/cli.py b/graphify/cli.py index 451863dd53..e78b15b66f 100644 --- a/graphify/cli.py +++ b/graphify/cli.py @@ -2352,6 +2352,12 @@ def dispatch_command(cmd: str) -> None: 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")) diff --git a/graphify/exporters/graphdb.py b/graphify/exporters/graphdb.py index e6346308ef..527a0779a9 100644 --- a/graphify/exporters/graphdb.py +++ b/graphify/exporters/graphdb.py @@ -510,12 +510,20 @@ def rows(cypher, params): edges_pushed += len(chunk) deleted = deleted_edges = 0 + surplus = 0 if prune: deleted, deleted_edges = _converge(run, count, 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(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/tests/test_falkordb_integration.py b/tests/test_falkordb_integration.py index 140e1541bd..7034e50622 100644 --- a/tests/test_falkordb_integration.py +++ b/tests/test_falkordb_integration.py @@ -430,3 +430,26 @@ def test_delta_refuses_a_manifest_aimed_at_the_wrong_database(db, global_src): ) 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 = _build() + 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 + + # A converged target reports no surplus. + 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