diff --git a/src/bin/index.rs b/src/bin/index.rs index 10340c9..1c4dd6d 100644 --- a/src/bin/index.rs +++ b/src/bin/index.rs @@ -162,27 +162,21 @@ fn fetch_lore_manifest() -> Result { Ok(manifest) } -/// Resolve a lore list name to a full git URL -fn resolve_lore_url(lore_arg: &str, manifest: &serde_json::Value) -> Result { - // Parse the list name and optional archive number - let (list_name, archive_num) = if let Some(slash_pos) = lore_arg.find('/') { - let (name, num) = lore_arg.split_at(slash_pos); - let num = num.trim_start_matches('/'); - (name, Some(num)) - } else { - (lore_arg, None) - }; - +/// Return all `(archive_num, url)` entries for `list_name` in the lore.kernel.org +/// manifest, sorted ascending by `archive_num`. +fn list_manifest_lore_archives( + list_name: &str, + manifest: &serde_json::Value, +) -> Result> { // The manifest structure has keys like "/lkml/git/0.git", "/lkml/git/1.git", etc. let lists = manifest .as_object() .ok_or_else(|| anyhow::anyhow!("Manifest is not a JSON object"))?; - // Find all entries matching the list name pattern let pattern = format!("/{}/git/", list_name); let mut matching_archives: Vec<(u32, String)> = Vec::new(); - for (key, _value) in lists.iter() { + for key in lists.keys() { if key.starts_with(&pattern) && key.ends_with(".git") { // Extract the archive number from the path // Example: "/lkml/git/17.git" -> extract "17" @@ -198,6 +192,23 @@ fn resolve_lore_url(lore_arg: &str, manifest: &serde_json::Value) -> Result Result { + // Parse the list name and optional archive number + let (list_name, archive_num) = if let Some(slash_pos) = lore_arg.find('/') { + let (name, num) = lore_arg.split_at(slash_pos); + let num = num.trim_start_matches('/'); + (name, Some(num)) + } else { + (lore_arg, None) + }; + + let matching_archives = list_manifest_lore_archives(list_name, manifest)?; + if matching_archives.is_empty() { return Err(anyhow::anyhow!( "List '{}' not found in manifest. Check available lists with SEMCODE_DEBUG=debug", @@ -205,9 +216,6 @@ fn resolve_lore_url(lore_arg: &str, manifest: &serde_json::Value) -> Result Option<(Str } /// Check for new lore archives available on lore.kernel.org that are not locally cached. -/// Returns a list of (list_name, new_archive_numbers) for lists with newer archives. +/// Returns a list of `(list_name, Vec<(archive_number, url)>)` for lists with newer archives. fn find_new_lore_archives( local_archives: &[PathBuf], lore_base: &Path, manifest: &serde_json::Value, -) -> Vec<(String, Vec)> { +) -> Vec<(String, Vec<(u32, String)>)> { use std::collections::HashMap; // Build a map of list_name -> max local archive number @@ -520,37 +528,18 @@ fn find_new_lore_archives( return Vec::new(); } - let lists = match manifest.as_object() { - Some(obj) => obj, - None => return Vec::new(), - }; - - let mut new_archives: Vec<(String, Vec)> = Vec::new(); + let mut new_archives: Vec<(String, Vec<(u32, String)>)> = Vec::new(); for (list_name, max_local) in &local_max { - // Find all archives for this list in the manifest - let pattern = format!("/{}/git/", list_name); - let mut remote_archives: Vec = Vec::new(); - - for key in lists.keys() { - if key.starts_with(&pattern) && key.ends_with(".git") { - if let Some(num_str) = key - .strip_prefix(&pattern) - .and_then(|s| s.strip_suffix(".git")) - { - if let Ok(num) = num_str.parse::() { - if num > *max_local { - remote_archives.push(num); - } - } - } + if let Ok(remote_archives) = list_manifest_lore_archives(list_name, manifest) { + let newer: Vec<(u32, String)> = remote_archives + .into_iter() + .filter(|(num, _)| *num > *max_local) + .collect(); + if !newer.is_empty() { + new_archives.push((list_name.clone(), newer)); } } - - if !remote_archives.is_empty() { - remote_archives.sort(); - new_archives.push((list_name.clone(), remote_archives)); - } } // Sort by list name for consistent output @@ -612,6 +601,71 @@ async fn fetch_lore_archive(repo_path: PathBuf) -> Result { struct LoreIndexResult { new_emails: usize, total_emails: usize, + /// True if all commits in the archive are recorded in `lore_indexed_commits` + /// (`recorded_commits == new_commits.len()`). + fully_indexed: bool, +} + +/// Status of a refreshed or newly cloned lore shard (`/`). +#[derive(Debug, Clone)] +struct LoreShardStatus { + list_name: String, + shard_num: u32, + fully_indexed: bool, +} + +/// Aggregated statistics and per-shard statuses across a lore refresh run. +#[derive(Default)] +struct LoreRefreshStats { + total_archives: usize, + archives_processed: usize, + total_new_emails: usize, + total_emails_all_archives: usize, + failed_archives: Vec<(String, String)>, + lists_with_errors: HashSet, + shard_statuses: Vec, +} + +impl LoreRefreshStats { + /// Record the outcome of refreshing or indexing a single `/` shard. + fn record_archive_outcome(&mut self, display_name: String, outcome: Result) { + self.total_archives += 1; + match outcome { + Ok(index_result) => { + self.total_new_emails += index_result.new_emails; + self.total_emails_all_archives += index_result.total_emails; + if index_result.fully_indexed { + self.archives_processed += 1; + } else { + self.failed_archives.push(( + display_name.clone(), + "incomplete commit indexing (some batches failed)".to_string(), + )); + } + + if let Some((list_name, num_str)) = display_name.split_once('/') { + let list_name = list_name.to_string(); + let shard_num = num_str.parse::().ok(); + if !index_result.fully_indexed { + self.lists_with_errors.insert(list_name.clone()); + } + if let Some(shard_num) = shard_num { + self.shard_statuses.push(LoreShardStatus { + list_name, + shard_num, + fully_indexed: index_result.fully_indexed, + }); + } + } + } + Err(e) => { + if let Some((list_name, _)) = display_name.split_once('/') { + self.lists_with_errors.insert(list_name.to_string()); + } + self.failed_archives.push((display_name, e.to_string())); + } + } + } } /// Index a lore archive repository, filtering out already-indexed commits. @@ -685,13 +739,14 @@ async fn index_lore_archive( return Ok(LoreIndexResult { new_emails: 0, total_emails: total_commits, + fully_indexed: true, }); } let new_count = new_commits.len(); // Process new commits - process_lore_commits_pipeline( + let (inserted, recorded_commits) = process_lore_commits_pipeline( archive_path, new_commits, db_manager.clone(), @@ -703,15 +758,198 @@ async fn index_lore_archive( println!( "Indexed {} new emails from {} (total in archive: {})", - new_count, display_name, total_commits + inserted, display_name, total_commits ); Ok(LoreIndexResult { - new_emails: new_count, + new_emails: inserted, total_emails: total_commits, + fully_indexed: recorded_commits == new_count, }) } +/// Clone a lore archive (if not already cloned) and index its commits. +async fn clone_and_index_lore_archive( + lore_url: &str, + display_name: &str, + database_path: &str, + db_manager: &Arc, + batch_size: usize, + num_workers: usize, + db_threads: usize, +) -> Result { + let clone_path = clone_lore_repository(lore_url, database_path).await?; + info!("Lore archive cloned to: {}", clone_path.display()); + + let lore_repo = gix::discover(&clone_path).map_err(|e| { + anyhow::anyhow!("Error opening repository {}: {}", clone_path.display(), e) + })?; + + index_lore_archive( + lore_repo, + &clone_path, + display_name, + db_manager, + batch_size, + num_workers, + db_threads, + ) + .await +} + +/// Remove local git clone directories for superseded shards (`< max_num`) of a mailing list +/// once: +/// 1. New shards (`> old_num`) from `manifest.js.gz` were checked and cloned *before* +/// fetching the superseded shards (guaranteeing the superseded shards were already +/// permanently sealed on `lore.kernel.org` prior to `git fetch`), AND +/// 2. All shards for the list in this run were indexed completely (`fully_indexed == true`) +/// with no fetch/index errors, AND +/// 3. A newer shard (`max_num > shard_num`) is present and fully indexed. +/// +/// Note: This only deletes the bare git repository under `/lore//`; +/// all indexed emails remain permanently stored in the LanceDB `lore` table. +fn retire_superseded_lore_archives( + lore_base: &Path, + shard_statuses: &[LoreShardStatus], + lists_with_errors: &HashSet, +) { + let mut by_list: std::collections::HashMap<&str, Vec<&LoreShardStatus>> = + std::collections::HashMap::new(); + for status in shard_statuses { + by_list + .entry(status.list_name.as_str()) + .or_default() + .push(status); + } + + for (list_name, shards) in by_list { + if lists_with_errors.contains(list_name) || shards.iter().any(|s| !s.fully_indexed) { + continue; + } + + let Some(max_num) = shards.iter().map(|s| s.shard_num).max() else { + continue; + }; + + for shard in shards { + if shard.shard_num < max_num { + let shard_path = lore_base + .join(list_name) + .join(shard.shard_num.to_string()); + if !shard_path.exists() { + continue; + } + + println!( + "Retiring superseded lore archive: {}/{} (verified complete)", + list_name, shard.shard_num + ); + if let Err(e) = std::fs::remove_dir_all(&shard_path) { + warn!( + "Failed to remove superseded archive {}: {}", + shard_path.display(), + e + ); + } + } + } + } +} + +/// Compact lore tables and update/optimize FTS indices after new emails are indexed. +async fn compact_and_optimize_lore_tables(db_manager: &DatabaseManager) { + println!("\nCompacting lore tables..."); + match db_manager.compact_lore_tables().await { + Ok(_) => println!("Lore table compaction completed successfully"), + Err(e) => error!("Failed to compact lore tables: {}", e), + } + + println!("\nUpdating FTS indices for lore table..."); + match db_manager.ensure_lore_fts_indices().await { + Ok(_) => {} + Err(e) => eprintln!("Warning: Failed to ensure FTS indices: {}", e), + } + match db_manager.optimize_lore_fts_indices().await { + Ok(_) => println!("FTS indices updated successfully"), + Err(e) => eprintln!("Warning: Failed to optimize FTS indices: {}", e), + } +} + +/// Check `lore.kernel.org` manifest for newly created shards of tracked mailing lists, +/// clone and index them, and return their `LoreRefreshStats`. +async fn sync_new_lore_archives( + archive_paths: &[PathBuf], + lore_base: &Path, + database_path: &str, + db_manager: &Arc, + batch_size: usize, + num_workers: usize, + db_threads: usize, +) -> LoreRefreshStats { + let mut stats = LoreRefreshStats::default(); + + println!("\nChecking for new lore archives..."); + let manifest = match fetch_lore_manifest() { + Ok(m) => m, + Err(e) => { + warn!("Could not check for new archives: {}", e); + println!( + "Note: Could not fetch lore manifest to check for new archives: {}", + e + ); + return stats; + } + }; + + let new_archives = find_new_lore_archives(archive_paths, lore_base, &manifest); + if new_archives.is_empty() { + println!("All tracked mailing lists are up to date."); + return stats; + } + + println!( + "\n{}", + "New archives available on lore.kernel.org (auto-indexing):".yellow() + ); + for (list_name, archive_entries) in &new_archives { + let nums_str: Vec = archive_entries + .iter() + .map(|(n, _)| n.to_string()) + .collect(); + println!(" {}: archive(s) {}", list_name.cyan(), nums_str.join(", ")); + } + + for (list_name, archive_entries) in &new_archives { + for (num, lore_url) in archive_entries { + let display_name = format!("{}/{}", list_name, num); + + println!( + "\n=== Cloning and indexing new lore archive: {} ===", + display_name + ); + info!("Cloning and indexing new lore archive: {}", display_name); + + let outcome = clone_and_index_lore_archive( + lore_url, + &display_name, + database_path, + db_manager, + batch_size, + num_workers, + db_threads, + ) + .await; + + if let Err(ref e) = outcome { + eprintln!("Error processing {}: {:#}", display_name, e); + } + stats.record_archive_outcome(display_name, outcome); + } + } + + stats +} + // ==================== Branch Indexing Support ==================== /// Collect branches to index from the various branch-related CLI flags @@ -1046,30 +1284,10 @@ async fn main() -> Result<()> { println!("\n=== Processing lore archive: {} ===", lore_url); info!("Processing lore archive: {}", lore_url); - // Clone the repository - let clone_path = match clone_lore_repository(lore_url, &database_path).await { - Ok(path) => path, - Err(e) => { - eprintln!("Error cloning {}: {:#}", lore_url, e); - continue; - } - }; - info!("Lore archive cloned to: {}", clone_path.display()); - - // Open the repository - let lore_repo = match gix::discover(&clone_path) { - Ok(repo) => repo, - Err(e) => { - eprintln!("Error opening repository {}: {}", clone_path.display(), e); - continue; - } - }; - - // Index the archive using the shared function - match index_lore_archive( - lore_repo, - &clone_path, + match clone_and_index_lore_archive( lore_url, + lore_url, + &database_path, &db_manager, batch_size, num_workers, @@ -1082,7 +1300,7 @@ async fn main() -> Result<()> { total_emails_all_archives += result.total_emails; } Err(e) => { - eprintln!("Error indexing {}: {}", lore_url, e); + eprintln!("Error processing {}: {:#}", lore_url, e); continue; } } @@ -1100,30 +1318,7 @@ async fn main() -> Result<()> { ); if total_new_emails > 0 { - // Compact only lore tables. The full optimize_database() - // method processes every table in the database including - // code-index tables and content shards that a lore run - // never touches. On memory-constrained systems the - // combined working set triggers the OOM killer. - println!("\nCompacting lore tables..."); - match db_manager.compact_lore_tables().await { - Ok(_) => println!("Lore table compaction completed successfully"), - Err(e) => error!("Failed to compact lore tables: {}", e), - } - - // Create FTS indices on first run; merge new rows on - // subsequent runs. LanceDB's native FTS engine serves - // unindexed rows via brute-force fallback, so queries - // remain correct before optimize completes. - println!("\nUpdating FTS indices for lore table..."); - match db_manager.ensure_lore_fts_indices().await { - Ok(_) => {} - Err(e) => eprintln!("Warning: Failed to ensure FTS indices: {}", e), - } - match db_manager.optimize_lore_fts_indices().await { - Ok(_) => println!("FTS indices updated successfully"), - Err(e) => eprintln!("Warning: Failed to optimize FTS indices: {}", e), - } + compact_and_optimize_lore_tables(&db_manager).await; } println!("\nTo query this database, run:"); @@ -1189,61 +1384,55 @@ async fn main() -> Result<()> { let batch_size = 1024; let num_workers = num_cpus::get(); let db_threads = args.db_threads; - let total_archives = archives_with_names.len(); // Process archives sequentially. LanceDB merge_insert // uses a shared DataFusion memory pool, and concurrent // pipelines writing large lore emails exhaust it, // causing both to stall indefinitely. - let mut results: Vec> = - Vec::with_capacity(total_archives); + // Check for new archives on lore.kernel.org FIRST before fetching + // existing shards. Because public-inbox v2 permanently seals shard M + // once shard N > M is created, discovering N before calling git fetch + // on M guarantees that M's fetch retrieves its final commits. + let mut stats = sync_new_lore_archives( + &archive_paths, + &lore_base, + &database_path, + &db_manager, + batch_size, + num_workers, + db_threads, + ) + .await; for (archive_path, display_name) in archives_with_names { println!("\n=== Refreshing lore archive: {} ===", display_name); println!("[{}] Fetching updates from remote...", display_name); - let lore_repo = match fetch_lore_archive(archive_path.clone()).await { - Ok(repo) => repo, - Err(e) => { - results.push(Err((display_name, e))); - continue; + let outcome = match fetch_lore_archive(archive_path.clone()).await { + Ok(lore_repo) => { + index_lore_archive( + lore_repo, + &archive_path, + &display_name, + &db_manager, + batch_size, + num_workers, + db_threads, + ) + .await } + Err(e) => Err(e), }; - - match index_lore_archive( - lore_repo, - &archive_path, - &display_name, - &db_manager, - batch_size, - num_workers, - db_threads, - ) - .await - { - Ok(result) => results.push(Ok((display_name, result))), - Err(e) => results.push(Err((display_name, e))), - } + stats.record_archive_outcome(display_name, outcome); } - // Aggregate results - let mut total_new_emails = 0usize; - let mut total_emails_all_archives = 0usize; - let mut archives_processed = 0usize; - let mut failed_archives: Vec<(String, String)> = Vec::new(); - - for result in results { - match result { - Ok((_, index_result)) => { - total_new_emails += index_result.new_emails; - total_emails_all_archives += index_result.total_emails; - archives_processed += 1; - } - Err((name, e)) => { - failed_archives.push((name, e.to_string())); - } - } - } + // Retire any superseded older shard git directories whose newer shard + // and final tail commits are now 100% indexed. + retire_superseded_lore_archives( + &lore_base, + &stats.shard_statuses, + &stats.lists_with_errors, + ); let total_time = start_time.elapsed(); @@ -1251,73 +1440,24 @@ async fn main() -> Result<()> { println!("Total time: {:.1}s", total_time.as_secs_f64()); println!( "Archives refreshed: {}/{}", - archives_processed, total_archives + stats.archives_processed, stats.total_archives ); - println!("New emails indexed: {}", total_new_emails); + println!("New emails indexed: {}", stats.total_new_emails); println!( - "Total emails across archives: {}", - total_emails_all_archives + "Total commits across active archives: {}", + stats.total_emails_all_archives ); // Report failed archives if any - if !failed_archives.is_empty() { + if !stats.failed_archives.is_empty() { eprintln!("\nFailed archives:"); - for (name, err) in &failed_archives { + for (name, err) in &stats.failed_archives { eprintln!(" {}: {}", name, err); } } - if total_new_emails > 0 { - println!("\nCompacting lore tables..."); - match db_manager.compact_lore_tables().await { - Ok(_) => println!("Lore table compaction completed successfully"), - Err(e) => error!("Failed to compact lore tables: {}", e), - } - - println!("\nUpdating FTS indices for lore table..."); - match db_manager.ensure_lore_fts_indices().await { - Ok(_) => {} - Err(e) => eprintln!("Warning: Failed to ensure FTS indices: {}", e), - } - match db_manager.optimize_lore_fts_indices().await { - Ok(_) => println!("FTS indices updated successfully"), - Err(e) => eprintln!("Warning: Failed to optimize FTS indices: {}", e), - } - } - - // Check for new archives available on lore.kernel.org - println!("\nChecking for new lore archives..."); - match fetch_lore_manifest() { - Ok(manifest) => { - let new_archives = - find_new_lore_archives(&archive_paths, &lore_base, &manifest); - if new_archives.is_empty() { - println!("All tracked mailing lists are up to date."); - } else { - println!( - "\n{}", - "New archives available on lore.kernel.org:".yellow() - ); - for (list_name, archive_nums) in &new_archives { - let nums_str: Vec = - archive_nums.iter().map(|n| n.to_string()).collect(); - println!(" {}: archive(s) {}", list_name.cyan(), nums_str.join(", ")); - } - println!("\nTo add these archives, run:"); - for (list_name, archive_nums) in &new_archives { - for num in archive_nums { - println!(" semcode-index --lore {}/{}", list_name, num); - } - } - } - } - Err(e) => { - warn!("Could not check for new archives: {}", e); - println!( - "Note: Could not fetch lore manifest to check for new archives: {}", - e - ); - } + if stats.total_new_emails > 0 { + compact_and_optimize_lore_tables(&db_manager).await; } println!("\nTo query this database, run:"); @@ -1746,3 +1886,74 @@ async fn run_pipeline(args: Args) -> Result<()> { } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_retire_superseded_lore_archives() { + let temp_dir = tempfile::tempdir().unwrap(); + let lore_base = temp_dir.path().join("lore"); + + let lkml_18 = lore_base.join("lkml").join("18"); + let lkml_19 = lore_base.join("lkml").join("19"); + let lkml_20 = lore_base.join("lkml").join("20"); + let input_0 = lore_base.join("linux-input").join("0"); + + std::fs::create_dir_all(&lkml_18).unwrap(); + std::fs::create_dir_all(&lkml_19).unwrap(); + std::fs::create_dir_all(&lkml_20).unwrap(); + std::fs::create_dir_all(&input_0).unwrap(); + + // Pass 1: if any shard for lkml is not fully_indexed, no shards for lkml + // should be retired. + let statuses_incomplete = vec![ + LoreShardStatus { + list_name: "lkml".to_string(), + shard_num: 18, + fully_indexed: true, + }, + LoreShardStatus { + list_name: "lkml".to_string(), + shard_num: 19, + fully_indexed: false, + }, + LoreShardStatus { + list_name: "lkml".to_string(), + shard_num: 20, + fully_indexed: true, + }, + ]; + retire_superseded_lore_archives(&lore_base, &statuses_incomplete, &HashSet::new()); + assert!(lkml_18.exists(), "shard 18 must not be retired when list has incomplete shard"); + assert!(lkml_19.exists(), "incomplete shard 19 must not be retired"); + assert!(lkml_20.exists(), "current shard 20 must be kept"); + + // Pass 2: once all shards (18, 19, 20) are fully indexed, superseded shards + // 18 and 19 are retired and shard 20 is kept. + let statuses_complete = vec![ + LoreShardStatus { + list_name: "lkml".to_string(), + shard_num: 18, + fully_indexed: true, + }, + LoreShardStatus { + list_name: "lkml".to_string(), + shard_num: 19, + fully_indexed: true, + }, + LoreShardStatus { + list_name: "lkml".to_string(), + shard_num: 20, + fully_indexed: true, + }, + ]; + retire_superseded_lore_archives(&lore_base, &statuses_complete, &HashSet::new()); + + assert!(!lkml_18.exists(), "superseded shard 18 should be retired"); + assert!(!lkml_19.exists(), "superseded shard 19 should be retired"); + assert!(lkml_20.exists(), "current shard 20 must be kept"); + assert!(input_0.exists(), "unrelated list must be untouched"); + } +} diff --git a/src/database/connection.rs b/src/database/connection.rs index ee7da02..774a435 100644 --- a/src/database/connection.rs +++ b/src/database/connection.rs @@ -188,6 +188,7 @@ pub struct DatabaseManager { /// refreshed for. `u64::MAX` means "never built", so the first build on /// generation 0 is not mistaken for already-current. workdir_index_generation: AtomicU64, + lore_message_id_cache: tokio::sync::Mutex>>, } impl DatabaseManager { @@ -235,6 +236,7 @@ impl DatabaseManager { git_only: AtomicBool::new(false), workdir_command_generation: AtomicU64::new(0), workdir_index_generation: AtomicU64::new(u64::MAX), + lore_message_id_cache: tokio::sync::Mutex::new(None), }) } @@ -417,6 +419,8 @@ impl DatabaseManager { } pub async fn clear_all_data(&self) -> Result<()> { + *self.lore_message_id_cache.lock().await = None; + // Delete all data from each main table for table_name in &[ "functions", @@ -823,11 +827,13 @@ impl DatabaseManager { } pub async fn compact_lore_tables(&self) -> Result<()> { + *self.lore_message_id_cache.lock().await = None; self.schema_manager.compact_lore_tables().await } /// Drop and recreate all tables for maximum space savings pub async fn drop_and_recreate_tables(&self) -> Result<()> { + *self.lore_message_id_cache.lock().await = None; self.schema_manager.drop_and_recreate_tables().await } @@ -875,6 +881,9 @@ impl DatabaseManager { /// Drop and recreate a specific table pub async fn drop_and_recreate_table(&self, table_name: &str) -> Result<()> { + if table_name == "lore" { + *self.lore_message_id_cache.lock().await = None; + } self.schema_manager .drop_and_recreate_table(table_name) .await @@ -5921,7 +5930,9 @@ impl DatabaseManager { // lore_indexed_commits. Filtering avoids duplicates without // relying on merge_insert, which hits a spurious null-column // error in lance-file 3.0.1. - let new_indices = Self::filter_existing_lore_ids(&table, emails, &dedup_indices).await?; + let new_indices = self + .filter_existing_lore_ids(&table, emails, &dedup_indices) + .await?; if new_indices.len() < dedup_indices.len() { tracing::info!( @@ -5983,6 +5994,15 @@ impl DatabaseManager { failed_indices.len(), new_indices.len() ); + + // Remove any failed message_ids that were pre-claimed in + // filter_existing_lore_ids so they can be retried later. + let mut cache_guard = self.lore_message_id_cache.lock().await; + if let Some(ref mut known_ids) = *cache_guard { + for &idx in &failed_indices { + known_ids.remove(&emails[idx].message_id); + } + } } tracing::info!("insert_lore_emails: Batch insertion complete"); @@ -5990,54 +6010,51 @@ impl DatabaseManager { Ok(failed_indices) } - /// Return the subset of `indices` whose message_id does not - /// already exist in the lore table. + /// Return the subset of `indices` whose `message_id` does not + /// already exist in the lore table, using an in-memory `HashSet` + /// populated once via a streaming column read instead of running a + /// full-table `message_id IN (...)` DataFusion scan on every batch. + /// + /// Atomically inserts candidate `message_id`s into the cached set + /// while holding `lore_message_id_cache` so concurrent inserter tasks + /// cannot both pass the check for the same `message_id`. async fn filter_existing_lore_ids( + &self, table: &lancedb::Table, emails: &[crate::types::LoreEmailInfo], indices: &[usize], ) -> Result> { - use futures::TryStreamExt as _; - - // Build an IN-list predicate. message_ids are already - // validated to be non-empty, but escape single quotes for - // the SQL literal. - let id_list: Vec = indices - .iter() - .map(|&i| format!("'{}'", emails[i].message_id.replace('\'', "''"))) - .collect(); - - let predicate = format!("message_id IN ({})", id_list.join(", ")); - - let stream = table - .query() - .select(lancedb::query::Select::Columns(vec![ - "message_id".to_string() - ])) - .only_if(&predicate) - .execute() - .await?; - let batches: Vec<_> = stream.try_collect().await?; + let mut cache_guard = self.lore_message_id_cache.lock().await; + if cache_guard.is_none() { + let mut stream = table + .query() + .select(lancedb::query::Select::Columns(vec![ + "message_id".to_string() + ])) + .execute() + .await?; - let mut existing = std::collections::HashSet::new(); - for batch in &batches { - if let Some(col) = batch.column_by_name("message_id") { - let arr = col - .as_any() - .downcast_ref::() - .expect("message_id column must be StringArray"); - for i in 0..arr.len() { - if !arr.is_null(i) { - existing.insert(arr.value(i).to_string()); + let mut existing = HashSet::new(); + while let Some(batch) = stream.try_next().await? { + if let Some(col) = batch.column_by_name("message_id") { + if let Some(arr) = col.as_any().downcast_ref::() { + existing.reserve(arr.len()); + for i in 0..arr.len() { + if !arr.is_null(i) { + existing.insert(arr.value(i).to_string()); + } + } } } } + *cache_guard = Some(existing); } + let existing = cache_guard.as_mut().unwrap(); Ok(indices .iter() .copied() - .filter(|&i| !existing.contains(&emails[i].message_id)) + .filter(|&i| existing.insert(emails[i].message_id.clone())) .collect()) } @@ -7780,7 +7797,7 @@ impl DatabaseManager { } }; - let stream = table + let mut stream = table .query() .select(lancedb::query::Select::Columns(vec![ "git_commit_sha".to_string() @@ -7788,9 +7805,8 @@ impl DatabaseManager { .execute() .await?; - let batches: Vec<_> = stream.try_collect().await?; let mut existing = HashSet::new(); - for batch in batches { + while let Some(batch) = stream.try_next().await? { if let Some(column) = batch.column_by_name("git_commit_sha") { if let Some(string_array) = column.as_any().downcast_ref::() { existing.reserve(string_array.len()); diff --git a/src/indexer.rs b/src/indexer.rs index a7a9f48..b4091d5 100644 --- a/src/indexer.rs +++ b/src/indexer.rs @@ -405,22 +405,20 @@ pub fn parse_email_from_commit( repo: &gix::Repository, commit_sha: &str, ) -> Result { - // Resolve commit - let commit = resolve_to_commit(repo, commit_sha)?; - - // Get the tree - let tree = commit.tree()?; - - // Find the 'm' file in the tree - let m_entry = tree - .find_entry("m") - .ok_or_else(|| anyhow::anyhow!("Commit {} does not contain file 'm'", commit_sha))?; - - // Read the file content - let object = m_entry.object()?; - let blob = object.try_into_blob()?; - let email_content = String::from_utf8_lossy(blob.data.as_slice()).to_string(); + match parse_lore_commit_entry(repo, commit_sha)? { + LoreCommitParseOutcome::Email(info) => Ok(*info), + LoreCommitParseOutcome::NoEmailFile(_) => Err(anyhow::anyhow!( + "Commit {} does not contain file 'm'", + commit_sha + )), + } +} +/// Parse email headers and body from raw `'m'` blob contents +fn parse_email_from_blob_content( + commit_sha: &str, + email_content: &str, +) -> Result { // Parse email headers let mut headers = EmailHeaders::new(); @@ -809,8 +807,44 @@ pub async fn process_commits_pipeline( Ok(()) } -/// Process lore commits using a streaming pipeline -/// Similar to process_commits_pipeline but parses email from 'm' file +/// Result of parsing a single commit in a lore git repository. +enum LoreCommitParseOutcome { + /// Normal commit containing an `'m'` email file. + Email(Box), + /// Administrative commit without an `'m'` file (e.g. a `public-inbox` `"rm"` + /// deletion commit containing `'d'`). Its SHA should still be recorded in + /// `lore_indexed_commits` so it is not re-processed on every subsequent run. + NoEmailFile(String), +} + +/// Parse a lore commit into either an email (`'m'` file) or `NoEmailFile` if the +/// commit tree has no `'m'` entry (such as a `public-inbox` `"rm"` commit). +fn parse_lore_commit_entry( + repo: &gix::Repository, + commit_sha: &str, +) -> Result { + let commit = resolve_to_commit(repo, commit_sha)?; + let tree = commit.tree()?; + let Some(m_entry) = tree.find_entry("m") else { + debug!( + "Commit {} has no 'm' file (likely a public-inbox 'rm' commit); marking as indexed", + commit_sha + ); + return Ok(LoreCommitParseOutcome::NoEmailFile(commit_sha.to_string())); + }; + + let object = m_entry.object()?; + let blob = object.try_into_blob()?; + let email_content = String::from_utf8_lossy(blob.data.as_slice()); + parse_email_from_blob_content(commit_sha, &email_content) + .map(|info| LoreCommitParseOutcome::Email(Box::new(info))) +} + +/// Process lore commits using a streaming pipeline. +/// Similar to process_commits_pipeline but parses email from 'm' file. +/// Returns `(inserted_emails, recorded_commits)` where `recorded_commits` +/// includes both stored email commits and administrative (`rm`) commits +/// recorded in `lore_indexed_commits`. pub async fn process_lore_commits_pipeline( repo_path: &std::path::Path, commit_shas: Vec, @@ -818,10 +852,16 @@ pub async fn process_lore_commits_pipeline( batch_size: usize, num_workers: usize, num_inserters: usize, -) -> Result<()> { +) -> Result<(usize, usize)> { use indicatif::{ProgressBar, ProgressStyle}; use std::sync::Mutex; + // Cap git worker threads so we don't open 100+ separate gix::Repository + // packfile mmaps/caches on high-core-count machines. + let batch_size = batch_size.max(1); + let num_workers = num_workers.clamp(1, 16); + let num_inserters = num_inserters.max(1); + // Create progress bar let pb = ProgressBar::new(commit_shas.len() as u64); pb.set_style( @@ -832,9 +872,15 @@ pub async fn process_lore_commits_pipeline( .progress_chars("█▓▒░ ") ); - // Create channels for streaming + // Workers send individual parsed items into a bounded channel (`batch_size * 2`), + // and a single collector thread assembles full `batch_size` batches for the + // database inserters. This prevents `num_workers` threads from each hoarding + // up to `batch_size` emails in RAM simultaneously (`num_workers * batch_size` + // emails) and flushing `num_workers` fragmented batches when `commit_tx` closes. let (commit_tx, commit_rx) = mpsc::channel::(); - let (result_tx, result_rx) = mpsc::sync_channel::>(10); + let (item_tx, item_rx) = mpsc::sync_channel::(batch_size * 2); + let (result_tx, result_rx) = + tokio::sync::mpsc::channel::<(Vec, Vec)>(4); // Wrap receiver for shared access let shared_commit_rx = Arc::new(Mutex::new(commit_rx)); @@ -842,6 +888,7 @@ pub async fn process_lore_commits_pipeline( // Shared counters let processed_count = Arc::new(AtomicUsize::new(0)); let inserted_count = Arc::new(AtomicUsize::new(0)); + let recorded_commits_count = Arc::new(AtomicUsize::new(0)); // Spawn progress updater thread let pb_clone = pb.clone(); @@ -859,7 +906,7 @@ pub async fn process_lore_commits_pipeline( let mut worker_handles = Vec::new(); for worker_id in 0..num_workers { let worker_commit_rx = shared_commit_rx.clone(); - let worker_result_tx = result_tx.clone(); + let worker_item_tx = item_tx.clone(); let worker_repo_path = repo_path.to_path_buf(); let worker_processed = processed_count.clone(); @@ -873,10 +920,7 @@ pub async fn process_lore_commits_pipeline( } }; - let mut batch = Vec::new(); - loop { - // Get next commit SHA let commit_sha = { match worker_commit_rx.lock() { Ok(rx) => rx.recv(), @@ -889,17 +933,11 @@ pub async fn process_lore_commits_pipeline( match commit_sha { Ok(sha) => { - match parse_email_from_commit(&thread_repo, &sha) { - Ok(email_info) => { - batch.push(email_info); + match parse_lore_commit_entry(&thread_repo, &sha) { + Ok(outcome) => { worker_processed.fetch_add(1, Ordering::Relaxed); - - // Send batch when it reaches batch_size - if batch.len() >= batch_size { - if worker_result_tx.send(batch.clone()).is_err() { - break; - } - batch.clear(); + if worker_item_tx.send(outcome).is_err() { + break; } } Err(e) => { @@ -911,19 +949,48 @@ pub async fn process_lore_commits_pipeline( } } } - Err(_) => { - // Channel closed, send remaining batch - if !batch.is_empty() { - let _ = worker_result_tx.send(batch); - } - break; - } + Err(_) => break, } } }); worker_handles.push(handle); } + drop(shared_commit_rx); + drop(item_tx); + + // Single batch-collector thread to assemble uniform `batch_size` batches + let collector_handle = thread::spawn(move || { + let mut emails_batch = Vec::with_capacity(batch_size); + let mut non_email_shas = Vec::new(); + + for outcome in item_rx { + match outcome { + LoreCommitParseOutcome::Email(email_info) => { + emails_batch.push(*email_info); + } + LoreCommitParseOutcome::NoEmailFile(sha) => { + non_email_shas.push(sha); + } + } + + if emails_batch.len() + non_email_shas.len() >= batch_size { + let send_emails = std::mem::replace( + &mut emails_batch, + Vec::with_capacity(batch_size), + ); + let send_non_email = std::mem::take(&mut non_email_shas); + if result_tx.blocking_send((send_emails, send_non_email)).is_err() { + return; + } + } + } + + if !emails_batch.is_empty() || !non_email_shas.is_empty() { + let _ = result_tx.blocking_send((emails_batch, non_email_shas)); + } + }); + // Spawn producer thread to feed commit SHAs let producer_handle = thread::spawn(move || { for commit_sha in commit_shas { @@ -933,11 +1000,8 @@ pub async fn process_lore_commits_pipeline( } }); - // Close original senders - drop(result_tx); - - // Wrap result receiver for shared access - let shared_result_rx = Arc::new(Mutex::new(result_rx)); + // Wrap async result receiver for shared access across inserter tasks + let shared_result_rx = Arc::new(tokio::sync::Mutex::new(result_rx)); // Shared flag to coordinate periodic optimization checks // Only one inserter should check at a time to avoid redundant checks @@ -950,6 +1014,7 @@ pub async fn process_lore_commits_pipeline( for inserter_id in 0..num_inserters { let db_manager_clone = Arc::clone(&db_manager); let inserted_clone = inserted_count.clone(); + let recorded_clone = recorded_commits_count.clone(); let pb_clone = pb.clone(); let result_rx_clone = shared_result_rx.clone(); let batches_counter = batches_inserted.clone(); @@ -957,59 +1022,65 @@ pub async fn process_lore_commits_pipeline( let handle = tokio::spawn(async move { loop { - // Get next batch from shared receiver + // Get next batch from shared async receiver without blocking Tokio threads let batch = { - match result_rx_clone.lock() { - Ok(rx) => rx.recv(), - Err(e) => { - error!("Inserter {} failed to lock receiver: {}", inserter_id, e); - break; - } - } + let mut rx = result_rx_clone.lock().await; + rx.recv().await }; match batch { - Ok(emails) => { + Some((emails, non_email_shas)) => { // Insert batch into database match db_manager_clone.insert_lore_emails(&emails).await { Err(e) => { error!("Inserter {} failed to insert batch: {}", inserter_id, e); } Ok(failed_indices) => { - let failed_set: std::collections::HashSet = - failed_indices.into_iter().collect(); - let success_count = emails.len() - failed_set.len(); - - // Record commit SHAs only for emails that - // were actually stored, so that failed - // emails are retried on the next run. - let shas: Vec = emails + let failed_msg_ids: std::collections::HashSet<&str> = + failed_indices + .iter() + .map(|&idx| emails[idx].message_id.as_str()) + .collect(); + let success_count = emails .iter() - .enumerate() - .filter(|(i, _)| !failed_set.contains(i)) - .map(|(_, e)| e.git_commit_sha.as_str()) - .collect::>() - .into_iter() - .map(String::from) + .filter(|e| !failed_msg_ids.contains(e.message_id.as_str())) + .count(); + + // Record commit SHAs for emails that were stored plus + // any administrative (`rm`) commits without an `'m'` file. + let mut shas_set: std::collections::HashSet = emails + .iter() + .filter(|e| !failed_msg_ids.contains(e.message_id.as_str())) + .map(|e| e.git_commit_sha.clone()) .collect(); + for sha in &non_email_shas { + shas_set.insert(sha.clone()); + } + let shas: Vec = shas_set.into_iter().collect(); + let recorded_in_batch = shas.len(); + if !shas.is_empty() { - if let Err(e) = - db_manager_clone.insert_lore_indexed_commits(&shas).await + match db_manager_clone.insert_lore_indexed_commits(&shas).await { - error!( - "Inserter {} failed to record indexed commits: {}", - inserter_id, e - ); + Ok(()) => { + let count = inserted_clone + .fetch_add(success_count, Ordering::Relaxed); + recorded_clone + .fetch_add(recorded_in_batch, Ordering::Relaxed); + pb_clone.set_message(format!( + "Inserted {} emails", + count + success_count + )); + } + Err(e) => { + error!( + "Inserter {} failed to record indexed commits: {}", + inserter_id, e + ); + } } } - let count = - inserted_clone.fetch_add(success_count, Ordering::Relaxed); - pb_clone.set_message(format!( - "Inserted {} emails", - count + success_count - )); - // Track successful batch insertions and check for periodic optimization let total_batches = batches_counter.fetch_add(1, Ordering::Relaxed) + 1; @@ -1023,7 +1094,7 @@ pub async fn process_lore_commits_pipeline( } } } - Err(_) => { + None => { // Channel closed break; } @@ -1033,29 +1104,36 @@ pub async fn process_lore_commits_pipeline( inserter_handles.push(handle); } - // Wait for all workers to finish - for handle in worker_handles { - let _ = handle.join(); - } + drop(shared_result_rx); - // Wait for producer - let _ = producer_handle.join(); + // Join OS worker, collector, and producer threads on the blocking pool + // so Tokio runtime threads remain free to drive the inserter futures. + let _ = tokio::task::spawn_blocking(move || { + for handle in worker_handles { + let _ = handle.join(); + } + let _ = collector_handle.join(); + let _ = producer_handle.join(); + }) + .await; // Wait for all inserters to finish for handle in inserter_handles { let _ = handle.await; } + let final_inserted = inserted_count.load(Ordering::Relaxed); + let final_recorded = recorded_commits_count.load(Ordering::Relaxed); pb.finish_with_message(format!( - "Processed {} emails, inserted {}", + "Processed {} commits, inserted {} emails", processed_count.load(Ordering::Relaxed), - inserted_count.load(Ordering::Relaxed) + final_inserted )); // Wait for progress thread progress_thread.join().unwrap(); - Ok(()) + Ok((final_inserted, final_recorded)) } /// Index commits in a git range