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");