From 034d98b62505f18df6f44bcc7dcc3e575fc6e34b Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Sun, 16 Aug 2026 13:38:50 -0600 Subject: [PATCH 1/6] test(scale): define bounded billion-edge scale ladder and first-fail contract (#736) Add the M5 (#735) root gate: a versioned, bounded-memory Graph500-parameter (ef=16 undirected Kronecker) scale ladder on the public facade. - scale_g500_ladder.v1.json: versioned profile (schema graphforge-billion-edge-ladder/1) pinning rungs S10..S26, seed, initiator, self-loop/duplicate policy, 128 GiB / 1 TiB / 24 h host envelope, metrics, and exact invocation. - scale_g500_ladder.rs: bounded generator (external sort + spill + k-way merge) with memory independent of total edge count; reconciles raw_attempts == live_unique_edges + self_loops_rejected + duplicates_rejected; first-fail envelope harness that stops at the first RSS/disk/time breach; streaming ingest through GraphForge::publish_bulk_*; per-rung evidence JSON. - SCALE-10 rung runs in normal CI and exercises the spill path; SCALE-20..26 are #[ignore], opt-in via `make bench-g500-ladder`. - Docs: perf-g500-ladder runbook + registration in scale-evaluation.md. Not Official-track, not TEPS; does not certify one billion live edges (#745). Co-Authored-By: Claude Opus 4.8 --- Makefile | 6 +- .../tests/fixtures/scale_g500_ladder.v1.json | 58 + .../graphforge-api/tests/scale_g500_ladder.rs | 1147 +++++++++++++++++ docs/development/perf-g500-ladder.md | 126 ++ docs/reference/scale-evaluation.md | 17 +- 5 files changed, 1352 insertions(+), 2 deletions(-) create mode 100644 crates/graphforge-api/tests/fixtures/scale_g500_ladder.v1.json create mode 100644 crates/graphforge-api/tests/scale_g500_ladder.rs create mode 100644 docs/development/perf-g500-ladder.md diff --git a/Makefile b/Makefile index 56ec24c7..c279170e 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: help lint format type-check security workflow-lint license-check third-party-notices third-party-notices-check cargo-deny-licenses test pre-push pre-push-clean pre-push-preflight pre-push-fast bazel-test clean test-tck docstring-coverage test-network benchmark test-perf test-perf-xs test-perf-slow test-perf-large coverage coverage-rust coverage-python coverage-node coverage-quick coverage-report coverage-diff coverage-strict check-coverage check-coverage-rust check-coverage-python check-coverage-node check-patch-coverage test-durations test-analytics docs-serve docs-build docs-clean cargo-build codspeed-build codspeed-run bench-traversal bench-fixed-hop-limit bench-fixed-hop-livejournal bench-m4-entry bench-g500-scale20 bench-adjacency-200m bench-file-backed-128m m4-entry-matrix-check durability-isolation-check native-consumers release-load-matrix-check release-load-matrix bulk-construction-conformance-check bulk-construction-conformance cargo-test cargo-check cargo-clippy cargo-fmt cargo-fmt-check clean-builds clean-builds-all pnpm-install pnpm-build pnpm-test-bdd install build release-version-check package-license-verify publish-dry-run publish-dry-run-npm publish-dry-run-docs publish-dry-run-python publish-dry-run-cargo record-release-artifacts clean-env-verify-check clean-env-verify-preflight clean-env-verify +.PHONY: help lint format type-check security workflow-lint license-check third-party-notices third-party-notices-check cargo-deny-licenses test pre-push pre-push-clean pre-push-preflight pre-push-fast bazel-test clean test-tck docstring-coverage test-network benchmark test-perf test-perf-xs test-perf-slow test-perf-large coverage coverage-rust coverage-python coverage-node coverage-quick coverage-report coverage-diff coverage-strict check-coverage check-coverage-rust check-coverage-python check-coverage-node check-patch-coverage test-durations test-analytics docs-serve docs-build docs-clean cargo-build codspeed-build codspeed-run bench-traversal bench-fixed-hop-limit bench-fixed-hop-livejournal bench-m4-entry bench-g500-scale20 bench-g500-ladder bench-adjacency-200m bench-file-backed-128m m4-entry-matrix-check durability-isolation-check native-consumers release-load-matrix-check release-load-matrix bulk-construction-conformance-check bulk-construction-conformance cargo-test cargo-check cargo-clippy cargo-fmt cargo-fmt-check clean-builds clean-builds-all pnpm-install pnpm-build pnpm-test-bdd install build release-version-check package-license-verify publish-dry-run publish-dry-run-npm publish-dry-run-docs publish-dry-run-python publish-dry-run-cargo record-release-artifacts clean-env-verify-check clean-env-verify-preflight clean-env-verify help: ## Show this help message @grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-20s\033[0m %s\n", $$1, $$2}' @@ -309,6 +309,10 @@ bench-m4-entry: ## Emit the M4 entry large/manual evidence envelope (#334; hard bench-g500-scale20: ## Official-parameter SCALE-20 public-facade engineering green (#710; ignored, not Official-track) cargo test -p graphforge-api --release --test scale_g500_scale20 scale20_public_facade_engineering_green -- --ignored --nocapture --test-threads=1 +bench-g500-ladder: ## Bounded billion-edge scale ladder S20-S26 first-fail evidence (#736; ignored, provisioned scale-host) + GF_G500_LADDER_EVIDENCE_OUT="$(CURDIR)/docs/development/g500-ladder-evidence.json" \ + cargo test -p graphforge-api --release --test scale_g500_ladder ladder_public_facade_first_fail_evidence -- --ignored --nocapture --test-threads=1 + bench-adjacency-200m: ## >200M-edge public adjacency build evidence (#336; ignored, scale-host) GF_ADJACENCY_SCALE_EVIDENCE_OUT="$(CURDIR)/docs/development/adjacency-200m-evidence.json" \ GF_ADJACENCY_SCALE_WORK="$(CURDIR)/build/adjacency-200m-work" \ diff --git a/crates/graphforge-api/tests/fixtures/scale_g500_ladder.v1.json b/crates/graphforge-api/tests/fixtures/scale_g500_ladder.v1.json new file mode 100644 index 00000000..2218b3e1 --- /dev/null +++ b/crates/graphforge-api/tests/fixtures/scale_g500_ladder.v1.json @@ -0,0 +1,58 @@ +{ + "schema": "graphforge-billion-edge-ladder/1", + "schema_version": "1", + "objective": "Bounded-memory Graph500-parameter (ef=16 undirected Kronecker) scale ladder that measures the first envelope bottleneck on the public GraphForge facade. Engineering-green, NOT Official-track and NOT TEPS. Certification of one billion live edges is #745, not this profile.", + "seed": 1, + "edgefactor": 16, + "initiator": { "A": 0.57, "B": 0.19, "C": 0.19, "D": 0.05 }, + "policy": { + "directionality": "undirected", + "self_loops": "drop", + "duplicates": "drop", + "canonicalization": "sorted_lo_hi_pair", + "identity": "uuidv7(vertex_index + 1)", + "reconciliation": "raw_attempts == live_unique_edges + self_loops_rejected + duplicates_rejected" + }, + "envelope": { + "rss_bytes": 137438953472, + "disk_bytes": 1099511627776, + "timeout_s": 86400, + "note": "128 GiB peak RSS, 1 TiB local NVMe, 24 hours (M5 #735 host envelope)." + }, + "metrics": [ + "raw_attempts", + "self_loops_rejected", + "duplicates_rejected", + "live_unique_edges", + "wall_time_s", + "rss_peak_bytes", + "disk_used_bytes" + ], + "invocation": "CARGO_TARGET_DIR=/tmp/cargo-g500-ladder make bench-g500-ladder", + "rungs": [ + { + "id": "S10", + "scale": 10, + "buffer_edges": 4096, + "tier": "ci", + "note": "Small always-on validation rung. buffer_edges is intentionally below the rung's edge count so the external sort/spill/merge path is exercised in normal CI." + }, + { + "id": "S20", + "scale": 20, + "buffer_edges": 8388608, + "tier": "provisioned", + "note": "Historical baseline (parity with the #710 SCALE-20 reference client)." + }, + { "id": "S22", "scale": 22, "buffer_edges": 16777216, "tier": "provisioned" }, + { "id": "S24", "scale": 24, "buffer_edges": 33554432, "tier": "provisioned" }, + { "id": "S25", "scale": 25, "buffer_edges": 67108864, "tier": "provisioned" }, + { + "id": "S26", + "scale": 26, + "buffer_edges": 134217728, + "tier": "provisioned", + "note": "SCALE-26/ef=16 = 1,073,741,824 raw attempts. Live persisted edges after self-loop and duplicate policy may fall below one billion; raw attempts MUST NOT be reported as live." + } + ] +} diff --git a/crates/graphforge-api/tests/scale_g500_ladder.rs b/crates/graphforge-api/tests/scale_g500_ladder.rs new file mode 100644 index 00000000..6e88ddbc --- /dev/null +++ b/crates/graphforge-api/tests/scale_g500_ladder.rs @@ -0,0 +1,1147 @@ +//! Billion-live-edge public-facade scale contract and first-fail ladder (#736). +//! +//! This is the M5 (#735) root gate. It defines a **versioned, bounded-memory** +//! Graph500-parameter (ef=16 undirected Kronecker/R-MAT) scale ladder that: +//! +//! 1. Generates edges with memory bounded **independently of total edge count** +//! via external sort + spill + k-way merge (the #710 SCALE-20 reference +//! client retains every raw tuple in memory; this one does not). +//! 2. Reconciles `raw_attempts == live_unique_edges + self_loops_rejected + +//! duplicates_rejected`, so raw generator attempts can never be mistaken for +//! live persisted edges. +//! 3. Stops at and records the **first** envelope (RSS / disk / time) violation +//! instead of making an unsupported SCALE-26 claim. +//! +//! It is **not** Official-track and **not** TEPS, and it does **not** itself +//! certify one billion live edges — that is #745. Small rungs run in normal CI; +//! large rungs are opt-in via `make bench-g500-ladder`. + +use std::cmp::Reverse; +use std::collections::BinaryHeap; +use std::fs::{self, File}; +use std::io::{BufReader, BufWriter, ErrorKind, Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::sync::Arc; +use std::time::Instant; + +use arrow::array::{Array, FixedSizeBinaryArray, Int64Array, StringArray, UInt64Array}; +use arrow::record_batch::RecordBatch; +use graphforge_api::{GraphForge, OperationId, bulk_edge_input_schema, bulk_node_input_schema}; +use graphforge_core::uuid::Uuid; +use serde::Deserialize; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use tempfile::TempDir; + +const PROFILE_SCHEMA: &str = "graphforge-billion-edge-ladder/1"; +const EVIDENCE_SCHEMA: &str = "graphforge-billion-edge-ladder-evidence/1"; +const SCHEMA_VERSION: &str = "1"; +const GENERATOR_NAME: &str = "graphforge-kronecker-rmat-bounded"; +const GENERATOR_SOURCE: &str = "crates/graphforge-api/tests/scale_g500_ladder.rs"; + +const NODE_LABEL: &str = "Node"; +const REL_TYPE: &str = "LINK"; +const BATCH_ROWS: usize = 8_192; +const EDGE_PUBLISH_ROWS: usize = 1_048_576; + +const ONE_HOP: &str = "MATCH (a)-[r]->(b) RETURN b.node_uuid AS id LIMIT 1000"; +const TWO_HOP: &str = "MATCH (a)-[r1]->(b)-[r2]->(c) RETURN c.node_uuid AS id LIMIT 1000"; +const COUNT_EDGES: &str = "MATCH ()-[r:LINK]->() RETURN count(r) AS total"; + +// --------------------------------------------------------------------------- +// Versioned profile (single source of truth for the ladder). +// --------------------------------------------------------------------------- + +#[derive(Debug, Deserialize)] +struct ScaleProfile { + schema: String, + schema_version: String, + seed: u64, + edgefactor: u32, + initiator: Initiator, + envelope: Envelope, + metrics: Vec, + invocation: String, + rungs: Vec, +} + +#[derive(Debug, Clone, Copy, Deserialize)] +struct Initiator { + #[serde(rename = "A")] + a: f64, + #[serde(rename = "B")] + b: f64, + #[serde(rename = "C")] + c: f64, + #[serde(rename = "D")] + d: f64, +} + +#[derive(Debug, Clone, Copy, Deserialize)] +struct Envelope { + rss_bytes: u64, + disk_bytes: u64, + timeout_s: u64, +} + +#[derive(Debug, Clone, Deserialize)] +struct Rung { + id: String, + scale: u32, + buffer_edges: usize, + tier: String, + #[serde(default)] + #[allow(dead_code)] + note: Option, +} + +fn profile_path() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/scale_g500_ladder.v1.json") +} + +fn load_profile() -> ScaleProfile { + let raw = fs::read_to_string(profile_path()).expect("read ladder profile fixture"); + serde_json::from_str(&raw).expect("parse ladder profile fixture") +} + +// --------------------------------------------------------------------------- +// Bounded generation: external sort + spill + k-way merge. +// Peak buffer size is `buffer_edges`, independent of total edge count. +// --------------------------------------------------------------------------- + +/// Result of the bounded generation pass (before the merge that yields live +/// unique edges). Spill runs live under `work` and are cleaned with it. +struct SpillRuns { + runs: Vec, + raw_attempts: u64, + self_loops_rejected: u64, + /// Largest number of edges ever resident in memory during generation. + peak_buffer_len: usize, +} + +/// Generate `2^scale * edge_factor` Kronecker attempts, dropping self-loops, +/// canonicalizing to undirected `(lo, hi)` pairs, and spilling **sorted (not +/// deduplicated)** fixed-size runs so cross-run and intra-run duplicates are +/// both accounted for at merge time. +fn generate_spill_runs( + scale: u32, + edge_factor: u32, + initiator: Initiator, + seed: u64, + buffer_edges: usize, + work: &Path, +) -> SpillRuns { + assert!((1..=31).contains(&scale), "SCALE must fit u32 vertex ids"); + assert!(buffer_edges >= 1, "buffer_edges must be positive"); + let raw_attempts = (1u64 << scale) + .checked_mul(u64::from(edge_factor)) + .expect("raw attempt count fits u64"); + + let mut rng = SplitMix64(seed); + let mut buf: Vec<(u32, u32)> = Vec::with_capacity(buffer_edges); + let mut runs = Vec::new(); + let mut self_loops_rejected = 0u64; + let mut peak_buffer_len = 0usize; + + for _ in 0..raw_attempts { + let (src, dst) = kronecker_edge(scale, initiator, &mut rng); + if src == dst { + self_loops_rejected += 1; + continue; + } + let (lo, hi) = if src < dst { (src, dst) } else { (dst, src) }; + buf.push(( + u32::try_from(lo).expect("src fits u32"), + u32::try_from(hi).expect("dst fits u32"), + )); + if buf.len() >= buffer_edges { + peak_buffer_len = peak_buffer_len.max(buf.len()); + flush_run(&mut buf, work, runs.len(), &mut runs); + } + } + peak_buffer_len = peak_buffer_len.max(buf.len()); + if !buf.is_empty() { + flush_run(&mut buf, work, runs.len(), &mut runs); + } + + SpillRuns { + runs, + raw_attempts, + self_loops_rejected, + peak_buffer_len, + } +} + +fn flush_run(buf: &mut Vec<(u32, u32)>, work: &Path, index: usize, runs: &mut Vec) { + buf.sort_unstable(); + let path = work.join(format!("run-{index:05}.bin")); + let mut writer = BufWriter::new(File::create(&path).expect("create spill run")); + for &(src, dst) in buf.iter() { + writer.write_all(&src.to_le_bytes()).expect("write src"); + writer.write_all(&dst.to_le_bytes()).expect("write dst"); + } + writer.flush().expect("flush spill run"); + buf.clear(); + runs.push(path); +} + +struct RunReader { + reader: BufReader, +} + +impl RunReader { + fn open(path: &Path) -> Self { + RunReader { + reader: BufReader::new(File::open(path).expect("open spill run")), + } + } + + fn next_pair(&mut self) -> Option<(u32, u32)> { + let mut bytes = [0u8; 8]; + match self.reader.read_exact(&mut bytes) { + Ok(()) => { + let src = u32::from_le_bytes(bytes[0..4].try_into().unwrap()); + let dst = u32::from_le_bytes(bytes[4..8].try_into().unwrap()); + Some((src, dst)) + } + Err(err) if err.kind() == ErrorKind::UnexpectedEof => None, + Err(err) => panic!("read spill run: {err}"), + } + } +} + +/// Counts produced by the merge that deduplicates the spilled runs. +struct MergeCounts { + live_unique_edges: u64, + duplicates_rejected: u64, +} + +/// K-way merge of sorted runs. Emits each unique undirected pair exactly once +/// in canonical sorted order and counts every dropped duplicate. Memory is +/// `O(number_of_runs)`, not `O(total_edges)`. +fn merge_runs(runs: &[PathBuf], mut emit: F) -> MergeCounts { + let mut readers: Vec = runs.iter().map(|p| RunReader::open(p)).collect(); + let mut heap: BinaryHeap> = BinaryHeap::new(); + for (idx, reader) in readers.iter_mut().enumerate() { + if let Some((src, dst)) = reader.next_pair() { + heap.push(Reverse((src, dst, idx))); + } + } + + let mut live_unique_edges = 0u64; + let mut duplicates_rejected = 0u64; + let mut last: Option<(u32, u32)> = None; + while let Some(Reverse((src, dst, idx))) = heap.pop() { + let pair = (src, dst); + if Some(pair) == last { + duplicates_rejected += 1; + } else { + emit(src, dst); + live_unique_edges += 1; + last = Some(pair); + } + if let Some((nsrc, ndst)) = readers[idx].next_pair() { + heap.push(Reverse((nsrc, ndst, idx))); + } + } + + MergeCounts { + live_unique_edges, + duplicates_rejected, + } +} + +/// Full reconciled generation summary for a rung. +struct GenSummary { + raw_attempts: u64, + self_loops_rejected: u64, + duplicates_rejected: u64, + live_unique_edges: u64, + peak_buffer_len: usize, + run_count: usize, + input_fingerprint: String, +} + +impl GenSummary { + fn reconciles(&self) -> bool { + self.raw_attempts + == self.live_unique_edges + self.self_loops_rejected + self.duplicates_rejected + } +} + +/// Generate + merge into memory (small rungs / unit tests only). +fn bounded_generation( + scale: u32, + edge_factor: u32, + initiator: Initiator, + seed: u64, + buffer_edges: usize, +) -> (GenSummary, Vec<(u32, u32)>) { + let work = TempDir::new().expect("spill workspace"); + let spill = generate_spill_runs( + scale, + edge_factor, + initiator, + seed, + buffer_edges, + work.path(), + ); + let mut edges = Vec::new(); + let mut hasher = Sha256::new(); + let merge = merge_runs(&spill.runs, |src, dst| { + hasher.update(src.to_le_bytes()); + hasher.update(dst.to_le_bytes()); + edges.push((src, dst)); + }); + let summary = GenSummary { + raw_attempts: spill.raw_attempts, + self_loops_rejected: spill.self_loops_rejected, + duplicates_rejected: merge.duplicates_rejected, + live_unique_edges: merge.live_unique_edges, + peak_buffer_len: spill.peak_buffer_len, + run_count: spill.runs.len(), + input_fingerprint: format!("sha256:{}", hex_encode(hasher.finalize())), + }; + (summary, edges) +} + +// --------------------------------------------------------------------------- +// Kronecker generation (bounded, no whole-graph retention). +// --------------------------------------------------------------------------- + +fn kronecker_edge(scale: u32, init: Initiator, rng: &mut SplitMix64) -> (u64, u64) { + let mut src = 0u64; + let mut dst = 0u64; + for bit in 0..scale { + let sample = rng.next_unit(); + let (di, dj) = if sample < init.a { + (0, 0) + } else if sample < init.a + init.b { + (0, 1) + } else if sample < init.a + init.b + init.c { + (1, 0) + } else { + (1, 1) + }; + src |= di << bit; + dst |= dj << bit; + } + (src, dst) +} + +// --------------------------------------------------------------------------- +// First-fail ladder harness over the public facade. +// --------------------------------------------------------------------------- + +/// A stop condition the ladder measures after every phase. +#[derive(Clone, Copy)] +struct RunEnvelope { + rss_bytes: u64, + disk_bytes: u64, + timeout_s: u64, +} + +impl From for RunEnvelope { + fn from(e: Envelope) -> Self { + RunEnvelope { + rss_bytes: e.rss_bytes, + disk_bytes: e.disk_bytes, + timeout_s: e.timeout_s, + } + } +} + +/// The evidence + pass/fail for a single rung attempt. +struct RungOutcome { + passed: bool, + evidence: Value, +} + +/// Check the envelope after a phase. Returns `Some(error_class)` on the first +/// violation so the caller can stop the ladder. +fn envelope_violation( + env: &RunEnvelope, + started: Instant, + project: &Path, + spill: &Path, +) -> Option<&'static str> { + if peak_rss_bytes().is_some_and(|rss| rss > env.rss_bytes) { + return Some("oom"); + } + let disk = directory_bytes(project).unwrap_or(0) + directory_bytes(spill).unwrap_or(0); + if disk > env.disk_bytes { + return Some("disk_exhaustion"); + } + if started.elapsed().as_secs() > env.timeout_s { + return Some("timeout"); + } + None +} + +#[allow(clippy::too_many_lines)] +fn run_rung( + profile: &ScaleProfile, + rung: &Rung, + env: RunEnvelope, + edge_factor: u32, +) -> RungOutcome { + let started = Instant::now(); + let workspace = TempDir::new().expect("rung workspace"); + let spill_dir = workspace.path().join("spill"); + let project = workspace.path().join("project"); + fs::create_dir_all(&spill_dir).expect("spill dir"); + fs::create_dir_all(&project).expect("project dir"); + + let mut steps: Vec = Vec::new(); + let mut first_failing_phase: Option<&'static str> = None; + let mut error_class: Option<&'static str> = None; + + // ---- generate ---- + let gen_started = Instant::now(); + let spill = generate_spill_runs( + rung.scale, + edge_factor, + profile.initiator, + profile.seed, + rung.buffer_edges, + &spill_dir, + ); + let generate_s = gen_started.elapsed().as_secs_f64(); + steps.push(json!({ + "id": "generate", + "pass": true, + "wall_time_s": generate_s, + "detail": { + "raw_attempts": spill.raw_attempts, + "self_loops_rejected": spill.self_loops_rejected, + "peak_buffer_len": spill.peak_buffer_len, + "buffer_edges": rung.buffer_edges, + "run_count": spill.runs.len(), + } + })); + if let Some(class) = envelope_violation(&env, started, &project, &spill_dir) { + first_failing_phase = Some("generate"); + error_class = Some(class); + } + + // ---- ingest (merge + publish through the public facade) ---- + let mut live_unique_edges = 0u64; + let mut duplicates_rejected = 0u64; + let mut input_fingerprint = String::from("sha256:"); + if first_failing_phase.is_none() { + let ingest_started = Instant::now(); + let graph = GraphForge::new(Some(project.to_str().expect("utf8 project"))) + .expect("open GraphForge for ingest"); + publish_nodes(&graph, 1u64 << rung.scale); + let mut sink = EdgeSink::new(&graph); + let merge = merge_runs(&spill.runs, |src, dst| sink.push(src, dst)); + sink.flush(); + live_unique_edges = merge.live_unique_edges; + duplicates_rejected = merge.duplicates_rejected; + input_fingerprint = format!("sha256:{}", hex_encode(sink.finish())); + drop(graph); + let ingest_s = ingest_started.elapsed().as_secs_f64(); + steps.push(json!({ + "id": "ingest", + "pass": true, + "wall_time_s": ingest_s, + "detail": { + "live_unique_edges": live_unique_edges, + "duplicates_rejected": duplicates_rejected, + "input_fingerprint": input_fingerprint, + } + })); + if let Some(class) = envelope_violation(&env, started, &project, &spill_dir) { + first_failing_phase = Some("ingest"); + error_class = Some(class); + } + } + + // ---- reopen + recount ---- + let mut node_count = 0u64; + let mut edge_count = 0u64; + let mut gsi = String::new(); + if first_failing_phase.is_none() { + let reopen_started = Instant::now(); + let graph = GraphForge::new(Some(project.to_str().expect("utf8 project"))) + .expect("reopen GraphForge"); + node_count = graph.node_count(NODE_LABEL).expect("node_count"); + edge_count = scalar_count(&graph.execute(COUNT_EDGES).expect("edge count")); + let reopen_s = reopen_started.elapsed().as_secs_f64(); + gsi = gsi_undirected(node_count, edge_count); + steps.push(json!({ + "id": "reopen", + "pass": true, + "wall_time_s": reopen_s, + "detail": { "node_count": node_count, "edge_count": edge_count, "gsi": gsi } + })); + + // ---- deterministic LIMIT queries ---- + let hop1_started = Instant::now(); + let hop1 = graph.execute(ONE_HOP).expect("one-hop LIMIT"); + let hop1_rows = row_count(&hop1); + steps.push(json!({ + "id": "cypher_limit_1hop", + "pass": hop1_rows <= 1_000, + "wall_time_s": hop1_started.elapsed().as_secs_f64(), + "detail": { "rows": hop1_rows } + })); + + let hop2_started = Instant::now(); + let hop2 = graph.execute(TWO_HOP).expect("two-hop LIMIT"); + let hop2_rows = row_count(&hop2); + steps.push(json!({ + "id": "cypher_limit_2hop", + "pass": hop2_rows <= 1_000, + "wall_time_s": hop2_started.elapsed().as_secs_f64(), + "detail": { "rows": hop2_rows } + })); + drop(graph); + if let Some(class) = envelope_violation(&env, started, &project, &spill_dir) { + first_failing_phase = Some("query"); + error_class = Some(class); + } + } + + let disk_used_bytes = + directory_bytes(&project).unwrap_or(0) + directory_bytes(&spill_dir).unwrap_or(0); + let reconciles = first_failing_phase.is_some() + || spill.raw_attempts + == live_unique_edges + spill.self_loops_rejected + duplicates_rejected; + let passed = first_failing_phase.is_none() && reconciles; + + let evidence = json!({ + "schema": EVIDENCE_SCHEMA, + "schema_version": SCHEMA_VERSION, + "profile_schema": profile.schema, + "track": null, + "rung": rung.id, + "scale": rung.scale, + "tier": rung.tier, + "edgefactor": edge_factor, + "gsi": if gsi.is_empty() { Value::Null } else { Value::String(gsi) }, + "pass": passed, + "first_failing_phase": first_failing_phase, + "error_class": error_class, + "reconciles": reconciles, + "counts": { + "raw_attempts": spill.raw_attempts, + "self_loops_rejected": spill.self_loops_rejected, + "duplicates_rejected": duplicates_rejected, + "live_unique_edges": live_unique_edges, + "reconciliation": "raw_attempts == live_unique_edges + self_loops_rejected + duplicates_rejected", + }, + "persisted": { "node_count": node_count, "edge_count": edge_count }, + "input_fingerprint": input_fingerprint, + "wall_time_s": started.elapsed().as_secs_f64(), + "rss_peak_bytes": peak_rss_bytes(), + "disk_used_bytes": disk_used_bytes, + "machine_envelope": { + "rss_bytes": env.rss_bytes, + "disk_bytes": env.disk_bytes, + "timeout_s": env.timeout_s, + }, + "generator": { + "name": GENERATOR_NAME, + "source": GENERATOR_SOURCE, + "version": "1", + "seed": profile.seed, + "initiator": { + "A": profile.initiator.a, "B": profile.initiator.b, + "C": profile.initiator.c, "D": profile.initiator.d + }, + }, + "sut": { "name": "graphforge", "version": env!("CARGO_PKG_VERSION"), "git_sha": git_sha() }, + "teps": null, + "notes": "Bounded-memory engineering green. NOT Official-track, NOT TEPS. Certification of one billion live edges is #745, not this profile.", + "steps": steps, + }); + + RungOutcome { passed, evidence } +} + +/// Drive the ladder rung-by-rung, stopping at the first failing rung. +fn run_ladder(profile: &ScaleProfile, env: RunEnvelope, rungs: &[Rung]) -> Vec { + let mut evidence = Vec::new(); + for rung in rungs { + let outcome = run_rung(profile, rung, env, profile.edgefactor); + let passed = outcome.passed; + evidence.push(outcome.evidence); + if !passed { + break; + } + } + evidence +} + +// --------------------------------------------------------------------------- +// Streaming edge publisher (bounded by EDGE_PUBLISH_ROWS). +// --------------------------------------------------------------------------- + +struct EdgeSink<'a> { + graph: &'a GraphForge, + buf: Vec<(u32, u32)>, + chunk_index: u128, + hasher: Sha256, +} + +impl<'a> EdgeSink<'a> { + fn new(graph: &'a GraphForge) -> Self { + EdgeSink { + graph, + buf: Vec::with_capacity(EDGE_PUBLISH_ROWS), + chunk_index: 0, + hasher: Sha256::new(), + } + } + + fn push(&mut self, src: u32, dst: u32) { + self.hasher.update(src.to_le_bytes()); + self.hasher.update(dst.to_le_bytes()); + self.buf.push((src, dst)); + if self.buf.len() >= EDGE_PUBLISH_ROWS { + self.flush(); + } + } + + fn flush(&mut self) { + if self.buf.is_empty() { + return; + } + let schema = bulk_edge_input_schema(Vec::new()).expect("edge schema"); + let mut batches = Vec::new(); + let mut offset = 0usize; + while offset < self.buf.len() { + let end = (offset + BATCH_ROWS).min(self.buf.len()); + let slice = &self.buf[offset..end]; + let mut edge_ids = Vec::with_capacity(slice.len()); + let mut sources = Vec::with_capacity(slice.len()); + let mut targets = Vec::with_capacity(slice.len()); + for (local, &(src, dst)) in slice.iter().enumerate() { + let ordinal = self + .chunk_index + .saturating_mul(EDGE_PUBLISH_ROWS as u128) + .saturating_add(u128::try_from(offset + local).expect("edge ordinal")); + edge_ids.push(uuidv7(0xE000_0000_0000u128 + ordinal + 1)); + sources.push(uuidv7(u128::from(src) + 1)); + targets.push(uuidv7(u128::from(dst) + 1)); + } + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new( + FixedSizeBinaryArray::try_from_iter(edge_ids.iter().map(Uuid::as_bytes)) + .expect("edge_uuid column"), + ), + Arc::new(StringArray::from(vec![REL_TYPE; slice.len()])), + Arc::new( + FixedSizeBinaryArray::try_from_iter(sources.iter().map(Uuid::as_bytes)) + .expect("source_uuid column"), + ), + Arc::new( + FixedSizeBinaryArray::try_from_iter(targets.iter().map(Uuid::as_bytes)) + .expect("target_uuid column"), + ), + ], + ) + .expect("edge batch"); + batches.push(batch); + offset = end; + } + self.graph + .publish_bulk_edges(OperationId(uuidv7(0xB100 + self.chunk_index)), &batches) + .expect("publish_bulk_edges"); + self.chunk_index += 1; + self.buf.clear(); + } + + fn finish(mut self) -> String { + // Any residual edges must already be flushed by the caller. + debug_assert!(self.buf.is_empty()); + hex_encode(std::mem::take(&mut self.hasher).finalize()) + } +} + +fn publish_nodes(graph: &GraphForge, vertex_count: u64) { + let total = usize::try_from(vertex_count).expect("vertex count fits usize"); + let schema = bulk_node_input_schema(Vec::new()).expect("node schema"); + let mut offset = 0usize; + while offset < total { + let chunk_end = (offset + EDGE_PUBLISH_ROWS).min(total); + let mut batches = Vec::new(); + let mut inner = offset; + while inner < chunk_end { + let end = (inner + BATCH_ROWS).min(chunk_end); + let count = end - inner; + let ids = (inner..end) + .map(|index| uuidv7(u128::try_from(index + 1).expect("node seed"))) + .collect::>(); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new( + FixedSizeBinaryArray::try_from_iter(ids.iter().map(Uuid::as_bytes)) + .expect("node_uuid column"), + ), + Arc::new(StringArray::from(vec![NODE_LABEL; count])), + ], + ) + .expect("node batch"); + batches.push(batch); + inner = end; + } + graph + .publish_bulk_nodes(OperationId(uuidv7(0xB001_0000 + offset as u128)), &batches) + .expect("publish_bulk_nodes"); + offset = chunk_end; + } +} + +// --------------------------------------------------------------------------- +// Shared small helpers (kept local so the #710 SCALE-20 client stays untouched). +// --------------------------------------------------------------------------- + +fn scalar_count(result: &graphforge_api::ExecutionResult) -> u64 { + assert_eq!(row_count(result), 1, "count query must return one row"); + let column = result.batches[0].column(0); + if let Some(values) = column.as_any().downcast_ref::() { + u64::try_from(values.value(0)).expect("non-negative count") + } else if let Some(values) = column.as_any().downcast_ref::() { + values.value(0) + } else { + panic!("unexpected count type {:?}", column.data_type()); + } +} + +fn row_count(result: &graphforge_api::ExecutionResult) -> usize { + result.batches.iter().map(RecordBatch::num_rows).sum() +} + +fn gsi_undirected(vertex_count: u64, edge_count: u64) -> String { + let (code, tag) = scale_band(vertex_count); + format!( + "GU-{code}-{tag}-{}", + density_code(undirected_density(vertex_count, edge_count)) + ) +} + +fn scale_band(vertex_count: u64) -> (&'static str, &'static str) { + match vertex_count { + 0..=99 => ("01", "XS"), + 100..=999 => ("02", "XS"), + 1_000..=9_999 => ("03", "XS"), + 10_000..=99_999 => ("04", "XS"), + 100_000..=999_999 => ("05", "SM"), + 1_000_000..=9_999_999 => ("06", "MD"), + 10_000_000..=99_999_999 => ("07", "LG"), + 100_000_000..=999_999_999 => ("08", "XL"), + 1_000_000_000..=9_999_999_999 => ("09", "2XL"), + 10_000_000_000..=99_999_999_999 => ("10", "3XL"), + _ => ("**", "BIG"), + } +} + +#[allow(clippy::cast_precision_loss)] +fn undirected_density(vertex_count: u64, edge_count: u64) -> f64 { + if vertex_count < 2 { + return 0.0; + } + let denom = (vertex_count as f64) * ((vertex_count - 1) as f64); + ((2.0 * edge_count as f64) / denom).clamp(0.0, 1.0) +} + +#[allow(clippy::cast_possible_truncation)] +fn density_code(density: f64) -> String { + let percent = (density * 100.0).round() as i32; + format!("D{percent:02}") +} + +fn hex_encode(bytes: impl AsRef<[u8]>) -> String { + const HEX: &[u8; 16] = b"0123456789abcdef"; + let bytes = bytes.as_ref(); + let mut out = String::with_capacity(bytes.len() * 2); + for byte in bytes { + out.push(HEX[(byte >> 4) as usize] as char); + out.push(HEX[(byte & 0xf) as usize] as char); + } + out +} + +fn uuidv7(seed: u128) -> Uuid { + let mut bytes = seed.to_be_bytes(); + bytes[6] = (bytes[6] & 0x0f) | 0x70; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + Uuid::from_bytes(bytes) +} + +fn git_sha() -> Value { + Command::new("git") + .args(["rev-parse", "HEAD"]) + .output() + .ok() + .filter(|output| output.status.success()) + .and_then(|output| String::from_utf8(output.stdout).ok()) + .map_or(Value::Null, |sha| Value::String(sha.trim().to_owned())) +} + +fn directory_bytes(path: &Path) -> std::io::Result { + if !path.exists() { + return Ok(0); + } + if path.is_file() { + return Ok(path.metadata()?.len()); + } + let mut total = 0u64; + for entry in fs::read_dir(path)? { + let entry = entry?; + let metadata = entry.metadata()?; + total += if metadata.is_dir() { + directory_bytes(&entry.path())? + } else { + metadata.len() + }; + } + Ok(total) +} + +fn peak_rss_bytes() -> Option { + if let Ok(contents) = fs::read_to_string("/proc/self/status") { + for line in contents.lines() { + if let Some(value) = line.strip_prefix("VmHWM:") { + let kb = value + .trim() + .trim_end_matches(" kB") + .trim() + .parse::() + .ok()?; + return Some(kb.saturating_mul(1024)); + } + } + } + let output = Command::new("ps") + .args(["-o", "rss=", "-p", &std::process::id().to_string()]) + .output() + .ok()?; + if !output.status.success() { + return None; + } + let kb = String::from_utf8_lossy(&output.stdout) + .trim() + .parse::() + .ok()?; + Some(kb.saturating_mul(1024)) +} + +struct SplitMix64(u64); + +impl SplitMix64 { + fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + + #[allow(clippy::cast_precision_loss)] + fn next_unit(&mut self) -> f64 { + (self.next_u64() >> 11) as f64 / ((1u64 << 53) as f64) + } +} + +/// Naive whole-graph reference generation (small scales only) used to prove the +/// bounded external-sort path yields the identical live edge set. +fn reference_generation( + scale: u32, + edge_factor: u32, + init: Initiator, + seed: u64, +) -> (Vec<(u32, u32)>, u64, u64) { + let raw = (1u64 << scale) * u64::from(edge_factor); + let mut rng = SplitMix64(seed); + let mut edges = Vec::new(); + let mut self_loops = 0u64; + for _ in 0..raw { + let (src, dst) = kronecker_edge(scale, init, &mut rng); + if src == dst { + self_loops += 1; + continue; + } + let (lo, hi) = if src < dst { (src, dst) } else { (dst, src) }; + edges.push((u32::try_from(lo).unwrap(), u32::try_from(hi).unwrap())); + } + edges.sort_unstable(); + edges.dedup(); + (edges, raw, self_loops) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +/// The versioned profile is well-formed and carries the M5 host envelope. +#[test] +fn ladder_profile_is_versioned_and_pinned() { + let profile = load_profile(); + assert_eq!(profile.schema, PROFILE_SCHEMA); + assert_eq!(profile.schema_version, SCHEMA_VERSION); + assert_eq!(profile.edgefactor, 16, "Official parameter ef must be 16"); + // M5 #735 host envelope: 128 GiB RSS, 1 TiB disk, 24 h. + assert_eq!(profile.envelope.rss_bytes, 137_438_953_472); + assert_eq!(profile.envelope.disk_bytes, 1_099_511_627_776); + assert_eq!(profile.envelope.timeout_s, 86_400); + assert!(!profile.invocation.is_empty()); + for required in [ + "raw_attempts", + "self_loops_rejected", + "duplicates_rejected", + "live_unique_edges", + ] { + assert!( + profile.metrics.iter().any(|m| m == required), + "profile metrics must declare {required}" + ); + } + // Exactly one always-on CI rung, and a SCALE-26 provisioned terminal rung. + assert_eq!( + profile.rungs.iter().filter(|r| r.tier == "ci").count(), + 1, + "exactly one CI rung must run in normal validation" + ); + assert!( + profile + .rungs + .iter() + .any(|r| r.scale == 26 && r.tier == "provisioned"), + "ladder must include a provisioned SCALE-26 terminal rung" + ); + // Ladder scales are strictly increasing. + assert!( + profile.rungs.windows(2).all(|w| w[0].scale < w[1].scale), + "rung scales must strictly increase" + ); +} + +/// Raw attempts, self-loops, duplicates, and live edges reconcile exactly. +#[test] +fn bounded_generation_reconciles_counts() { + let init = Initiator { + a: 0.57, + b: 0.19, + c: 0.19, + d: 0.05, + }; + let (summary, edges) = bounded_generation(10, 16, init, 1, 4096); + assert_eq!(summary.raw_attempts, (1u64 << 10) * 16); + assert!( + summary.self_loops_rejected > 0, + "Kronecker must yield self-loops" + ); + assert!( + summary.duplicates_rejected > 0, + "Kronecker must yield duplicates" + ); + assert_eq!(summary.live_unique_edges, edges.len() as u64); + assert!( + summary.reconciles(), + "raw {} != live {} + self_loops {} + dups {}", + summary.raw_attempts, + summary.live_unique_edges, + summary.self_loops_rejected, + summary.duplicates_rejected + ); + // Live edges are canonical (lo < hi) and strictly sorted/unique. + assert!( + edges.iter().all(|(s, d)| s < d), + "edges must be canonical undirected" + ); + assert!( + edges.windows(2).all(|w| w[0] < w[1]), + "edges must be strictly sorted & unique" + ); +} + +/// Raw attempts can never be reported as live persisted edges. +#[test] +fn raw_attempts_exceed_live_edges() { + let init = Initiator { + a: 0.57, + b: 0.19, + c: 0.19, + d: 0.05, + }; + let (summary, _) = bounded_generation(12, 16, init, 1, 8192); + assert!( + summary.live_unique_edges < summary.raw_attempts, + "live persisted edges must be strictly fewer than raw attempts" + ); +} + +/// Same seed + profile ⇒ identical fingerprint and counts across two runs. +#[test] +fn bounded_generation_is_deterministic() { + let init = Initiator { + a: 0.57, + b: 0.19, + c: 0.19, + d: 0.05, + }; + let (a, _) = bounded_generation(11, 16, init, 1, 3000); + // Different buffer size must not change the deduplicated result. + let (b, _) = bounded_generation(11, 16, init, 1, 512); + assert_eq!(a.input_fingerprint, b.input_fingerprint); + assert_eq!(a.live_unique_edges, b.live_unique_edges); + assert_eq!(a.self_loops_rejected, b.self_loops_rejected); + assert_eq!(a.duplicates_rejected, b.duplicates_rejected); +} + +/// The bounded external-sort path yields the identical live edge set as a +/// naive whole-graph in-memory generation. +#[test] +fn bounded_generation_matches_reference() { + let init = Initiator { + a: 0.57, + b: 0.19, + c: 0.19, + d: 0.05, + }; + let (reference, raw, self_loops) = reference_generation(9, 16, init, 1); + let (summary, edges) = bounded_generation(9, 16, init, 1, 777); + assert_eq!(summary.raw_attempts, raw); + assert_eq!(summary.self_loops_rejected, self_loops); + assert_eq!( + edges, reference, + "bounded live edge set must equal reference" + ); +} + +/// A buffer smaller than the rung's edge count must spill multiple runs while +/// keeping peak resident edges at or below the buffer. +#[test] +fn bounded_generation_spills_and_stays_bounded() { + let init = Initiator { + a: 0.57, + b: 0.19, + c: 0.19, + d: 0.05, + }; + let buffer = 1024usize; + let (summary, _) = bounded_generation(12, 16, init, 1, buffer); + assert!( + summary.run_count > 1, + "small buffer must spill multiple runs" + ); + assert!( + summary.peak_buffer_len <= buffer, + "resident edges {} must stay within buffer {buffer}", + summary.peak_buffer_len + ); +} + +/// The first envelope violation stops the ladder and records the first failing +/// phase without any pass claim. Uses a deliberately tiny RSS ceiling so the +/// stop is deterministic on any host. +#[test] +fn first_fail_stops_at_envelope_violation() { + let profile = load_profile(); + let tiny_env = RunEnvelope { + rss_bytes: 1, + disk_bytes: 1, + timeout_s: 86_400, + }; + let ci_rung = profile + .rungs + .iter() + .find(|r| r.tier == "ci") + .cloned() + .expect("ci rung"); + let outcome = run_rung(&profile, &ci_rung, tiny_env, profile.edgefactor); + assert!(!outcome.passed, "a violated envelope must not pass"); + assert_eq!(outcome.evidence["first_failing_phase"], "generate"); + assert_eq!(outcome.evidence["error_class"], "oom"); + assert!(outcome.evidence["teps"].is_null()); + assert!( + outcome.evidence["scale"].as_u64().unwrap() < 26, + "a failed low rung must not carry a SCALE-26 pass claim" + ); +} + +/// Always-on CI rung: full bounded generate → ingest → reopen → LIMIT flow on +/// the public facade, with reconciliation and persisted-count evidence. +#[test] +fn ci_rung_public_facade_engineering_green() { + let profile = load_profile(); + let ci_rung = profile + .rungs + .iter() + .find(|r| r.tier == "ci") + .cloned() + .expect("ci rung"); + let outcome = run_rung( + &profile, + &ci_rung, + profile.envelope.into(), + profile.edgefactor, + ); + assert!(outcome.passed, "CI rung must pass: {:#}", outcome.evidence); + let ev = &outcome.evidence; + assert_eq!(ev["schema"], EVIDENCE_SCHEMA); + assert!(ev["track"].is_null(), "must not claim Official track"); + assert!(ev["teps"].is_null()); + assert_eq!(ev["reconciles"], true); + assert!(ev["first_failing_phase"].is_null()); + // Persisted counts equal the reconciled live edge count. + let live = ev["counts"]["live_unique_edges"].as_u64().unwrap(); + assert_eq!(ev["persisted"]["edge_count"].as_u64().unwrap(), live); + assert_eq!( + ev["persisted"]["node_count"].as_u64().unwrap(), + 1u64 << ci_rung.scale + ); + assert!(live > 0, "CI rung must persist a non-empty graph"); +} + +/// Provisioned full ladder (SCALE-20 → SCALE-26). Opt-in via +/// `make bench-g500-ladder`. Writes one evidence object per attempted rung and +/// stops at the first rung that exceeds the declared 128 GiB / 1 TiB / 24 h +/// envelope. Never asserts a billion-edge product claim (that is #745). +#[test] +#[ignore = "Provisioned billion-edge scale ladder; make bench-g500-ladder"] +fn ladder_public_facade_first_fail_evidence() { + let profile = load_profile(); + let provisioned: Vec = profile + .rungs + .iter() + .filter(|r| r.tier == "provisioned") + .cloned() + .collect(); + let evidence = run_ladder(&profile, profile.envelope.into(), &provisioned); + assert!( + !evidence.is_empty(), + "ladder must attempt at least one rung" + ); + + let out = std::env::var("GF_G500_LADDER_EVIDENCE_OUT").map_or_else( + |_| PathBuf::from("build/g500-ladder-evidence.json"), + PathBuf::from, + ); + if let Some(parent) = out.parent() { + fs::create_dir_all(parent).expect("evidence parent"); + } + fs::write( + &out, + serde_json::to_vec_pretty(&json!({ + "schema": EVIDENCE_SCHEMA, + "schema_version": SCHEMA_VERSION, + "profile_schema": profile.schema, + "rungs": evidence, + })) + .expect("serialize ladder evidence"), + ) + .expect("write ladder evidence"); + + // Every attempted rung reconciles; the ladder stops at the first failure. + for rung in &evidence { + assert_eq!(rung["reconciles"], true, "attempted rung must reconcile"); + } +} diff --git a/docs/development/perf-g500-ladder.md b/docs/development/perf-g500-ladder.md new file mode 100644 index 00000000..cd47700d --- /dev/null +++ b/docs/development/perf-g500-ladder.md @@ -0,0 +1,126 @@ +# Bounded billion-edge scale ladder (#736) + +This is the M5 ([#735](https://github.com/DecisionNerd/graphforge/issues/735)) +**root gate**: a versioned, **bounded-memory** Graph500-parameter (`edgefactor=16` +undirected Kronecker/R-MAT) scale ladder that measures the *first real +bottleneck* on the published Rust facade before any billion-edge certification +([#745](https://github.com/DecisionNerd/graphforge/issues/745)). + +It is a thin in-tree **reference client** under the +[Scale Evaluation](../reference/scale-evaluation.md) contract — **not** +Official-track and **not** TEPS. It does **not** itself claim one billion live +edges. + +| Claim | Status | +|---|---| +| Official Graph500 submission | No | +| `track: official` in evidence JSON | Must not be set (`track` is `null`) | +| Pinned `github.com/graph500/graph500` generator | No — bounded bench-local Kronecker in the test file | +| Graph500 BFS kernel / harmonic-mean TEPS | Non-goal (`teps` is `null`) | +| One-billion-live-edge product certification | No — that is #745 | +| Engineering green (generate → ingest → reopen → GSI → `LIMIT 1000`) | Yes, on `GraphForge::publish_bulk_*` + `execute` | + +## What is new versus the #710 SCALE-20 client + +[perf-g500-scale20.md](perf-g500-scale20.md) retains **every** raw tuple in +memory (a `Vec` of all `2^scale * 16` attempts) and sorts/deduplicates in RAM. +That is fine at SCALE-20 but allocates ~8.6 GiB of raw tuples at SCALE-26. + +This ladder replaces that with **external sort + spill + k-way merge**: + +1. Generate Kronecker attempts into a fixed `buffer_edges` buffer. +2. When full, sort the buffer and **spill a run file** to disk (no in-buffer + dedup, so every duplicate is counted at merge). +3. K-way merge the sorted runs, emitting each unique undirected pair once. + +Peak resident edges never exceed `buffer_edges`, **independent of total edge +count**. Live edges stream straight into `publish_bulk_edges` during the merge, +so ingest is bounded too. + +## Counts always reconcile + +Every attempted rung proves: + +``` +raw_attempts == live_unique_edges + self_loops_rejected + duplicates_rejected +``` + +`raw_attempts` (`2^scale * 16`) can therefore **never** be reported as live +persisted edges. SCALE-26 produces `1,073,741,824` raw attempts; the live +persisted count after self-loop and duplicate policy may fall **below** one +billion, and the evidence records both distinctly. + +## The versioned profile + +The ladder, seed, initiator, policy, envelope, metrics, and exact invocation +live in a single committed profile, +[`crates/graphforge-api/tests/fixtures/scale_g500_ladder.v1.json`](../../crates/graphforge-api/tests/fixtures/scale_g500_ladder.v1.json) +(`schema: graphforge-billion-edge-ladder/1`). The runner reads it; do not +hard-code rungs elsewhere. + +| Field | Value | +|---|---| +| Rungs | S10 (CI), S20, S22, S24, S25, S26 (provisioned) | +| Seed / initiator | `1` / `A,B,C,D = 0.57, 0.19, 0.19, 0.05` | +| Policy | undirected, drop self-loops, drop duplicates, canonical `(lo,hi)` | +| Host envelope | **128 GiB** peak RSS, **1 TiB** local NVMe, **24 h** | + +The **S10** rung runs in normal CI and deliberately sets `buffer_edges` below +its own edge count, so the spill/merge path is exercised on every run. S20–S26 +are `#[ignore]` and opt-in. + +## First-fail ladder + +`run_ladder` walks the provisioned rungs in increasing scale and, after each +phase (`generate`, `ingest`, `reopen`, `query`), compares peak RSS / disk / +elapsed time against the envelope. On the **first** violation it records the +failing phase and `error_class` (`oom` | `disk_exhaustion` | `timeout`) and +stops — no larger rung is attempted and no SCALE-26 pass is claimed. + +> RSS fidelity: peak RSS is read from `/proc/self/status` `VmHWM` on Linux and +> falls back to sampled `ps` RSS on macOS (not a kernel high-water mark). Run +> provisioned certification rungs on **Linux**. + +## Commands + +Always-on CI (SCALE-10 smoke + all reconciliation / determinism / bounded / +first-fail unit tests): + +```bash +cargo test -p graphforge-api --test scale_g500_ladder +``` + +Provisioned full ladder (long; isolate the target dir; Linux scale-host): + +```bash +CARGO_TARGET_DIR=/tmp/cargo-g500-ladder make bench-g500-ladder +# override the evidence path: +GF_G500_LADDER_EVIDENCE_OUT=build/g500-ladder-evidence.json make bench-g500-ladder +``` + +Default evidence path: `docs/development/g500-ladder-evidence.json`. + +## Evidence JSON + +One object per attempted rung (schema +`graphforge-billion-edge-ladder-evidence/1`), carrying: + +- `counts`: `raw_attempts`, `self_loops_rejected`, `duplicates_rejected`, + `live_unique_edges`, and the reconciliation identity. +- `persisted`: reopened `node_count` / `edge_count` (must equal + `live_unique_edges`). +- `first_failing_phase`, `error_class`, `pass`, `reconciles`. +- `input_fingerprint` (deterministic SHA-256 of the sorted live edge set), + `rss_peak_bytes`, `disk_used_bytes`, `wall_time_s`, per-phase `steps`. +- `machine_envelope` (128 GiB / 1 TiB / 24 h), `sut`, `generator`. +- `track` and `teps` are always `null`. + +Wall-clock and RSS numbers are hardware-specific observations, never CI +millisecond gates. + +## CI placement + +Per the [Scale Evaluation](../reference/scale-evaluation.md) contract, large +Graph500 runs **must not** be wired into normal GitHub Actions. Only the S10 CI +rung runs in `test.yml`. The provisioned ladder is Make-only until an approved +dedicated runner / evidence job exists (tracked with #745 certification). diff --git a/docs/reference/scale-evaluation.md b/docs/reference/scale-evaluation.md index 49777445..8e93bac7 100644 --- a/docs/reference/scale-evaluation.md +++ b/docs/reference/scale-evaluation.md @@ -267,7 +267,7 @@ until published. | Progressive / first-fail stop + evidence artifacts (Official track) | Yes | Spec + issue links for product claims | | Dedicated runners / disk budgets | Yes | No | | Normal GitHub Actions CI for Graph500 Toy+ / LDBC SF≥1 | No | Must not | -| Thin reference clients | May call | Optional only; no bulk generators. In-tree Official-parameter client: [perf-g500-scale20.md](../development/perf-g500-scale20.md) (SCALE-6 CI smoke + ignored SCALE-20; not `track: official`) | +| Thin reference clients | May call | Optional only; no bulk generators. In-tree Official-parameter client: [perf-g500-scale20.md](../development/perf-g500-scale20.md) (SCALE-6 CI smoke + ignored SCALE-20; not `track: official`) and bounded first-fail ladder [perf-g500-ladder.md](../development/perf-g500-ladder.md) (#736; SCALE-10 CI + ignored SCALE-20→26) | | Chunked ingest / CSR / Cypher via GraphForge APIs | Invokes published APIs | Engine + thin bindings | ### Expected inputs @@ -391,11 +391,26 @@ silently mix. `track: "official"`, and `teps` stays null. SCALE-6 is CI; SCALE-20 is `make bench-g500-scale20` only. +### In-tree bounded billion-edge scale ladder (#736) + +[perf-g500-ladder.md](../development/perf-g500-ladder.md) extends the +Official-parameter client with a versioned, **bounded-memory** ladder +(SCALE-20 → SCALE-26). Unlike the SCALE-20 client it does **not** retain raw +tuples in memory: it spills sorted runs and k-way merges, so peak resident +edges are independent of total edge count. Every attempted rung reconciles +`raw_attempts == live_unique_edges + self_loops_rejected + duplicates_rejected`, +and the ladder stops at the **first** envelope (RSS / disk / time) violation +rather than making an unsupported SCALE-26 claim. Host envelope is 128 GiB RSS / +1 TiB NVMe / 24 h. Still **not** Official-track and **not** TEPS, and it does +**not** certify one billion live edges (that is #745). SCALE-10 is CI; larger +rungs are `make bench-g500-ladder` only. + --- ## Further reading - [Official-parameter SCALE-20 client](../development/perf-g500-scale20.md) — public-facade engineering green (not Official-track) +- [Bounded billion-edge scale ladder](../development/perf-g500-ladder.md) — M5 #736 first-fail contract (bounded memory, not a billion-edge claim) - [Graph Scale Index](graph-scale-index.md) — size axis (node band + density) - [Scale Limits](scale-limits.md) — product envelopes; disk-limited DataFusion framing - [LDBC full suite](../guide/datasets/ldbc.md) — SNB, Graphalytics, FinBench, SPB From 820f99e1c0a8e015cb8ac448b93fe2ef7cc33087 Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Tue, 18 Aug 2026 14:51:17 -0600 Subject: [PATCH 2/6] test(scale): register Bazel target and harden ladder evidence integrity (#736) - Register the scale_g500_ladder integration test in BUILD.bazel (fixture as compile_data) and the Bazel migration ledger (cargo_target_count 101->102), fixing the cargo/bazel parity gate. Load the profile via include_str! so the runner is hermetic under Bazel. CodeRabbit review follow-ups (evidence integrity): - Enforce the 24 h host envelope across the whole ladder via a shared clock, not per rung. - Derive step and rung verdicts from measurement: generate/ingest step pass come from the envelope check, hop-query results fold into the rung verdict, and reconciles is a tri-state (null until ingest runs) instead of forced true. - Label RSS source (vmhwm high-water vs ps_sampled lower bound) in evidence and record per-phase rss_peak_bytes. - Document that ingest-phase RSS reflects upstream bulk-publication identity-set cost, not a generator regression (upstream storage work is a non-goal, #745). Co-Authored-By: Claude Opus 4.8 --- crates/graphforge-api/BUILD.bazel | 12 ++ .../graphforge-api/tests/scale_g500_ladder.rs | 114 +++++++++++++----- docs/development/perf-g500-ladder.md | 14 ++- tools/bazel/parity/migration_target_map.json | 12 +- 4 files changed, 117 insertions(+), 35 deletions(-) diff --git a/crates/graphforge-api/BUILD.bazel b/crates/graphforge-api/BUILD.bazel index f6ae3067..edd51c25 100644 --- a/crates/graphforge-api/BUILD.bazel +++ b/crates/graphforge-api/BUILD.bazel @@ -186,6 +186,17 @@ gf_rust_integration_test( deps = _API_DEPS, ) +gf_rust_integration_test( + name = "scale_g500_ladder", + srcs = ["tests/scale_g500_ladder.rs"], + compile_data = ["tests/fixtures/scale_g500_ladder.v1.json"], + crate = ":graphforge_api", + data = _API_TEST_DATA, + size = "large", + timeout = "long", + deps = _API_DEPS, +) + gf_rust_integration_test( name = "adjacency_scale_evidence", srcs = ["tests/adjacency_scale_evidence.rs"], @@ -504,6 +515,7 @@ test_suite( ":public_facade_remaining_conformance", ":public_lifecycle_conformance", ":release_load_construction", + ":scale_g500_ladder", ":scale_g500_scale20", ":strict_runtime_properties", ":value_access_semantics", diff --git a/crates/graphforge-api/tests/scale_g500_ladder.rs b/crates/graphforge-api/tests/scale_g500_ladder.rs index 6e88ddbc..a7d52352 100644 --- a/crates/graphforge-api/tests/scale_g500_ladder.rs +++ b/crates/graphforge-api/tests/scale_g500_ladder.rs @@ -96,13 +96,12 @@ struct Rung { note: Option, } -fn profile_path() -> PathBuf { - PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/scale_g500_ladder.v1.json") -} +/// The profile is embedded at compile time so the runner is hermetic under both +/// Cargo and Bazel (no runtime `CARGO_MANIFEST_DIR` path dependency). +const PROFILE_JSON: &str = include_str!("fixtures/scale_g500_ladder.v1.json"); fn load_profile() -> ScaleProfile { - let raw = fs::read_to_string(profile_path()).expect("read ladder profile fixture"); - serde_json::from_str(&raw).expect("parse ladder profile fixture") + serde_json::from_str(PROFILE_JSON).expect("parse ladder profile fixture") } // --------------------------------------------------------------------------- @@ -359,32 +358,40 @@ struct RungOutcome { } /// Check the envelope after a phase. Returns `Some(error_class)` on the first -/// violation so the caller can stop the ladder. +/// violation so the caller can stop the ladder. `ladder_started` is the +/// ladder-level clock so the 24 h host envelope bounds the whole run, not each +/// rung independently. fn envelope_violation( env: &RunEnvelope, - started: Instant, + ladder_started: Instant, project: &Path, spill: &Path, ) -> Option<&'static str> { - if peak_rss_bytes().is_some_and(|rss| rss > env.rss_bytes) { + if peak_rss().is_some_and(|(rss, _)| rss > env.rss_bytes) { return Some("oom"); } let disk = directory_bytes(project).unwrap_or(0) + directory_bytes(spill).unwrap_or(0); if disk > env.disk_bytes { return Some("disk_exhaustion"); } - if started.elapsed().as_secs() > env.timeout_s { + if ladder_started.elapsed().as_secs() > env.timeout_s { return Some("timeout"); } None } +/// Current peak RSS as a JSON value (bytes or null). +fn rss_value() -> Value { + peak_rss().map_or(Value::Null, |(bytes, _)| json!(bytes)) +} + #[allow(clippy::too_many_lines)] fn run_rung( profile: &ScaleProfile, rung: &Rung, env: RunEnvelope, edge_factor: u32, + ladder_started: Instant, ) -> RungOutcome { let started = Instant::now(); let workspace = TempDir::new().expect("rung workspace"); @@ -408,10 +415,16 @@ fn run_rung( &spill_dir, ); let generate_s = gen_started.elapsed().as_secs_f64(); + let gen_violation = envelope_violation(&env, ladder_started, &project, &spill_dir); + if let Some(class) = gen_violation { + first_failing_phase = Some("generate"); + error_class = Some(class); + } steps.push(json!({ "id": "generate", - "pass": true, + "pass": gen_violation.is_none(), "wall_time_s": generate_s, + "rss_peak_bytes": rss_value(), "detail": { "raw_attempts": spill.raw_attempts, "self_loops_rejected": spill.self_loops_rejected, @@ -420,15 +433,12 @@ fn run_rung( "run_count": spill.runs.len(), } })); - if let Some(class) = envelope_violation(&env, started, &project, &spill_dir) { - first_failing_phase = Some("generate"); - error_class = Some(class); - } // ---- ingest (merge + publish through the public facade) ---- let mut live_unique_edges = 0u64; let mut duplicates_rejected = 0u64; let mut input_fingerprint = String::from("sha256:"); + let mut ingest_ran = false; if first_failing_phase.is_none() { let ingest_started = Instant::now(); let graph = GraphForge::new(Some(project.to_str().expect("utf8 project"))) @@ -441,21 +451,27 @@ fn run_rung( duplicates_rejected = merge.duplicates_rejected; input_fingerprint = format!("sha256:{}", hex_encode(sink.finish())); drop(graph); + ingest_ran = true; let ingest_s = ingest_started.elapsed().as_secs_f64(); + let ingest_violation = envelope_violation(&env, ladder_started, &project, &spill_dir); + if let Some(class) = ingest_violation { + first_failing_phase = Some("ingest"); + error_class = Some(class); + } steps.push(json!({ "id": "ingest", - "pass": true, + "pass": ingest_violation.is_none(), "wall_time_s": ingest_s, + // NOTE: ingest RSS includes the upstream bulk-publication identity + // set (see runbook), so an ingest-phase `oom` is not a generator + // bottleneck. Recorded here so consumers can attribute it. + "rss_peak_bytes": rss_value(), "detail": { "live_unique_edges": live_unique_edges, "duplicates_rejected": duplicates_rejected, "input_fingerprint": input_fingerprint, } })); - if let Some(class) = envelope_violation(&env, started, &project, &spill_dir) { - first_failing_phase = Some("ingest"); - error_class = Some(class); - } } // ---- reopen + recount ---- @@ -474,6 +490,7 @@ fn run_rung( "id": "reopen", "pass": true, "wall_time_s": reopen_s, + "rss_peak_bytes": rss_value(), "detail": { "node_count": node_count, "edge_count": edge_count, "gsi": gsi } })); @@ -498,7 +515,7 @@ fn run_rung( "detail": { "rows": hop2_rows } })); drop(graph); - if let Some(class) = envelope_violation(&env, started, &project, &spill_dir) { + if let Some(class) = envelope_violation(&env, ladder_started, &project, &spill_dir) { first_failing_phase = Some("query"); error_class = Some(class); } @@ -506,10 +523,20 @@ fn run_rung( let disk_used_bytes = directory_bytes(&project).unwrap_or(0) + directory_bytes(&spill_dir).unwrap_or(0); - let reconciles = first_failing_phase.is_some() - || spill.raw_attempts - == live_unique_edges + spill.self_loops_rejected + duplicates_rejected; - let passed = first_failing_phase.is_none() && reconciles; + // Tri-state: reconciliation is only *evaluated* once ingest has run. A rung + // stopped in the generate phase is reported as null (not evaluated), never + // as a forced `true`. + let reconciles: Option = ingest_ran.then(|| { + spill.raw_attempts == live_unique_edges + spill.self_loops_rejected + duplicates_rejected + }); + let all_steps_pass = steps + .iter() + .all(|step| step["pass"].as_bool().unwrap_or(false)); + let passed = first_failing_phase.is_none() && all_steps_pass && reconciles == Some(true); + let (rss_peak_bytes, rss_source) = match peak_rss() { + Some((bytes, source)) => (json!(bytes), json!(source)), + None => (Value::Null, Value::Null), + }; let evidence = json!({ "schema": EVIDENCE_SCHEMA, @@ -525,6 +552,7 @@ fn run_rung( "first_failing_phase": first_failing_phase, "error_class": error_class, "reconciles": reconciles, + "reconciliation_evaluated": reconciles.is_some(), "counts": { "raw_attempts": spill.raw_attempts, "self_loops_rejected": spill.self_loops_rejected, @@ -535,7 +563,8 @@ fn run_rung( "persisted": { "node_count": node_count, "edge_count": edge_count }, "input_fingerprint": input_fingerprint, "wall_time_s": started.elapsed().as_secs_f64(), - "rss_peak_bytes": peak_rss_bytes(), + "rss_peak_bytes": rss_peak_bytes, + "rss_source": rss_source, "disk_used_bytes": disk_used_bytes, "machine_envelope": { "rss_bytes": env.rss_bytes, @@ -563,9 +592,10 @@ fn run_rung( /// Drive the ladder rung-by-rung, stopping at the first failing rung. fn run_ladder(profile: &ScaleProfile, env: RunEnvelope, rungs: &[Rung]) -> Vec { + let ladder_started = Instant::now(); let mut evidence = Vec::new(); for rung in rungs { - let outcome = run_rung(profile, rung, env, profile.edgefactor); + let outcome = run_rung(profile, rung, env, profile.edgefactor, ladder_started); let passed = outcome.passed; evidence.push(outcome.evidence); if !passed { @@ -805,7 +835,11 @@ fn directory_bytes(path: &Path) -> std::io::Result { Ok(total) } -fn peak_rss_bytes() -> Option { +/// Returns `(bytes, source)`. `"vmhwm"` (Linux `/proc/self/status`) is a true +/// high-water mark; `"ps_sampled"` (fallback) is the instantaneous RSS at the +/// moment of the call, i.e. a **lower bound** on the real peak. Consumers must +/// treat a `ps_sampled` value as a floor, not a ceiling. +fn peak_rss() -> Option<(u64, &'static str)> { if let Ok(contents) = fs::read_to_string("/proc/self/status") { for line in contents.lines() { if let Some(value) = line.strip_prefix("VmHWM:") { @@ -815,7 +849,7 @@ fn peak_rss_bytes() -> Option { .trim() .parse::() .ok()?; - return Some(kb.saturating_mul(1024)); + return Some((kb.saturating_mul(1024), "vmhwm")); } } } @@ -830,7 +864,7 @@ fn peak_rss_bytes() -> Option { .trim() .parse::() .ok()?; - Some(kb.saturating_mul(1024)) + Some((kb.saturating_mul(1024), "ps_sampled")) } struct SplitMix64(u64); @@ -1056,10 +1090,19 @@ fn first_fail_stops_at_envelope_violation() { .find(|r| r.tier == "ci") .cloned() .expect("ci rung"); - let outcome = run_rung(&profile, &ci_rung, tiny_env, profile.edgefactor); + let outcome = run_rung( + &profile, + &ci_rung, + tiny_env, + profile.edgefactor, + Instant::now(), + ); assert!(!outcome.passed, "a violated envelope must not pass"); assert_eq!(outcome.evidence["first_failing_phase"], "generate"); assert_eq!(outcome.evidence["error_class"], "oom"); + // Reconciliation was never evaluated (stopped before ingest): tri-state null. + assert!(outcome.evidence["reconciles"].is_null()); + assert_eq!(outcome.evidence["reconciliation_evaluated"], false); assert!(outcome.evidence["teps"].is_null()); assert!( outcome.evidence["scale"].as_u64().unwrap() < 26, @@ -1083,6 +1126,7 @@ fn ci_rung_public_facade_engineering_green() { &ci_rung, profile.envelope.into(), profile.edgefactor, + Instant::now(), ); assert!(outcome.passed, "CI rung must pass: {:#}", outcome.evidence); let ev = &outcome.evidence; @@ -1140,8 +1184,14 @@ fn ladder_public_facade_first_fail_evidence() { ) .expect("write ladder evidence"); - // Every attempted rung reconciles; the ladder stops at the first failure. + // Every rung that reached ingest reconciles; a rung stopped in the generate + // phase reports reconciles=null (not evaluated). The ladder stops at the + // first failure. for rung in &evidence { - assert_eq!(rung["reconciles"], true, "attempted rung must reconcile"); + let rec = &rung["reconciles"]; + assert!( + rec.is_null() || rec == &Value::Bool(true), + "an evaluated rung must reconcile; got {rec}" + ); } } diff --git a/docs/development/perf-g500-ladder.md b/docs/development/perf-g500-ladder.md index cd47700d..673bc1f5 100644 --- a/docs/development/perf-g500-ladder.md +++ b/docs/development/perf-g500-ladder.md @@ -77,10 +77,20 @@ elapsed time against the envelope. On the **first** violation it records the failing phase and `error_class` (`oom` | `disk_exhaustion` | `timeout`) and stops — no larger rung is attempted and no SCALE-26 pass is claimed. -> RSS fidelity: peak RSS is read from `/proc/self/status` `VmHWM` on Linux and -> falls back to sampled `ps` RSS on macOS (not a kernel high-water mark). Run +> RSS fidelity: peak RSS is read from `/proc/self/status` `VmHWM` on Linux +> (a true high-water mark) and falls back to sampled `ps` RSS otherwise (an +> *instantaneous lower bound*, not a peak). Evidence records `rss_source` +> (`vmhwm` | `ps_sampled`) so a `ps_sampled` value is read as a floor. Run > provisioned certification rungs on **Linux**. +> Ingest-phase attribution: `publish_bulk_nodes` / `publish_bulk_edges` build an +> in-memory set of existing identities per call (`bulk_construction.rs`), so RSS +> during the `ingest` phase grows with the persisted graph, not just the +> generator buffer. Per-phase `rss_peak_bytes` is recorded on each step, and an +> `oom` with `first_failing_phase: "ingest"` reflects that upstream +> bulk-publication cost — **not** a generator-memory regression. Reducing that +> cost is upstream storage work (a non-goal here), tracked toward #745. + ## Commands Always-on CI (SCALE-10 smoke + all reconciliation / determinism / bounded / diff --git a/tools/bazel/parity/migration_target_map.json b/tools/bazel/parity/migration_target_map.json index 7267e303..4a9f078f 100644 --- a/tools/bazel/parity/migration_target_map.json +++ b/tools/bazel/parity/migration_target_map.json @@ -1,7 +1,7 @@ { "schema": "graphforge.bazel-migration-target-map.v1", "issue": 6, - "cargo_target_count": 101, + "cargo_target_count": 102, "targets": [ { "package": "graphforge-api", @@ -253,6 +253,16 @@ "exception_id": null, "notes": "#710 Official-parameter SCALE-20 engineering green; SCALE-6 CI smoke" }, + { + "package": "graphforge-api", + "target": "scale_g500_ladder", + "class": "integration-test", + "source": "crates/graphforge-api/tests/scale_g500_ladder.rs", + "status": "mapped", + "bazel_label": "//crates/graphforge-api:scale_g500_ladder", + "exception_id": null, + "notes": "#736 bounded billion-edge scale ladder + first-fail contract; SCALE-10 CI" + }, { "package": "graphforge-api", "target": "provider_public_surface", From 19035c88726d744982eafa0359fd684dc8e32bc3 Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Tue, 18 Aug 2026 15:02:21 -0600 Subject: [PATCH 3/6] fix(ci): count both merged Cargo targets in the Bazel ledger (#736) Merging main brought filesystem_admission alongside scale_g500_ladder, so CI's PR merge commit saw 103 Cargo targets against a map of 102. Co-authored-by: Cursor --- docs/development/bazel-migration-ledger.md | 1 + tools/bazel/parity/migration_target_map.json | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/development/bazel-migration-ledger.md b/docs/development/bazel-migration-ledger.md index 3a5db7b8..dfc48bd5 100644 --- a/docs/development/bazel-migration-ledger.md +++ b/docs/development/bazel-migration-ledger.md @@ -85,6 +85,7 @@ Authoritative machine-readable map: `tools/bazel/parity/migration_target_map.jso | `graphforge-api` | `algorithm_public_surface` | `integration-test` | `crates/graphforge-api/tests/algorithm_public_surface.rs` | `//crates/graphforge-api:algorithm_public_surface` | `mapped` | #8 | | `graphforge-api` | `search_public_surface` | `integration-test` | `crates/graphforge-api/tests/search_public_surface.rs` | `//crates/graphforge-api:search_public_surface` | `mapped` | #8 | | `graphforge-api` | `scale_g500_scale20` | `integration-test` | `crates/graphforge-api/tests/scale_g500_scale20.rs` | `//crates/graphforge-api:scale_g500_scale20` | `mapped` | #710 | +| `graphforge-api` | `scale_g500_ladder` | `integration-test` | `crates/graphforge-api/tests/scale_g500_ladder.rs` | `//crates/graphforge-api:scale_g500_ladder` | `mapped` | #736 | | `graphforge-api` | `provider_public_surface` | `integration-test` | `crates/graphforge-api/tests/provider_public_surface.rs` | `//crates/graphforge-api:provider_public_surface` | `mapped` | #8 | | `graphforge-api` | `m4_entry_baseline` | `integration-test` | `crates/graphforge-api/tests/m4_entry_baseline.rs` | `//crates/graphforge-api:m4_entry_baseline` | `mapped` | #8 | | `graphforge-api` | `file_backed_graph_generation` | `integration-test` | `crates/graphforge-api/tests/file_backed_graph_generation.rs` | `//crates/graphforge-api:file_backed_graph_generation` | `mapped` | #338 | diff --git a/tools/bazel/parity/migration_target_map.json b/tools/bazel/parity/migration_target_map.json index 55ca6e23..6066b129 100644 --- a/tools/bazel/parity/migration_target_map.json +++ b/tools/bazel/parity/migration_target_map.json @@ -1,7 +1,7 @@ { "schema": "graphforge.bazel-migration-target-map.v1", "issue": 6, - "cargo_target_count": 102, + "cargo_target_count": 103, "targets": [ { "package": "graphforge-api", From 4aef1fd4242bc270a6992b78806f689e200db7b9 Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Tue, 18 Aug 2026 16:28:31 -0600 Subject: [PATCH 4/6] fix(storage): give concurrent CURRENT publishers unique atomic temps Optimistic Node workers failed CI with ENOENT "file was not replaced" because publish_atomic_bytes hashed only the target name, so two CURRENT publishers shared one temp. Required to unblock #788; the race is on main's publication path. Co-authored-by: Cursor --- .../src/project_publication.rs | 167 +++++++++++------- 1 file changed, 103 insertions(+), 64 deletions(-) diff --git a/crates/graphforge-storage/src/project_publication.rs b/crates/graphforge-storage/src/project_publication.rs index b9b969b3..58110c29 100644 --- a/crates/graphforge-storage/src/project_publication.rs +++ b/crates/graphforge-storage/src/project_publication.rs @@ -2049,69 +2049,70 @@ pub(crate) fn publish_atomic_bytes( let target_text = target_name .to_str() .ok_or_else(|| std::io::Error::other("atomic publication target is not UTF-8"))?; - let digest: [u8; 32] = Sha256::digest(target_text.as_bytes()).into(); - let temp_name = format!(".graphforge-atomic-{}.tmp", hex_digest(digest)); + // Hash the target name plus a per-attempt identity. Hashing only the + // target made concurrent CURRENT publishers share one temp, so one + // writer's `create_new` prep deleted the other's in-flight file and + // `replace_file` failed with ENOENT ("file was not replaced"). + let temp_name = unique_atomic_temp_name(target_text); let temp_path = parent.join(&temp_name); - remove_stale_atomic_temp(&temp_path)?; - let mut temp = OpenOptions::new() - .write(true) - .create_new(true) - .open(&temp_path)?; - temp.write_all(bytes)?; - after_write()?; - temp.sync_all()?; - after_sync()?; - drop(temp); - before_replace()?; - - let directory = crate::filesystem_admission::open_directory_handle(parent)?; - match std::fs::symlink_metadata(path) { - Ok(metadata) => { - if !metadata.is_file() - || metadata.file_type().is_symlink() - || graphforge_filesystem::path_link_count(path)? != 1 - { - return Err(std::io::Error::other( - "atomic publication target is not a regular single-link file", + let publish = || -> Result<(), AtomicPublishError> { + let mut temp = OpenOptions::new() + .write(true) + .create_new(true) + .open(&temp_path)?; + temp.write_all(bytes)?; + after_write()?; + temp.sync_all()?; + after_sync()?; + drop(temp); + before_replace()?; + + let directory = crate::filesystem_admission::open_directory_handle(parent)?; + match std::fs::symlink_metadata(path) { + Ok(metadata) => { + if !metadata.is_file() + || metadata.file_type().is_symlink() + || graphforge_filesystem::path_link_count(path)? != 1 + { + return Err(std::io::Error::other( + "atomic publication target is not a regular single-link file", + ) + .into()); + } + graphforge_filesystem::replace_file( + &directory, + std::ffi::OsStr::new(&temp_name), + target_name, ) - .into()); + .map_err(AtomicPublishError::Replacement) } - graphforge_filesystem::replace_file( - &directory, - std::ffi::OsStr::new(&temp_name), - target_name, - ) - .map_err(AtomicPublishError::Replacement) - } - Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - graphforge_filesystem::install_new_file( - &directory, - std::ffi::OsStr::new(&temp_name), - target_name, - ) - .map_err(AtomicPublishError::Io) + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + graphforge_filesystem::install_new_file( + &directory, + std::ffi::OsStr::new(&temp_name), + target_name, + ) + .map_err(AtomicPublishError::Io) + } + Err(error) => Err(AtomicPublishError::Io(error)), } - Err(error) => Err(AtomicPublishError::Io(error)), + }; + let result = publish(); + if result.is_err() { + let _ = std::fs::remove_file(&temp_path); } + result } -fn remove_stale_atomic_temp(path: &Path) -> std::io::Result<()> { - match std::fs::symlink_metadata(path) { - Ok(metadata) => { - if !metadata.is_file() - || metadata.file_type().is_symlink() - || graphforge_filesystem::path_link_count(path)? != 1 - { - return Err(std::io::Error::other( - "atomic publication temporary is not a regular single-link file", - )); - } - std::fs::remove_file(path) - } - Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), - Err(error) => Err(error), - } +fn unique_atomic_temp_name(target_text: &str) -> String { + let mut hasher = Sha256::new(); + hasher.update(target_text.as_bytes()); + hasher.update(Uuid::now_v7().as_bytes()); + format!( + ".graphforge-atomic-{}.tmp", + hex_digest(hasher.finalize().into()) + ) } pub(crate) fn read_journal(path: &Path) -> Result { @@ -3325,15 +3326,53 @@ mod tests { assert_eq!(std::fs::read(&target).unwrap(), b"first\n"); publish_atomic_bytes(&target, b"second\n", || Ok(()), || Ok(()), || Ok(())).unwrap(); assert_eq!(std::fs::read(&target).unwrap(), b"second\n"); - assert!( - std::fs::read_dir(root.path()) - .unwrap() - .filter_map(Result::ok) - .all(|entry| !entry - .file_name() - .to_string_lossy() - .starts_with(".graphforge-atomic-")) - ); + assert!(atomic_temp_names(root.path()).is_empty()); + } + + #[test] + fn concurrent_atomic_replace_of_the_same_target_does_not_share_a_temp() { + use std::sync::{Arc, Barrier}; + + let root = tempfile::tempdir().unwrap(); + let target = root.path().join("CURRENT"); + publish_atomic_bytes(&target, b"seed\n", || Ok(()), || Ok(()), || Ok(())).unwrap(); + + for _ in 0..32 { + let barrier = Arc::new(Barrier::new(2)); + let mut joins = Vec::new(); + for payload in [b"left\n" as &[u8], b"right\n"] { + let path = target.clone(); + let barrier = Arc::clone(&barrier); + joins.push(std::thread::spawn(move || { + barrier.wait(); + publish_atomic_bytes(&path, payload, || Ok(()), || Ok(()), || Ok(())) + })); + } + let results: Vec<_> = joins + .into_iter() + .map(|thread| thread.join().expect("publisher thread")) + .collect(); + assert!( + results.iter().all(Result::is_ok), + "concurrent CURRENT replace must not fail with a shared temp: {results:?}" + ); + let published = std::fs::read(&target).unwrap(); + assert!( + published == b"left\n" || published == b"right\n", + "{}", + String::from_utf8_lossy(&published) + ); + } + assert!(atomic_temp_names(root.path()).is_empty()); + } + + fn atomic_temp_names(root: &Path) -> Vec { + std::fs::read_dir(root) + .unwrap() + .filter_map(Result::ok) + .map(|entry| entry.file_name().to_string_lossy().into_owned()) + .filter(|name| name.starts_with(".graphforge-atomic-")) + .collect() } #[test] From b23516c197beba4da5e582a10689d27d96efa6cb Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Tue, 18 Aug 2026 16:44:59 -0600 Subject: [PATCH 5/6] fix(storage): drop racy CURRENT lstat before atomic replace Concurrent publishers already use unique temps; a second path_link_count on CURRENT still failed Bazel when nlink changed mid-replace. Let the native replace primitive verify the open target. Co-authored-by: Cursor --- .../src/project_publication.rs | 26 +++++++------------ 1 file changed, 9 insertions(+), 17 deletions(-) diff --git a/crates/graphforge-storage/src/project_publication.rs b/crates/graphforge-storage/src/project_publication.rs index 58110c29..b859e97f 100644 --- a/crates/graphforge-storage/src/project_publication.rs +++ b/crates/graphforge-storage/src/project_publication.rs @@ -2069,24 +2069,16 @@ pub(crate) fn publish_atomic_bytes( before_replace()?; let directory = crate::filesystem_admission::open_directory_handle(parent)?; + // Existence is one snapshot; `replace_file` / `install_new_file` verify + // regular single-link identity on the open handles. A second path lstat + // here raced concurrent CURRENT replace (nlink != 1) and failed Bazel. match std::fs::symlink_metadata(path) { - Ok(metadata) => { - if !metadata.is_file() - || metadata.file_type().is_symlink() - || graphforge_filesystem::path_link_count(path)? != 1 - { - return Err(std::io::Error::other( - "atomic publication target is not a regular single-link file", - ) - .into()); - } - graphforge_filesystem::replace_file( - &directory, - std::ffi::OsStr::new(&temp_name), - target_name, - ) - .map_err(AtomicPublishError::Replacement) - } + Ok(_) => graphforge_filesystem::replace_file( + &directory, + std::ffi::OsStr::new(&temp_name), + target_name, + ) + .map_err(AtomicPublishError::Replacement), Err(error) if error.kind() == std::io::ErrorKind::NotFound => { graphforge_filesystem::install_new_file( &directory, From e6a924ba735ef5a14e771f381cdb62eb98f28adc Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Tue, 18 Aug 2026 17:02:17 -0600 Subject: [PATCH 6/6] fix(storage): serialize in-process CURRENT replace per target Unique temps stopped the shared-name ENOENT, but concurrent replace_file still returned StateUnknown when a peer won the rename. Lock the namespace mutation per target so same-process publishers both complete. Co-authored-by: Cursor --- .../src/project_publication.rs | 27 +++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/crates/graphforge-storage/src/project_publication.rs b/crates/graphforge-storage/src/project_publication.rs index b859e97f..aa036cad 100644 --- a/crates/graphforge-storage/src/project_publication.rs +++ b/crates/graphforge-storage/src/project_publication.rs @@ -4,9 +4,11 @@ //! domain semantics remain in their owning crates and enter validation through //! opaque callbacks. +use std::collections::HashMap; use std::fs::{File, OpenOptions}; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, OnceLock}; use graphforge_core::{GfError, ProjectErrorCode}; use serde::{Deserialize, Serialize}; @@ -2068,10 +2070,18 @@ pub(crate) fn publish_atomic_bytes( drop(temp); before_replace()?; + // Concurrent same-process replace of one target must not overlap + // `replace_file`: the loser's post-rename identity check sees the + // winner's inode and returns StateUnknown. Unique temps already + // prevent the shared-name ENOENT; this lock serializes the rename. + let namespace_lock = lock_atomic_publish_target(path); + let _namespace_guard = namespace_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let directory = crate::filesystem_admission::open_directory_handle(parent)?; // Existence is one snapshot; `replace_file` / `install_new_file` verify - // regular single-link identity on the open handles. A second path lstat - // here raced concurrent CURRENT replace (nlink != 1) and failed Bazel. + // regular single-link identity on the open handles. match std::fs::symlink_metadata(path) { Ok(_) => graphforge_filesystem::replace_file( &directory, @@ -2107,6 +2117,19 @@ fn unique_atomic_temp_name(target_text: &str) -> String { ) } +fn lock_atomic_publish_target(path: &Path) -> Arc> { + static LOCKS: OnceLock>>>> = OnceLock::new(); + let mut locks = LOCKS + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Arc::clone( + locks + .entry(path.to_path_buf()) + .or_insert_with(|| Arc::new(Mutex::new(()))), + ) +} + pub(crate) fn read_journal(path: &Path) -> Result { let metadata = std::fs::symlink_metadata(path).map_err(publication_io)?; if !metadata.is_file()