From fde5526a09aa5b2ae6465a07639b17e0f53bf63f Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:07:14 -0800 Subject: [PATCH 1/3] lore: append indexed commit SHAs instead of merging them insert_lore_indexed_commits() recorded each batch's SHAs with a merge_insert keyed on git_commit_sha. lore_indexed_commits has no index, so every merge_insert hash-joined the batch against the whole table. The join grew with every batch, and on a 316-CPU host most of the indexer's busy threads sat in that join while the email parsers waited on the inserters. The merge buys nothing here. The pipeline only hands it SHAs that were filtered against get_indexed_lore_commits() before parsing, and the readers treat the column as a set, so a rare duplicate row is harmless. Append with table.add() instead. --- src/database/connection.rs | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/src/database/connection.rs b/src/database/connection.rs index 22d8d48..0605df8 100644 --- a/src/database/connection.rs +++ b/src/database/connection.rs @@ -8543,6 +8543,12 @@ impl DatabaseManager { } /// Record git commit SHAs that have been processed for lore indexing. + /// + /// This is a plain append, not a merge_insert. The callers only pass + /// SHAs that were filtered against get_indexed_lore_commits(), and + /// readers treat the column as a set, so a rare duplicate row is + /// harmless. merge_insert on this unindexed table hash-joins + /// against every existing row, so its cost grows with the table. pub async fn insert_lore_indexed_commits(&self, commit_shas: &[String]) -> Result<()> { use arrow::array::{ArrayRef, StringArray}; use arrow::datatypes::{DataType, Field, Schema}; @@ -8561,21 +8567,14 @@ impl DatabaseManager { let columns: Vec = vec![Arc::new(StringArray::from(commit_shas.to_vec()))]; - let batch = RecordBatch::try_new(schema.clone(), columns)?; - let batches = vec![Ok(batch)]; - let batch_iterator = - arrow::record_batch::RecordBatchIterator::new(batches.into_iter(), schema); + let batch = RecordBatch::try_new(schema, columns)?; let table = self .connection .open_table("lore_indexed_commits") .execute() .await?; - let mut merge_insert = table.merge_insert(&["git_commit_sha"]); - merge_insert - .when_matched_update_all(None) - .when_not_matched_insert_all(); - merge_insert.execute(Box::new(batch_iterator)).await?; + table.add(vec![batch]).execute().await?; Ok(()) } From dd978581fb2b9d52a53c7261923f596de8df486c Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:07:14 -0800 Subject: [PATCH 2/3] lore: check for duplicate message_ids in memory insert_lore_emails() filtered every batch through a `message_id IN (...)` query with one literal per email. lance plans that predicate once per fragment, and the BTree index on message_id only covers fragments that existed when it was last built, so each batch also scanned every fragment appended during the run. The cost of a batch grew with fragment count times batch size, and the indexer spent its time building and freeing DataFusion literal expressions. The check is still needed: a crossposted email has the same message_id in every list's archive, and a partially failed run can leave stored rows whose commit SHA was never recorded. So load the table's message_ids into a map with one unfiltered column scan (get_lore_message_ids()) when the first archive of a run with new mail needs them, and share the map between the inserters of every archive for the rest of the run. Once a pipeline's inserters finish, the map holds exactly the ids stored so far, this run's included, so a crosspost of an email stored from an earlier archive of the run is still caught. A run in which no archive has new commits never scans the column. LoreIdState, the map's value type, is exported next to DatabaseManager. Within an archive, a message_id can appear in more than one commit, and those copies can land in batches that different inserters store at the same time. Each batch claims its new ids as pending under the map's lock before appending, marks them stored afterwards, and releases the ids of emails that failed so another copy can still be stored. A copy whose id another batch has pending is deferred rather than skipped: it is reported with the failed emails, so its commit SHA is not recorded and the next run checks it against the table. Skipping it would record the SHA even if the other batch's insert then failed. The map costs one string per stored email. On a 316-CPU host, with the other two changes in this series applied, this took lore inserts from about 85 to 400-660 batches a minute. --- src/bin/index.rs | 14 +- src/database/connection.rs | 137 ++++++++++++------ src/database/mod.rs | 2 +- src/indexer.rs | 283 ++++++++++++++++++++++++++++++++++++- src/lib.rs | 2 +- 5 files changed, 389 insertions(+), 49 deletions(-) diff --git a/src/bin/index.rs b/src/bin/index.rs index aaa1fa3..660498f 100644 --- a/src/bin/index.rs +++ b/src/bin/index.rs @@ -3,7 +3,7 @@ use anyhow::Result; use clap::Parser; use colored::Colorize; use semcode::indexer::{ - list_shas_in_range, process_commits_pipeline, process_lore_commits_pipeline, + list_shas_in_range, process_commits_pipeline, process_lore_commits_pipeline, KnownLoreIds, }; use semcode::{measure, process_database_path, CodeVectorizer, DatabaseManager}; // Temporary call relationships are now embedded in function JSON columns @@ -682,6 +682,8 @@ struct LoreIndexResult { /// Index a lore archive repository, filtering out already-indexed commits. /// This is the common logic shared by both --lore and --refresh-lore handlers. +/// Every archive of one run shares `known_ids`. +#[allow(clippy::too_many_arguments)] async fn index_lore_archive( lore_repo: gix::Repository, archive_path: &Path, @@ -690,6 +692,7 @@ async fn index_lore_archive( batch_size: usize, num_workers: usize, db_threads: usize, + known_ids: &KnownLoreIds, ) -> Result { // Get the remote tracking branch to include all fetched commits // Try refs/remotes/origin/master first, then refs/remotes/origin/main as fallback @@ -764,6 +767,7 @@ async fn index_lore_archive( batch_size, num_workers, db_threads, + known_ids, ) .await?; @@ -1117,6 +1121,7 @@ async fn main() -> Result<()> { let mut total_new_emails = 0usize; let mut total_emails_all_archives = 0usize; let mut failed_archives: Vec<(String, String)> = Vec::new(); + let known_ids = KnownLoreIds::new(); // Process each lore archive for lore_url in lore_args { @@ -1153,6 +1158,7 @@ async fn main() -> Result<()> { batch_size, num_workers, args.db_threads, + &known_ids, ) .await; @@ -1170,6 +1176,8 @@ async fn main() -> Result<()> { } } } + // One string per stored email: free it before compaction. + drop(known_ids); let total_time = start_time.elapsed(); @@ -1300,6 +1308,7 @@ async fn main() -> Result<()> { // causing both to stall indefinitely. let mut results: Vec> = Vec::with_capacity(total_archives); + let known_ids = KnownLoreIds::new(); for (archive_path, display_name) in archives_with_names { println!("\n=== Refreshing lore archive: {} ===", display_name); @@ -1321,6 +1330,7 @@ async fn main() -> Result<()> { batch_size, num_workers, db_threads, + &known_ids, ) .await; @@ -1331,6 +1341,8 @@ async fn main() -> Result<()> { Err(e) => results.push(Err((display_name, e))), } } + // One string per stored email: free it before compaction. + drop(known_ids); // Aggregate results let mut total_new_emails = 0usize; diff --git a/src/database/connection.rs b/src/database/connection.rs index 0605df8..23c3092 100644 --- a/src/database/connection.rs +++ b/src/database/connection.rs @@ -150,6 +150,16 @@ fn log_absence(kind: &str, name: &str, git_sha: &str, why: Absent, elsewhere: &s /// moves the generation and drops every entry). type CallerMemoKey = (String, String, u64); +/// Where a message_id stands in the duplicate check that the lore +/// inserters share (see DatabaseManager::insert_lore_emails()). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LoreIdState { + /// The email is in the lore table. + Stored, + /// A batch has claimed the id and its append has not finished. + Pending, +} + pub struct DatabaseManager { connection: Connection, git_repo_path: String, @@ -6574,11 +6584,22 @@ impl DatabaseManager { /// any emails that could not be stored. Callers must not record /// commit SHAs for failed indices in the lore_indexed_commits /// table, otherwise those emails are permanently lost. + /// + /// `known_ids` holds every message_id already in the lore table + /// (see get_lore_message_ids()) and is shared by all inserters. + /// Emails whose message_id is stored are skipped. New ids are + /// claimed as pending before the append, then marked stored, or + /// released if their email failed. An email whose id another + /// inserter has pending is deferred: it is returned with the failed + /// indices, so its SHA is not recorded and the next run retries it + /// against the table. pub async fn insert_lore_emails( &self, emails: &[crate::types::LoreEmailInfo], + known_ids: &std::sync::Mutex>, ) -> Result> { use arrow::datatypes::{DataType, Field, Schema}; + use std::collections::hash_map::Entry; use std::sync::Arc; if emails.is_empty() { @@ -6629,27 +6650,58 @@ impl DatabaseManager { let mut failed_indices: Vec = Vec::new(); // Filter out emails whose message_id already exists in the - // table. Normally every email here is new because - // index_lore_archive skips already-indexed commits, but a - // partial failure on a previous run can leave rows in the - // table whose commit SHA was never recorded in - // 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?; - - if new_indices.len() < dedup_indices.len() { + // table. A crossposted email has the same message_id in + // every list's archive, and a partial failure on a previous + // run can leave rows in the table whose commit SHA was never + // recorded in lore_indexed_commits. Filtering avoids + // duplicates without relying on merge_insert, which hits a + // spurious null-column error in lance-file 3.0.1. + // + // The check runs against an in-memory set rather than a + // `message_id IN (...)` query: lance plans that predicate once + // per fragment, and a filtered query costs a scan of every + // fragment appended since the last index update. Crossposts + // are caught because one set, loaded from the table once per + // run, is shared by the inserters of every archive in the run. + // An id is claimed under the lock so that two batches holding + // the same message_id cannot both store it. + let mut new_indices: Vec = Vec::new(); + let mut deferred: Vec = Vec::new(); + { + let mut known = known_ids.lock().unwrap_or_else(|e| e.into_inner()); + for &i in &dedup_indices { + match known.entry(emails[i].message_id.clone()) { + Entry::Vacant(slot) => { + slot.insert(LoreIdState::Pending); + new_indices.push(i); + } + Entry::Occupied(slot) if *slot.get() == LoreIdState::Pending => { + deferred.push(i); + } + Entry::Occupied(_) => {} + } + } + } + + let skipped = dedup_indices.len() - new_indices.len() - deferred.len(); + if skipped > 0 { tracing::info!( "insert_lore_emails: {} of {} already in table, inserting {}", - dedup_indices.len() - new_indices.len(), + skipped, dedup_indices.len(), new_indices.len(), ); } + if !deferred.is_empty() { + tracing::info!( + "insert_lore_emails: deferring {} emails another batch is storing", + deferred.len() + ); + } if new_indices.is_empty() { tracing::info!("insert_lore_emails: no new emails to insert"); - return Ok(failed_indices); + return Ok(deferred); } // Use table.add() instead of merge_insert. merge_insert in @@ -6657,7 +6709,7 @@ impl DatabaseManager { // internal join/write phase, causing every insert to fail // with "subject contained null values". Plain append avoids // the merge codepath entirely. Duplicates are prevented by - // the filter_existing_lore_ids check above. + // the known_ids check above. if let Err(e) = Self::add_lore_chunk(&table, emails, &new_indices, &schema).await { tracing::warn!( "insert_lore_emails: full batch of {} failed ({}), \ @@ -6700,60 +6752,59 @@ impl DatabaseManager { ); } + // Mark the claimed ids stored, and release those whose email + // failed so that another copy of it can still be stored. + { + let mut known = known_ids.lock().unwrap_or_else(|e| e.into_inner()); + for &idx in &new_indices { + if let Some(state) = known.get_mut(&emails[idx].message_id) { + *state = LoreIdState::Stored; + } + } + for &idx in &failed_indices { + known.remove(&emails[idx].message_id); + } + } + tracing::info!("insert_lore_emails: Batch insertion complete"); + failed_indices.extend(deferred); Ok(failed_indices) } - /// Return the subset of `indices` whose message_id does not - /// already exist in the lore table. - async fn filter_existing_lore_ids( - 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(", ")); + /// Return every message_id in the lore table, marked stored, for the + /// duplicate check in insert_lore_emails(). One unfiltered column + /// scan is far cheaper than a filtered query per batch. + pub async fn get_lore_message_ids(&self) -> Result> { + let table = self.connection.open_table("lore").execute().await?; - let stream = table + let mut 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 existing = std::collections::HashSet::new(); - for batch in &batches { + // Consume batch by batch, so the Arrow copy of the column is + // never held in memory next to the map. + let mut existing = HashMap::new(); + while let Some(batch) = stream.try_next().await? { if let Some(col) = batch.column_by_name("message_id") { let arr = col .as_any() - .downcast_ref::() + .downcast_ref::() .expect("message_id column must be StringArray"); + existing.reserve(arr.len()); for i in 0..arr.len() { if !arr.is_null(i) { - existing.insert(arr.value(i).to_string()); + existing.insert(arr.value(i).to_string(), LoreIdState::Stored); } } } } - Ok(indices - .iter() - .copied() - .filter(|&i| !existing.contains(&emails[i].message_id)) - .collect()) + Ok(existing) } /// Build a [`RecordBatch`] from the given email indices and diff --git a/src/database/mod.rs b/src/database/mod.rs index 468de7f..3ab5eaa 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs @@ -21,7 +21,7 @@ mod symbol_filename; mod types; mod vectors; -pub use connection::DatabaseManager; +pub use connection::{DatabaseManager, LoreIdState}; use anyhow::Result; use arrow::array::RecordBatch; diff --git a/src/indexer.rs b/src/indexer.rs index 2752ea1..2f7a096 100644 --- a/src/indexer.rs +++ b/src/indexer.rs @@ -6,7 +6,7 @@ use anyhow::Result; use gix::revision::walk::Sorting; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc; @@ -15,7 +15,7 @@ use std::thread; use tracing::{debug, error, info, warn}; use crate::git::resolve_to_commit; -use crate::{DatabaseManager, GitCommitInfo}; +use crate::{DatabaseManager, GitCommitInfo, LoreIdState}; /// Perform periodic optimization check during indexing operations /// Returns true if optimization was performed, false otherwise @@ -809,8 +809,46 @@ pub async fn process_commits_pipeline( Ok(()) } +/// The duplicate check the lore inserters share: message_id to its state. +pub type LoreIdMap = Arc>>; + +/// message_ids already in the lore table, shared by every archive of one +/// lore run and loaded when the first archive with new mail needs them. +/// Between pipelines every entry is `LoreIdState::Stored`: inserters mark +/// what they stored and remove what failed before they exit. The +/// exception is an inserter that panics between claiming ids and marking +/// them. Its ids stay `Pending` for the rest of the run, so later +/// archives defer those emails and the next run retries them; debug +/// builds fail an assertion at the end of the pipeline instead. +#[derive(Default)] +pub struct KnownLoreIds(tokio::sync::OnceCell); + +impl KnownLoreIds { + pub fn new() -> Self { + Self::default() + } + + /// The shared map, scanning the lore table's message_ids on the first + /// call only. A failed scan leaves it unloaded, so the next call + /// retries. + pub async fn get(&self, db: &DatabaseManager) -> Result { + let ids = self + .0 + .get_or_try_init(|| async { + let ids = db.get_lore_message_ids().await?; + info!("Loaded {} existing lore message_ids", ids.len()); + Ok::<_, anyhow::Error>(Arc::new(std::sync::Mutex::new(ids))) + }) + .await?; + Ok(Arc::clone(ids)) + } +} + /// Process lore commits using a streaming pipeline /// Similar to process_commits_pipeline but parses email from 'm' file +/// +/// `known_ids` is the duplicate check for crossposts and partially failed +/// runs; pass the same one for every archive of a run. pub async fn process_lore_commits_pipeline( repo_path: &std::path::Path, commit_shas: Vec, @@ -818,10 +856,13 @@ pub async fn process_lore_commits_pipeline( batch_size: usize, num_workers: usize, num_inserters: usize, + known_ids: &KnownLoreIds, ) -> Result<()> { use indicatif::{ProgressBar, ProgressStyle}; use std::sync::Mutex; + let known_ids = known_ids.get(&db_manager).await?; + // Create progress bar let pb = ProgressBar::new(commit_shas.len() as u64); pb.set_style( @@ -954,6 +995,7 @@ pub async fn process_lore_commits_pipeline( let result_rx_clone = shared_result_rx.clone(); let batches_counter = batches_inserted.clone(); let optimization_check_timer = last_optimization_check.clone(); + let known_ids = known_ids.clone(); let handle = tokio::spawn(async move { loop { @@ -971,7 +1013,10 @@ pub async fn process_lore_commits_pipeline( match batch { Ok(emails) => { // Insert batch into database - match db_manager_clone.insert_lore_emails(&emails).await { + match db_manager_clone + .insert_lore_emails(&emails, &known_ids) + .await + { Err(e) => { error!("Inserter {} failed to insert batch: {}", inserter_id, e); } @@ -1046,6 +1091,15 @@ pub async fn process_lore_commits_pipeline( let _ = handle.await; } + debug_assert!( + known_ids + .lock() + .unwrap_or_else(|e| e.into_inner()) + .values() + .all(|state| *state == LoreIdState::Stored), + "lore message_id map holds pending ids after every inserter finished" + ); + pb.finish_with_message(format!( "Processed {} emails, inserted {}", processed_count.load(Ordering::Relaxed), @@ -1286,4 +1340,227 @@ mod tests { assert_eq!(expected, got, "HEAD (no range) should return all 5 commits"); } + + async fn lore_database() -> (tempfile::TempDir, Arc) { + let dir = tempfile::tempdir().unwrap(); + let db = DatabaseManager::new( + dir.path().join(".semcode.db").to_str().unwrap(), + dir.path().to_string_lossy().into_owned(), + ) + .await + .unwrap(); + db.create_tables().await.unwrap(); + (dir, Arc::new(db)) + } + + fn lore_email(git_commit_sha: &str, message_id: &str) -> crate::LoreEmailInfo { + crate::LoreEmailInfo { + git_commit_sha: git_commit_sha.to_string(), + from: "A U Thor ".to_string(), + date: "Mon, 2 Nov 2026 10:00:00 +0000".to_string(), + date_timestamp: 1_793_613_600, + message_id: message_id.to_string(), + in_reply_to: None, + subject: "[PATCH] test".to_string(), + references: None, + recipients: "list@example.org".to_string(), + body: "body".to_string(), + symbols: Vec::new(), + } + } + + async fn lore_rows(db: &DatabaseManager, filter: Option) -> usize { + db.connection() + .open_table("lore") + .execute() + .await + .unwrap() + .count_rows(filter) + .await + .unwrap() + } + + /// The map is scanned once: a second get() returns the same map and + /// does not see a row stored behind its back. + #[tokio::test] + async fn known_lore_ids_loads_once() { + let (_dir, db) = lore_database().await; + let side_map = std::sync::Mutex::new(std::collections::HashMap::new()); + let failed = db + .insert_lore_emails( + &[lore_email("1111", ""), lore_email("2222", "")], + &side_map, + ) + .await + .unwrap(); + assert!(failed.is_empty()); + + let known = KnownLoreIds::new(); + let first = known.get(&db).await.unwrap(); + { + let ids = first.lock().unwrap(); + assert_eq!(ids.len(), 2); + assert_eq!(ids.get(""), Some(&LoreIdState::Stored)); + assert_eq!(ids.get(""), Some(&LoreIdState::Stored)); + } + + let failed = db + .insert_lore_emails(&[lore_email("3333", "")], &side_map) + .await + .unwrap(); + assert!(failed.is_empty()); + + let second = known.get(&db).await.unwrap(); + assert!(Arc::ptr_eq(&first, &second)); + assert!(!second.lock().unwrap().contains_key("")); + + // The row is in the table: a fresh map loads it. + let fresh = KnownLoreIds::new().get(&db).await.unwrap(); + assert!(fresh.lock().unwrap().contains_key("")); + } + + /// A batch holding an id that another batch has pending defers that + /// email: its index is returned as not stored, no row is written for + /// it, and the other batch's claim is left alone. The rest of the + /// batch is stored. + #[tokio::test] + async fn pending_id_is_deferred() { + let (_dir, db) = lore_database().await; + let map = std::sync::Mutex::new(HashMap::from([( + "".to_string(), + LoreIdState::Pending, + )])); + + let failed = db + .insert_lore_emails( + &[ + lore_email("4444", ""), + lore_email("5555", ""), + ], + &map, + ) + .await + .unwrap(); + + assert_eq!(failed, vec![0]); + assert_eq!( + lore_rows(&db, Some("message_id = ''".to_string())).await, + 0 + ); + assert_eq!( + lore_rows(&db, Some("message_id = ''".to_string())).await, + 1 + ); + let map = map.lock().unwrap(); + assert_eq!(map.get(""), Some(&LoreIdState::Pending)); + assert_eq!(map.get(""), Some(&LoreIdState::Stored)); + } + + /// A failed scan leaves the map unloaded: the next get() scans again. + #[tokio::test] + async fn known_lore_ids_retries_after_failed_load() { + let dir = tempfile::tempdir().unwrap(); + let db = DatabaseManager::new( + dir.path().join(".semcode.db").to_str().unwrap(), + dir.path().to_string_lossy().into_owned(), + ) + .await + .unwrap(); + let known = KnownLoreIds::new(); + + // No lore table yet. + assert!(known.get(&db).await.is_err()); + + db.create_tables().await.unwrap(); + let side_map = std::sync::Mutex::new(HashMap::new()); + let failed = db + .insert_lore_emails(&[lore_email("6666", "")], &side_map) + .await + .unwrap(); + assert!(failed.is_empty()); + + let ids = known.get(&db).await.unwrap(); + assert_eq!( + ids.lock().unwrap().get(""), + Some(&LoreIdState::Stored) + ); + } + + /// Commit one RFC 2822 message as file `m`, the layout of a lore + /// archive, and return the commit's SHA. + fn commit_lore_message( + repo: &std::path::Path, + message_id: &str, + commit_message: &str, + ) -> String { + std::fs::write( + repo.join("m"), + format!( + "From: A U Thor \n\ + Date: Mon, 2 Nov 2026 10:00:00 +0000\n\ + Message-ID: {message_id}\n\ + Subject: [PATCH] test\n\ + To: list@example.org\n\ + \n\ + body\n" + ), + ) + .unwrap(); + git(repo, &["add", "m"]); + git(repo, &["commit", "-q", "-m", commit_message]); + git(repo, &["rev-parse", "HEAD"]) + } + + /// A crosspost appears in two archives of one run. The second + /// archive's pipeline must see the id the first one stored, store no + /// second row, and still record its commit as indexed. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn crossposted_email_is_stored_once_across_archives() { + let (_dir, db) = lore_database().await; + + let archive_a = tempfile::tempdir().unwrap(); + git(archive_a.path(), &["init", "-q", "-b", "master"]); + let cross_a = commit_lore_message(archive_a.path(), "", "list a"); + + let archive_b = tempfile::tempdir().unwrap(); + git(archive_b.path(), &["init", "-q", "-b", "master"]); + let cross_b = commit_lore_message(archive_b.path(), "", "list b"); + let only_b = commit_lore_message(archive_b.path(), "", "list b only"); + assert_ne!(cross_a, cross_b); + + let known = KnownLoreIds::new(); + process_lore_commits_pipeline( + archive_a.path(), + vec![cross_a.clone()], + db.clone(), + 1024, + 1, + 1, + &known, + ) + .await + .unwrap(); + process_lore_commits_pipeline( + archive_b.path(), + vec![cross_b.clone(), only_b.clone()], + db.clone(), + 1024, + 1, + 1, + &known, + ) + .await + .unwrap(); + + assert_eq!( + lore_rows(&db, Some("message_id = ''".to_string())).await, + 1 + ); + assert_eq!(lore_rows(&db, None).await, 2); + + let indexed = db.get_indexed_lore_commits().await.unwrap(); + for sha in [&cross_a, &cross_b, &only_b] { + assert!(indexed.contains(sha), "{sha} not recorded as indexed"); + } + } } diff --git a/src/lib.rs b/src/lib.rs index 102d6ee..ae7281a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,8 +32,8 @@ pub mod search; // Re-export the main types and structs pub use database::processed_files::ProcessedFileRecord; pub use database::resolution::{Evidence, IndirectCaller}; -pub use database::DatabaseManager; pub use database::SCHEMA_VERSION; +pub use database::{DatabaseManager, LoreIdState}; pub use database_utils::process_database_path; pub use git::{get_git_sha, get_git_sha_for_workdir}; pub use hash::{compute_content_hash, compute_file_hash}; From 0fad4197888dbf0f2568c799df637822afaf1e2c Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:07:14 -0800 Subject: [PATCH 3/3] index: cap tokio workers and lore parser threads at 32 #[tokio::main] starts one worker per CPU, and the lore pipeline started one email parser thread per CPU. On a 316-CPU host that is over 600 threads for a job limited by two database inserters: the parsers sat blocked on the result channel, and the process spent most of its CPU time in the kernel, in futex and spinlock contention from idle tokio workers parking and waking. Build the runtime by hand with min(num_cpus, 32) workers, still letting TOKIO_WORKER_THREADS override it as it does for the default runtime, and cap the lore parser threads the same way. The rayon pool is left alone; -j already sizes it. Machines with 32 CPUs or fewer see no change. The other modes do their CPU work on rayon, std threads or spawn_blocking, so the tokio cap does not touch them. The exception is lance's FTS index build, which tokenizes on tokio workers: a full lore FTS (re)build on a large host is now limited to 32 workers, and TOKIO_WORKER_THREADS lifts that. The database inserters are another matter. The inserter tasks of the --git, tree, --commits and lore pipelines are tokio tasks that block their worker in a std channel recv() while they wait for a batch, so with --db-threads at or above the worker count they can occupy every worker and starve the async work the inserters wait on. That was already reachable at one worker per CPU on small hosts; with the cap it is reachable at 32 inserters on any host. The defect predates this change, and #119 moves those receives off the workers. --- src/bin/index.rs | 33 ++++++++++++++++++++++++++++----- 1 file changed, 28 insertions(+), 5 deletions(-) diff --git a/src/bin/index.rs b/src/bin/index.rs index 660498f..034d10f 100644 --- a/src/bin/index.rs +++ b/src/bin/index.rs @@ -1,5 +1,5 @@ // SPDX-License-Identifier: MIT OR Apache-2.0 -use anyhow::Result; +use anyhow::{Context, Result}; use clap::Parser; use colored::Colorize; use semcode::indexer::{ @@ -13,6 +13,16 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use tracing::{error, info, warn}; +/// Upper bound on tokio worker threads and lore parser threads. One +/// thread per CPU (300+ on large hosts) spends more time in futex and +/// allocator contention than it gains: lore indexing is limited by the +/// database inserters, not by parsing. The other modes do their CPU +/// work on rayon, std threads or spawn_blocking, not tokio workers. +/// The exception is lance's FTS index build, which tokenizes on tokio +/// workers, so a full lore FTS (re)build is capped too; set +/// TOKIO_WORKER_THREADS to lift the cap. +const MAX_THREADS: usize = 32; + #[derive(Parser, Debug)] #[command(name = "semcode-index")] #[command(about = "Index a codebase for semantic analysis", long_about = None)] @@ -1012,8 +1022,21 @@ async fn run_branch_indexing(args: Args, branches: Vec) // ==================== End Branch Indexing Support ==================== -#[tokio::main] -async fn main() -> Result<()> { +fn main() -> Result<()> { + // #[tokio::main] starts one worker per CPU. Cap it, but let + // TOKIO_WORKER_THREADS override as it does for the default runtime. + let mut runtime = tokio::runtime::Builder::new_multi_thread(); + runtime.enable_all(); + if std::env::var_os("TOKIO_WORKER_THREADS").is_none() { + runtime.worker_threads(num_cpus::get().min(MAX_THREADS)); + } + runtime + .build() + .context("failed to build the tokio runtime")? + .block_on(async_main()) +} + +async fn async_main() -> Result<()> { // Suppress ORT verbose logging std::env::set_var("ORT_LOG_LEVEL", "ERROR"); @@ -1116,7 +1139,7 @@ async fn main() -> Result<()> { let start_time = std::time::Instant::now(); let batch_size = 1024; - let num_workers = num_cpus::get(); + let num_workers = num_cpus::get().min(MAX_THREADS); let mut total_new_emails = 0usize; let mut total_emails_all_archives = 0usize; @@ -1298,7 +1321,7 @@ async fn main() -> Result<()> { let start_time = std::time::Instant::now(); let batch_size = 1024; - let num_workers = num_cpus::get(); + let num_workers = num_cpus::get().min(MAX_THREADS); let db_threads = args.db_threads; let total_archives = archives_with_names.len();