feat(graph): wave-2 engine improvements (W2-1..W2-14) + P0 v3 graph-durability fix, GCP-hardened

CHANGELOG.md

@@ -327,7 +327,50 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   HNSW bridge (heap-owned; mmap-backed sections count 0 per the
   `RawF16Store` precedent), keeping the elastic memory budget honest.
 
-### Changed — consolidated dependency bumps (PR #TBD)
+### Changed — graph engine wave 2: durability, Cypher coverage, hardening (PR #TBD)
+
+- **Durability (P0, found by GCP production-hardening soak):** graph data now
+  survives kill -9 under the DEFAULT disk-offload (WAL v3) configuration.
+  Two stacked pre-existing bugs erased graphs on crash: (A) v3 recovery
+  collected graph WAL commands into a throwaway replay engine — graph replay
+  was a complete no-op in v3 mode (a dedicated `replay_graph_wal_v3` boot
+  pass now applies them); (B) the checkpoint advanced the WAL replay floor
+  and recycled segments without ever snapshotting the graph store —
+  `save_graph_store` ran only on graceful shutdown and never persisted the
+  mutable tier (checkpoint finalize now freezes each graph's write buffer,
+  persists segments + a `snapshot_lsn` replay floor, and ABORTS the
+  checkpoint if the graph snapshot fails, keeping the old floor). The prior
+  crash suite ran exclusively with `--disk-offload disable`; new G4/G5
+  scenarios cover the default config with and without a checkpoint crossing.
+- **Durability:** stable external node/edge ids across WAL replay (handles
+  handed to clients before a crash resolve to the same rows after recovery);
+  Cypher `SET` property/label writes now emit WAL records
+  (`GRAPH.SETPROP`/`GRAPH.SETLABEL` — previously a documented durability gap:
+  SET mutations silently vanished on kill -9); new kill-9 crash-recovery
+  suite (`tests/crash_recovery_graph_durability.rs`: mutable tier, frozen
+  64K-edge tier rebuilt through replay-time freeze, double-crash idempotence).
+- **Cypher coverage:** aggregations `count`/`sum`/`avg`/`min`/`max`/`collect`
+  with implicit grouping, `DISTINCT`, and count-over-zero-rows semantics;
+  `OPTIONAL MATCH` null-pads unmatched expansions (previously compiled
+  silently to inner MATCH; unsupported shapes now reject loudly);
+  `WITH` rebinds the pipeline mid-query (aggregate + `WHERE`-as-HAVING,
+  `ORDER BY`/`SKIP`/`LIMIT` between WITH and RETURN; previously every clause
+  after WITH ran on an empty row stream). `WITH *` rejects loudly.
+- **Copy-up writes:** `SET`/`DELETE`/`MERGE` on frozen rows copy the row up
+  into the write buffer instead of silently missing the frozen tier.
+- **Query performance:** IndexScan range predicates (`WHERE n.p > x` prunes
+  via per-segment B-tree-ish numeric index, superset semantics + residual
+  filter); write-side plan cache (repeated write shapes skip parse+compile);
+  row-BFS fast path engages on multi-segment fully-frozen graphs (~4× vs
+  reader fallback, criterion-pinned); `Value::String` holds `Bytes` for a
+  zero-copy reply path; HYB-02/HYB-04 use the HNSW bridge with off-thread
+  bridge builds.
+- **Operability:** traversal timeout is configurable — `--graph-timeout-ms`
+  server default plus per-query `GRAPH.QUERY ... TIMEOUT <ms>` (RedisGraph
+  parity, 0 = unlimited).
+- **Dead code:** cross-shard traverse scaffolding deleted (532 lines, no
+  sender existed); the single-shard-per-graph sharding model is now
+  documented in `src/graph/mod.rs`.
 
 - Cargo: `ringbuf` 0.4.8 → 0.5.0 and `metrics-exporter-prometheus` 0.16.2 →
   0.18.3 (semver-major; both compile and test green with no code changes —

benches/graph_traversal.rs

@@ -139,5 +139,119 @@ fn bench_parallel_vs_sequential_bfs(c: &mut Criterion) {
     group.finish();
 }
 
-criterion_group!(graph_traversal, bench_parallel_vs_sequential_bfs);
+// ---------------------------------------------------------------------------
+// Frozen-tier BFS benchmarks (W2-10)
+//
+// The CSR row-space fast path (`row_bfs`) only engages on a fully-frozen
+// graph — the mutable-only fixtures above never exercise it. These fixtures
+// freeze the SAME 10K/50 graph so the three production shapes are directly
+// comparable:
+//   frozen_single — one CSR segment: full row-BFS (parallel levels + Beamer)
+//   frozen_multi2 — the graph frozen twice (identical clone segments, the
+//                   W2-2 copy-up aftermath shape): W2-6 multi-segment row
+//                   path with worst-case key-level boundary sync (every node
+//                   resident in both segments)
+//   mixed_tier    — same segment + a non-empty mutable tail: the gate
+//                   declines and BFS falls back to the SegmentMergeReader
+//                   path (what row-BFS saves)
+// ---------------------------------------------------------------------------
+
+fn bench_frozen_tier_bfs(c: &mut Criterion) {
+    use moon::graph::csr::CsrSegment;
+    use slotmap::Key;
+
+    let mut group = c.benchmark_group("frozen_bfs");
+
+    const N: usize = 10_000;
+    const DEGREE: usize = 50;
+
+    // Freeze the standard fixture once → single segment.
+    let (mut g, nodes) = build_memgraph(N, DEGREE);
+    let seed = nodes[500];
+    let seg_old = Arc::new(CsrStorage::from(
+        CsrSegment::from_frozen(g.freeze().expect("freeze"), 3).expect("csr"),
+    ));
+
+    // Re-materialize every node at its ORIGINAL key and re-add the same LCG
+    // edges, then freeze again → an identical clone segment (the shape a
+    // W2-2 copy-up + re-freeze leaves behind). Every node is resident in
+    // BOTH segments — worst-case boundary sync for the multi-segment path.
+    g.thaw();
+    for &nk in &nodes {
+        g.add_node_with_id(nk.data().as_ffi(), smallvec![0], empty_props(), None, 4);
+    }
+    let mut rng_state: u32 = 42;
+    for i in 0..N {
+        for _ in 0..DEGREE {
+            rng_state = rng_state.wrapping_mul(1664525).wrapping_add(1013904223);
+            let target = (rng_state as usize) % N;
+            if target == i {
+                continue;
+            }
+            let _ = g.add_edge(nodes[i], nodes[target], 1, 1.0, None, 5);
+        }
+    }
+    let seg_new = Arc::new(CsrStorage::from(
+        CsrSegment::from_frozen(g.freeze().expect("freeze"), 6).expect("csr"),
+    ));
+
+    let single: Vec<Arc<CsrStorage>> = vec![seg_old.clone()];
+    let multi: Vec<Arc<CsrStorage>> = vec![seg_new, seg_old.clone()];
+
+    // Mixed tier: one live node in the write buffer flips the gate off and
+    // forces the reader fallback over the same frozen data.
+    let mut tail = MemGraph::new(usize::MAX >> 1);
+    let _ = tail.add_node(smallvec![0], empty_props(), None, 7);
+
+    let run = |mg: Option<&MemGraph>, segs: &[Arc<CsrStorage>]| {
+        let reader = SegmentMergeReader::new(mg, segs, Direction::Outgoing, u64::MAX - 1, None);
+        BoundedBfs::new(3).execute(&reader, seed)
+    };
+
+    // Correctness: all three shapes must visit the same node set (the clone
+    // segment adds no new reachability).
+    {
+        let key_set = |r: &moon::graph::traversal::BfsResult| {
+            let mut v: Vec<u64> = r.visited.iter().map(|e| e.0.data().as_ffi()).collect();
+            v.sort_unstable();
+            v
+        };
+        let s = run(None, &single).expect("single ok");
+        let m = run(None, &multi).expect("multi ok");
+        let x = run(Some(&tail), &single).expect("mixed ok");
+        assert_eq!(key_set(&s), key_set(&m), "multi must match single");
+        assert_eq!(key_set(&s), key_set(&x), "reader fallback must match");
+        eprintln!(
+            "frozen BFS precheck: 10K nodes, degree 50, depth 3 -> {} nodes visited",
+            s.visited.len()
+        );
+    }
+
+    group.bench_function("row_bfs_frozen_single_10k_depth3", |b| {
+        b.iter(|| black_box(run(None, black_box(&single))))
+    });
+
+    group.bench_function("row_bfs_frozen_multi2_10k_depth3", |b| {
+        b.iter(|| black_box(run(None, black_box(&multi))))
+    });
+
+    group.bench_function("reader_bfs_mixed_tier_10k_depth3", |b| {
+        b.iter(|| black_box(run(Some(black_box(&tail)), black_box(&single))))
+    });
+
+    // Apples-to-apples W2-6 baseline: the reader path over the SAME two
+    // segments (multi2 doubles the edge probes vs `single`, so only this
+    // pairing isolates the multi-segment row path's win).
+    group.bench_function("reader_bfs_frozen_multi2_10k_depth3", |b| {
+        b.iter(|| black_box(run(Some(black_box(&tail)), black_box(&multi))))
+    });
+
+    group.finish();
+}
+
+criterion_group!(
+    graph_traversal,
+    bench_parallel_vs_sequential_bfs,
+    bench_frozen_tier_bfs
+);
 criterion_main!(graph_traversal);