diff --git a/crates/graphforge-exec/src/algorithm_analyze.rs b/crates/graphforge-exec/src/algorithm_analyze.rs index 5b9d3570e..4bdf5cc4a 100644 --- a/crates/graphforge-exec/src/algorithm_analyze.rs +++ b/crates/graphforge-exec/src/algorithm_analyze.rs @@ -3981,6 +3981,26 @@ mod tests { ) } + fn execute_with_compute_threads( + graph: &AdjacencyGraph, + algorithm: AnalyzeAlgorithm, + directed: bool, + threads: usize, + ) -> Result { + let mut registry = AlgorithmRegistry::default(); + register_analyze_algorithms(&mut registry, directed)?; + let control = AlgorithmControl::new( + AlgorithmLimits::default().with_compute_threads(threads), + AlgorithmCancellation::default(), + ) + .with_compute_pool(Arc::new(crate::ComputePool::new(threads).unwrap())); + registry.execute(Algorithm::Analyze(algorithm), graph, &control) + } + + fn output_fingerprint(output: &AlgorithmOutput) -> String { + format!("{:?}|{:?}", output.schema, output.rows()) + } + fn execute_minimum_k_spanning_tree( graph: &AdjacencyGraph, k: usize, @@ -6945,6 +6965,54 @@ mod tests { ); } + #[test] + fn bridges_keeps_serial_fingerprint_under_thread_budgets() { + let graph = AdjacencyGraph::with_test_undirected_multigraph( + 10, + &[ + (10, 0, 1), + (11, 1, 2), + (12, 2, 0), + (13, 1, 3), + (14, 3, 4), + (15, 4, 5), + (16, 5, 3), + (17, 5, 6), + (18, 6, 7), + (19, 7, 5), + (20, 7, 8), + (21, 8, 9), + (22, 7, 8), + ], + ); + let serial = + execute_with_compute_threads(&graph, AnalyzeAlgorithm::Bridges, false, 1).unwrap(); + assert_eq!( + serial.rows(), + [ + vec![ + AlgorithmValue::Uuid(13_u128.to_be_bytes()), + AlgorithmValue::Uuid(1_u128.to_be_bytes()), + AlgorithmValue::Uuid(3_u128.to_be_bytes()), + ], + vec![ + AlgorithmValue::Uuid(21_u128.to_be_bytes()), + AlgorithmValue::Uuid(8_u128.to_be_bytes()), + AlgorithmValue::Uuid(9_u128.to_be_bytes()), + ], + ] + ); + let serial_fingerprint = output_fingerprint(&serial); + + for threads in [2_usize, 4, 8] { + let output = + execute_with_compute_threads(&graph, AnalyzeAlgorithm::Bridges, false, threads) + .unwrap(); + assert_eq!(output.schema, serial.schema); + assert_eq!(output_fingerprint(&output), serial_fingerprint); + } + } + #[test] fn graphsage_source_resource_contract_validates_feature_matrix() { let empty = AdjacencyGraph::with_test_counts(0, 0); diff --git a/docs/development/execution-resource-policy.md b/docs/development/execution-resource-policy.md index 81db49f25..3a6d8641a 100644 --- a/docs/development/execution-resource-policy.md +++ b/docs/development/execution-resource-policy.md @@ -262,6 +262,21 @@ and fingerprints match the one-thread result at `1`/`2`/`4`/`8`/automatic configurations. Cancellation remains cooperative both before worker launch and inside chunk scans. +## Serial bridges (#564) + +`analyze(by="bridges")` is intentionally SERIAL. Bridge classification depends +on the shared low-link DFS traversal: discovery order, parent-edge identity, +multigraph parallel-edge handling, and low-link propagation determine whether an +edge is a bridge. Parallelizing that state would change the canonical edge +evidence or require synchronization around each DFS transition, so there is no +safe independent kernel for the private compute pool. + +The handler keeps Rust-owned adjacency projection and bounded Arrow output, +does not use Rayon's global pool, and preserves cancellation and shared resource +limits. A fingerprint test attaches private compute pools for `1`/`2`/`4`/`8` +configured compute threads and requires identical schemas, bridge ordering, and +rows. + ## Observability `GraphForge::resource_policy()` and `GraphForge::resource_diagnostics()` expose