From b484f685f88ab5dfc7cd1d47e349d4b6639182dc Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:38:00 -0800 Subject: [PATCH] git: fail --git indexing when a batch is not stored A `--git` or tree run that fails to store a batch of files logs the failure, prints "Indexing completed successfully!" and exits 0. In two of three runs of the v7.3..v7.4-rc1 merge window, an inserter hit lance's "Too many concurrent writers" retry limit on the `types` and `functions` tables, and the database came out 512 `processed_files` rows short with exit status 0. `insert_batch_and_mark()` already leaves a failed batch's files unmarked, but only a later run over a range containing those files reads them again, and nobody reruns a range that reported success. Such a run now exits non-zero and says how many files it left unindexed: - `insert_batch_and_mark()` returns the run's stored-batch count when every fact table and the mark were stored, and `None` otherwise. - The range and tree inserters count failed batches and their files, and `GitTupleStats` carries both counts. An inserter task that panics counts as one failed batch, with the files of the batch it was storing. The periodic optimization check moves from `insert_batch_and_mark()` into the inserters, after the batch is marked or failed, so a panic there does not count marked files. - `process_git_range()` and `process_git_tree()` finish as before (summary, optimization, `call_edges` index), then return an error such as "2 batches (512 files) were not stored; indexing again reads them". Their summaries say "Incomplete" and give the same count. `process_git_tree()` returns the error before `forget_older_processed_files()`, `record_index_build()` and `record_tree_sha()`, because those record that the whole tree was read. The change covers the file batches only. A failed `git_commits` insert in `process_commits_pipeline()` is still only logged. `semcode-index` calls both functions with `?`, so a failed run no longer prints the success line. Branch mode stops without recording the branch as indexed, and the branches after it wait for the next run. Running the same command again reads only the unmarked files. The other callers already handle an error: - `semcode --reindex-if-stale` now exits with the error and leaves the index marked stale. Before, `record_index_build()` ran anyway, the stale flag cleared, and semcode answered from an index with holes. - semcode's auto-indexing of the current commit no longer prints "Indexing complete"; its warning carrying the error shows with `SEMCODE_DEBUG=warn`. - semcode-mcp reports a failed index instead of a completed one. `insert_batch_and_mark_reports_a_failed_mark` checks the return value with `processed_files` dropped, and the existing per-table test now checks it with each fact table dropped. `a_run_with_an_unstored_batch_fails` checks that tree and range runs over a one-file repository fail, record no tree SHA while the `types` table is missing, and succeed once it exists. `unstored_batches_error_counts_batches_and_files` checks the message. --- src/git_range.rs | 330 +++++++++++++++++++++++++++++++++++++---------- 1 file changed, 260 insertions(+), 70 deletions(-) diff --git a/src/git_range.rs b/src/git_range.rs index 8612688..d67140e 100644 --- a/src/git_range.rs +++ b/src/git_range.rs @@ -113,6 +113,40 @@ struct GitTupleStats { files_processed: usize, functions_count: usize, types_count: usize, + /// Batches whose facts or mark were not all stored, counting the batch + /// a panicked inserter held, and the files they left unmarked. + failed_batches: usize, + failed_files: usize, +} + +impl GitTupleStats { + /// The run's error if any batch was not stored. The run otherwise looks + /// complete, and only a later run over the same files reads them again. + fn unstored_batches_error(&self) -> Result<()> { + if self.failed_batches > 0 { + anyhow::bail!( + "{} not stored; indexing again reads them", + self.unstored_summary() + ); + } + Ok(()) + } + + /// "N batches (M files)", for the summary and the error. + fn unstored_summary(&self) -> String { + let plural = + |n: usize, one: &str, many: &str| format!("{n} {}", if n == 1 { one } else { many }); + format!( + "{} ({}) {}", + plural(self.failed_batches, "batch", "batches"), + plural(self.failed_files, "file", "files"), + if self.failed_batches == 1 { + "was" + } else { + "were" + } + ) + } } /// A repo-relative path prefix bounding enumeration scope. @@ -559,10 +593,16 @@ fn tuple_worker_shared(ctx: TupleWorkerContext) { /// completes what the failed run wrote. Three shapes escape that -- a blob /// read at a new path, a failure that fails the same way every run, a run /// whose input never contains the blob -- and those stay holes until the -/// run learns to retry per file and to account for unmarked files. +/// run learns to retry per file. The run reports them as unstored files, +/// so they are not silent. /// /// A failed mark is the same shape: the facts are in, the file is not /// marked, the next run re-reads it and the merges change nothing. +/// +/// Returns the run's count of stored batches, this one included, when +/// every fact table and the mark were stored, and `None` otherwise, so the +/// run can report the files it left unmarked instead of claiming success. +/// The caller runs the periodic optimization with that count. async fn insert_batch_and_mark( db: &DatabaseManager, batch: GitTupleResults, @@ -570,8 +610,7 @@ async fn insert_batch_and_mark( functions_counter: &AtomicUsize, types_counter: &AtomicUsize, batches_inserted_counter: &AtomicUsize, - optimization_check_timer: &Arc>, -) { +) -> Option { // Exhaustive: a new fact table must be inserted and checked here, not // silently left out of the batch the mark commits. let GitTupleResults { @@ -713,18 +752,7 @@ async fn insert_batch_and_mark( // Note: not tracking processed_files_count in a counter since it's not displayed // Only increment batches_inserted if insertion was successful - if insertion_successful { - let total_batches = batches_inserted_counter.fetch_add(1, Ordering::Relaxed) + 1; - - // Periodic optimization check using shared function - crate::indexer::check_and_optimize_if_needed( - db, - inserter_id, - total_batches, - optimization_check_timer, - ) - .await; - } + insertion_successful.then(|| batches_inserted_counter.fetch_add(1, Ordering::Relaxed) + 1) } /// Process git file tuples using streaming pipeline with database inserters @@ -738,11 +766,7 @@ async fn process_git_tuples_streaming(config: StreamingConfig) -> Result Result Result Result { - insert_batch_and_mark( + let files = batch.processed_files.len(); + in_flight.store(files, Ordering::Relaxed); + let stored = insert_batch_and_mark( &db_manager_clone, batch, inserter_id, &functions_counter, &types_counter, &batches_inserted_counter, - &optimization_check_timer, ) .await; + // Marked or failed: a panic from here on loses no + // batch. + in_flight.store(0, Ordering::Relaxed); + match stored { + Some(total_batches) => { + crate::indexer::check_and_optimize_if_needed( + &db_manager_clone, + inserter_id, + total_batches, + &optimization_check_timer, + ) + .await; + } + None => { + failed_batches_counter.fetch_add(1, Ordering::Relaxed); + failed_files_counter.fetch_add(files, Ordering::Relaxed); + } + } } Err(_) => break, // Channel closed, exit task } } }); - inserter_handles.push(handle); + inserter_handles.push((handle, in_flight_files)); } // Wait for all generators to complete @@ -943,9 +994,11 @@ async fn process_git_tuples_streaming(config: StreamingConfig) -> Result Result Result Result Result { - insert_batch_and_mark( + let files = batch.processed_files.len(); + in_flight.store(files, Ordering::Relaxed); + let stored = insert_batch_and_mark( &db_manager_clone, batch, inserter_id, &functions_counter, &types_counter, &batches_inserted_counter, - &optimization_check_timer, ) .await; + // Marked or failed: a panic from here on loses no + // batch. + in_flight.store(0, Ordering::Relaxed); + match stored { + Some(total_batches) => { + crate::indexer::check_and_optimize_if_needed( + &db_manager_clone, + inserter_id, + total_batches, + &optimization_check_timer, + ) + .await; + } + None => { + failed_batches_counter.fetch_add(1, Ordering::Relaxed); + failed_files_counter.fetch_add(files, Ordering::Relaxed); + } + } } Err(_) => break, } } }); - inserter_handles.push(handle); + inserter_handles.push((handle, in_flight_files)); } // Wait for generator to complete @@ -1204,9 +1286,11 @@ async fn process_git_tree_streaming(config: TreeStreamingConfig) -> Result Result 0 { + println!("\n=== Git Tree Indexing Incomplete ==="); + println!("Not stored: {}", stats.unstored_summary()); + } else { + println!("\n=== Git Tree Indexing Complete ==="); + } println!("Total time: {:.1}s", total_time.as_secs_f64()); println!("Files processed: {}", stats.files_processed); println!("Functions indexed: {}", stats.functions_count); @@ -1337,6 +1428,10 @@ pub async fn process_git_tree( error!("Failed to ensure call_edges index: {}", e); } + // A batch that was not stored leaves its files unread, so the coverage + // records below would claim a tree this run did not finish. + stats.unstored_batches_error()?; + // The whole tree has been read with this build, so a file still recorded // against an older extractor is content the tree no longer holds. Left in // place it keeps the index answering "older than this build" for ever, so @@ -1526,7 +1621,12 @@ pub async fn process_git_range( let total_time = start_time.elapsed(); - info!("\n=== Git Range Pipeline Complete ==="); + if stats.failed_batches > 0 { + info!("\n=== Git Range Pipeline Incomplete ==="); + info!("Not stored: {}", stats.unstored_summary()); + } else { + info!("\n=== Git Range Pipeline Complete ==="); + } info!("Total time: {:.1}s", total_time.as_secs_f64()); info!("Commits indexed: {commit_count}"); info!("Files processed: {}", stats.files_processed); @@ -1560,7 +1660,7 @@ pub async fn process_git_range( // Do not write a database version. Part of the database was updated here, // but other parts could have been written by an older version of semcode. - Ok(()) + stats.unstored_batches_error() } #[cfg(test)] @@ -1666,17 +1766,11 @@ mod tests { } } - fn test_counters() -> ( - AtomicUsize, - AtomicUsize, - AtomicUsize, - Arc>, - ) { + fn test_counters() -> (AtomicUsize, AtomicUsize, AtomicUsize) { ( AtomicUsize::new(0), AtomicUsize::new(0), AtomicUsize::new(0), - Arc::new(std::sync::Mutex::new(std::time::Instant::now())), ) } @@ -1711,17 +1805,13 @@ mod tests { let (_dir, db) = test_db().await; db.connection().drop_table(table, &[]).await.unwrap(); - let (functions, types_count, batches, timer) = test_counters(); - insert_batch_and_mark( - &db, - test_batch(), - 0, - &functions, - &types_count, - &batches, - &timer, - ) - .await; + let (functions, types_count, batches) = test_counters(); + assert!( + insert_batch_and_mark(&db, test_batch(), 0, &functions, &types_count, &batches) + .await + .is_none(), + "{table} failing must report the batch as not stored" + ); assert!( !db.is_file_processed("abc123").await.unwrap(), "{table} failing must leave the file unmarked" @@ -1729,16 +1819,12 @@ mod tests { assert_eq!(batches.load(Ordering::Relaxed), 0); db.create_tables().await.unwrap(); - insert_batch_and_mark( - &db, - test_batch(), - 0, - &functions, - &types_count, - &batches, - &timer, - ) - .await; + assert!( + insert_batch_and_mark(&db, test_batch(), 0, &functions, &types_count, &batches) + .await + == Some(1), + "{table} healed must report the batch as stored" + ); assert!( db.is_file_processed("abc123").await.unwrap(), "{table} healed must mark the file" @@ -1752,18 +1838,9 @@ mod tests { #[tokio::test] async fn a_retried_batch_merges_onto_its_rows() { let (_dir, db) = test_db().await; - let (functions, types_count, batches, timer) = test_counters(); + let (functions, types_count, batches) = test_counters(); for _ in 0..2 { - insert_batch_and_mark( - &db, - test_batch(), - 0, - &functions, - &types_count, - &batches, - &timer, - ) - .await; + insert_batch_and_mark(&db, test_batch(), 0, &functions, &types_count, &batches).await; } let found = db.get_all_functions_metadata_only().await.unwrap(); assert_eq!(found.len(), 1); @@ -1771,6 +1848,119 @@ mod tests { assert_eq!(landed_types.len(), 1); } + #[test] + fn unstored_batches_error_counts_batches_and_files() { + assert!(GitTupleStats::default().unstored_batches_error().is_ok()); + let stats = |failed_batches, failed_files| GitTupleStats { + failed_batches, + failed_files, + ..Default::default() + }; + assert_eq!( + stats(1, 1) + .unstored_batches_error() + .unwrap_err() + .to_string(), + "1 batch (1 file) was not stored; indexing again reads them" + ); + assert_eq!( + stats(2, 512) + .unstored_batches_error() + .unwrap_err() + .to_string(), + "2 batches (512 files) were not stored; indexing again reads them" + ); + // A panicked inserter that held no batch still fails the run. + assert_eq!( + stats(1, 0) + .unstored_batches_error() + .unwrap_err() + .to_string(), + "1 batch (0 files) was not stored; indexing again reads them" + ); + } + + /// A mark that fails after every fact landed still leaves the file + /// unread, so the batch is reported as not stored. + #[tokio::test] + async fn insert_batch_and_mark_reports_a_failed_mark() { + let (_dir, db) = test_db().await; + db.connection() + .drop_table("processed_files", &[]) + .await + .unwrap(); + let (functions, types_count, batches) = test_counters(); + let stored = + insert_batch_and_mark(&db, test_batch(), 0, &functions, &types_count, &batches).await; + assert_eq!(stored, None); + assert_eq!(batches.load(Ordering::Relaxed), 0); + } + + fn git(repo: &std::path::Path, args: &[&str]) { + let status = std::process::Command::new("git") + .args(args) + .current_dir(repo) + // Do not inherit developer's git configuration. + .env("GIT_CONFIG_GLOBAL", "/dev/null") + .env("GIT_CONFIG_SYSTEM", "/dev/null") + .env("GIT_AUTHOR_NAME", "test") + .env("GIT_AUTHOR_EMAIL", "test@test.com") + .env("GIT_COMMITTER_NAME", "test") + .env("GIT_COMMITTER_EMAIL", "test@test.com") + .status() + .unwrap(); + assert!(status.success(), "git {args:?} failed"); + } + + /// A tree or range run that could not store a batch fails, names the + /// unstored files, and records no tree coverage. Once the database + /// accepts the batch, the same runs succeed. + #[tokio::test] + async fn a_run_with_an_unstored_batch_fails() { + let repo_dir = tempfile::tempdir().unwrap(); + let repo = repo_dir.path().to_path_buf(); + git(&repo, &["init", "-q"]); + std::fs::write( + repo.join("a.c"), + "struct a_struct { int x; };\nint a_function(void)\n{\n\treturn 0;\n}\n", + ) + .unwrap(); + git(&repo, &["add", "a.c"]); + git(&repo, &["commit", "-q", "-m", "a"]); + let head = crate::git::get_git_sha(&repo).unwrap().unwrap(); + let extensions = crate::file_extensions::default_extensions(); + + let (_db_dir, db) = test_db().await; + let db = Arc::new(db); + db.connection().drop_table("types", &[]).await.unwrap(); + + let err = process_git_tree(&repo, &head, &extensions, db.clone(), false, 1, &[]) + .await + .unwrap_err(); + assert!( + err.to_string().contains("1 batch (1 file) was not stored"), + "{err}" + ); + assert_eq!(db.recorded_tree_sha().await.unwrap(), None); + + let err = process_git_range(&repo, "HEAD", &extensions, db.clone(), false, 1, &[]) + .await + .unwrap_err(); + assert!( + err.to_string().contains("1 batch (1 file) was not stored"), + "{err}" + ); + + db.create_tables().await.unwrap(); + process_git_range(&repo, "HEAD", &extensions, db.clone(), false, 1, &[]) + .await + .unwrap(); + process_git_tree(&repo, &head, &extensions, db.clone(), false, 1, &[]) + .await + .unwrap(); + assert_eq!(db.recorded_tree_sha().await.unwrap(), Some(head)); + } + #[test] fn scope_prefixes_empty_for_repo_root() { let workdir = std::path::Path::new("/repo");