fix(native): propagate edge/node write failures out of run_pipeline

crates/codegraph-core/src/domain/graph/builder/pipeline.rs

@@ -18,6 +18,14 @@
 //!    that their edges match this revision (#1731)
 //! 8. Structure metrics + role classification
 //! 9. Finalize (metadata, journal)
+//!
+//! Steps 5 and 7 propagate write failures via `?` instead of discarding
+//! them: a transaction that never started (or never committed) for nodes or
+//! edges now aborts `run_pipeline` with `Err`, which `NativeDatabase::build_graph`
+//! turns into a thrown napi error. The JS caller (`tryNativeOrchestrator`)
+//! already catches that and falls back to the JS/WASM pipeline, so a write
+//! failure now triggers a real retry instead of a "successful" build over an
+//! incomplete graph (#1827).
 
 use crate::domain::graph::builder::stages::detect_changes;
 use crate::infrastructure::config::{BuildConfig, BuildOpts, BuildPathAliases};
@@ -523,14 +531,19 @@ pub fn run_pipeline(
     // desyncing file_hashes from the edges it's supposed to gate re-parsing
     // on (#1731). The hash is committed at the end of Stage 7 instead, once
     // edges genuinely match this revision — see `commit_file_hashes` below.
+    // A failure here propagates via `?` instead of being discarded: nodes
+    // are the foundation every later stage builds on, so a transaction
+    // failure must abort the pipeline and surface as a thrown error rather
+    // than a "successful" build with missing data (#1827).
     let t0 = Instant::now();
     let insert_batches = build_insert_batches(&file_symbols);
     let file_hashes = build_file_hash_entries(&parse_changes);
-    let _ = crate::domain::graph::builder::stages::insert_nodes::do_insert_nodes(
+    crate::domain::graph::builder::stages::insert_nodes::do_insert_nodes(
         conn,
         &insert_batches,
         &change_result.removed,
-    );
+    )
+    .map_err(|e| format!("insert_nodes failed: {e}"))?;
     detect_changes::heal_metadata(conn, &change_result.metadata_updates);
     timing.insert_ms = t0.elapsed().as_secs_f64() * 1000.0;
 
@@ -566,9 +579,14 @@ pub fn run_pipeline(
     import_ctx.reexport_map = import_edges::build_reexport_map(&import_ctx);
     import_ctx.barrel_only_files = import_edges::detect_barrel_only_files(&import_ctx);
 
-    // Build import edges
+    // Build import edges. A write failure here (transaction-start, a
+    // malformed chunk, or commit) propagates via `?` instead of being
+    // discarded — the old `run_pipeline` had no way to know edges were
+    // never written for some or all files, so it returned `Ok(...)` (a
+    // "successful" build) over an incomplete edge set (#1827).
     let import_edge_rows = import_edges::build_import_edges(conn, &import_ctx);
-    import_edges::insert_edges(conn, &import_edge_rows);
+    import_edges::insert_edges(conn, &import_edge_rows)
+        .map_err(|e| format!("import edge insertion failed: {e}"))?;
 
     // Phase 8.2: cross-file return-type propagation — seed each file's
     // type_map with the return types of imported functions before call-edge
@@ -578,18 +596,28 @@ pub fn run_pipeline(
     // Build call edges using existing Rust edge_builder (internal path)
     // For now, call edges are built via the existing napi-exported function's
     // internal logic. We load nodes from DB and pass to the edge builder.
+    // Same error-propagation rationale as import edges above (#1827) — this
+    // call used to run unchecked, with its `Result` never captured.
     build_and_insert_call_edges(
         conn,
         &file_symbols,
         &import_ctx,
         !change_result.is_full_build,
         config.analysis.points_to_max_iterations,
-    );
+    )
+    .map_err(|e| format!("call edge insertion failed: {e}"))?;
 
     reconnect_saved_reverse_dep_edges(conn, &saved_reverse_dep_edges);
 
     // Now that edges reflect this revision, commit file_hashes for the
     // changed files (#1731). Deferred from Stage 5 — see the comment there.
+    // Only reached once import and call edges above are confirmed written —
+    // an edge-insertion failure now aborts the pipeline (via `?`) before this
+    // point instead of committing a hash over an incomplete edge set (#1827).
+    // A failure of this commit itself stays non-fatal (log and continue):
+    // it only affects bookkeeping, not correctness — the file's hash simply
+    // keeps its old value, so the next build re-detects and re-processes it
+    // (the same self-healing property #1731 relies on).
     if let Err(e) = crate::domain::graph::builder::stages::insert_nodes::commit_file_hashes(
         conn,
         &file_hashes,
@@ -1505,10 +1533,18 @@ fn inject_return_types_for_file(
     symbols.type_map.extend(injections);
 }
 
-/// Insert the edges produced by the native edge builder into the edges table.
-fn insert_call_edge_rows(conn: &Connection, edges: &[crate::domain::graph::builder::stages::build_edges::ComputedEdge]) {
+/// Insert the edges produced by the native edge builder into the edges
+/// table. Propagates `do_insert_edges`'s `Result` instead of discarding it
+/// (#1827) — `do_insert_edges` already fails fast (transaction-start,
+/// bind/execute, or commit) via `?`, but the previous `let _ = …` here threw
+/// that signal away, so `run_pipeline` had no way to detect a transaction
+/// that never started, or failed to commit, for this file's call edges.
+fn insert_call_edge_rows(
+    conn: &Connection,
+    edges: &[crate::domain::graph::builder::stages::build_edges::ComputedEdge],
+) -> Result<(), String> {
     if edges.is_empty() {
-        return;
+        return Ok(());
     }
     let edge_rows: Vec<crate::db::repository::edges::EdgeRow> = edges
         .iter()
@@ -1521,7 +1557,8 @@ fn insert_call_edge_rows(conn: &Connection, edges: &[crate::domain::graph::build
             dynamic_kind: e.dynamic_kind.clone(),
         })
         .collect();
-    let _ = crate::db::repository::edges::do_insert_edges(conn, &edge_rows);
+    crate::db::repository::edges::do_insert_edges(conn, &edge_rows)
+        .map_err(|e| format!("call edge insertion failed: {e}"))
 }
 
 /// Full builds always load every node — there is no smaller set anyway.
@@ -1534,12 +1571,12 @@ fn build_and_insert_call_edges(
     import_ctx: &ImportEdgeContext,
     is_incremental: bool,
     max_iterations: u32,
-) {
+) -> Result<(), String> {
     use crate::domain::graph::builder::stages::build_edges::*;
 
     let all_nodes = load_edge_node_set(conn, file_symbols, import_ctx, is_incremental);
     if all_nodes.is_empty() {
-        return;
+        return Ok(());
     }
 
     let builtin_receivers = builtin_call_receivers();
@@ -1632,7 +1669,7 @@ fn build_and_insert_call_edges(
     }
 
     let computed_edges = build_call_edges(file_entries, all_nodes, builtin_receivers, max_iterations);
-    insert_call_edge_rows(conn, &computed_edges);
+    insert_call_edge_rows(conn, &computed_edges)
 }
 
 // ── Analysis persistence helpers ─────────────────────────────────────────

crates/codegraph-core/src/domain/graph/builder/stages/import_edges.rs

@@ -557,32 +557,49 @@ const INSERT_CHUNK: usize = 199;
 /// per-edge bind/step/reset cycle. With the chunked path each chunk runs a
 /// single VM execution against a freshly prepared statement (#1013).
 ///
-/// Bind/execute errors are surfaced via a stderr warning and the offending
-/// chunk is skipped — silently swallowing them previously could produce
-/// `NULL` columns in the inserted edge rows.
-pub fn insert_edges(conn: &Connection, edges: &[EdgeRow]) {
+/// Every failure mode — transaction-start, a chunk's bind/execute, and
+/// commit — is both logged to stderr for immediate diagnosis and returned to
+/// the caller as `Err` (#1827). A single malformed chunk still doesn't
+/// sacrifice every other edge in the batch: it's skipped so the remaining
+/// chunks commit, but the `Err` return means `run_pipeline` finds out the
+/// edge set is incomplete instead of reporting a silently "successful"
+/// build — which also keeps `file_hashes` from being advanced over
+/// incomplete data (#1731).
+pub fn insert_edges(conn: &Connection, edges: &[EdgeRow]) -> Result<(), String> {
     if edges.is_empty() {
-        return;
+        return Ok(());
     }
-    let tx = match conn.unchecked_transaction() {
-        Ok(tx) => tx,
-        Err(e) => {
-            eprintln!("[codegraph] insert_edges: failed to start transaction: {e}");
-            return;
-        }
-    };
-
+    let tx = conn.unchecked_transaction().map_err(|e| {
+        let msg = format!("insert_edges: failed to start transaction: {e}");
+        eprintln!("[codegraph] {msg}");
+        msg
+    })?;
+
+    let mut total_chunks = 0usize;
+    let mut chunk_failures: Vec<String> = Vec::new();
     for chunk in edges.chunks(INSERT_CHUNK) {
+        total_chunks += 1;
         if let Err(e) = insert_edge_chunk(&tx, chunk) {
-            eprintln!(
-                "[codegraph] insert_edges: skipped chunk of {} rows due to error: {e}",
-                chunk.len()
-            );
+            let msg = format!("chunk of {} row(s): {e}", chunk.len());
+            eprintln!("[codegraph] insert_edges: skipped {msg}");
+            chunk_failures.push(msg);
         }
     }
-    if let Err(e) = tx.commit() {
-        eprintln!("[codegraph] insert_edges: commit failed: {e}");
+
+    tx.commit().map_err(|e| {
+        let msg = format!("insert_edges: commit failed: {e}");
+        eprintln!("[codegraph] {msg}");
+        msg
+    })?;
+
+    if !chunk_failures.is_empty() {
+        return Err(format!(
+            "insert_edges: {} of {total_chunks} chunk(s) failed to insert: {}",
+            chunk_failures.len(),
+            chunk_failures.join("; ")
+        ));
     }
+    Ok(())
 }
 
 /// Bind and execute a single chunk in its own fallible scope so the caller
@@ -755,4 +772,70 @@ mod tests {
         let pairs = import_name_pairs(&imp);
         assert!(pairs.iter().all(|(_, _, type_only)| !*type_only));
     }
+
+    /// Minimal in-memory `edges` schema covering only the columns
+    /// `insert_edge_chunk` writes.
+    fn edges_test_conn() -> Connection {
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute_batch(
+            "CREATE TABLE edges (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                source_id INTEGER NOT NULL,
+                target_id INTEGER NOT NULL,
+                kind TEXT NOT NULL,
+                confidence REAL DEFAULT 1.0,
+                dynamic INTEGER DEFAULT 0
+            );",
+        )
+        .unwrap();
+        conn
+    }
+
+    fn sample_edge_row() -> EdgeRow {
+        EdgeRow {
+            source_id: 1,
+            target_id: 2,
+            kind: "imports".to_string(),
+            confidence: 1.0,
+            dynamic: 0,
+        }
+    }
+
+    #[test]
+    fn insert_edges_commits_rows_and_returns_ok_on_success() {
+        let conn = edges_test_conn();
+        let result = insert_edges(&conn, &[sample_edge_row()]);
+        assert!(result.is_ok(), "expected Ok, got {result:?}");
+
+        let count: i64 = conn
+            .query_row("SELECT COUNT(*) FROM edges", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(count, 1);
+    }
+
+    #[test]
+    fn insert_edges_returns_ok_for_an_empty_batch() {
+        let conn = edges_test_conn();
+        assert!(insert_edges(&conn, &[]).is_ok());
+    }
+
+    /// Regression test for #1827: a transaction-start failure must surface
+    /// as `Err`, not be swallowed behind a stderr-only warning while the
+    /// caller sees nothing (`run_pipeline` previously ignored this entirely).
+    #[test]
+    fn insert_edges_returns_err_when_transaction_cannot_start() {
+        let conn = edges_test_conn();
+        // SQLite refuses to start a second transaction on a connection that
+        // already has one active — a reliable way to force
+        // `conn.unchecked_transaction()` to fail deterministically.
+        conn.execute_batch("BEGIN").unwrap();
+
+        let result = insert_edges(&conn, &[sample_edge_row()]);
+        assert!(
+            result.is_err(),
+            "insert_edges must return Err instead of silently no-op'ing when the transaction can't start"
+        );
+
+        conn.execute_batch("ROLLBACK").unwrap();
+    }
 }