diff --git a/Cargo.lock b/Cargo.lock index 768ecb2..a5395bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -123,7 +123,7 @@ dependencies = [ [[package]] name = "bread-theme" version = "0.2.3" -source = "git+https://github.com/Breadway/bread-ecosystem?tag=v0.2.8#5e58558dd36031433d4a8d8e70c71206c3f1f8f4" +source = "git+https://github.com/Breadway/bread-ecosystem?tag=v0.2.10#17d1bb85801b9a8c195b64c02d288cd662c9c780" dependencies = [ "dirs", "gtk4", diff --git a/breadmill/src/indexer.rs b/breadmill/src/indexer.rs index 3ad20b5..539b58c 100644 --- a/breadmill/src/indexer.rs +++ b/breadmill/src/indexer.rs @@ -10,7 +10,7 @@ use ignore::WalkBuilder; use notify::{RecommendedWatcher, RecursiveMode, Watcher, EventKind}; use sha2::{Digest, Sha256}; -use crate::{embed::OrtEmbedder, extract, chunk, power, store::Store}; +use crate::{embed::OrtEmbedder, extract, chunk, power, store::Store, sync_ext::MutexExt}; pub struct SharedState { pub store: Mutex, @@ -64,7 +64,7 @@ impl Indexer { pub fn full_reindex(&self) { eprintln!("breadmill: full reindex triggered"); { - let mut store = self.state.store.lock().unwrap(); + let mut store = self.state.store.lock_recover(); // Clear all state let _ = store.conn.execute_batch("DELETE FROM chunks; DELETE FROM files;"); let _ = store.index.reserve(4096); @@ -87,7 +87,7 @@ impl Indexer { // Snapshot existing indexed files let known: HashMap = { - let store = self.state.store.lock().unwrap(); + let store = self.state.store.lock_recover(); store.all_files() .unwrap_or_default() .into_iter() @@ -155,7 +155,7 @@ impl Indexer { .collect(); if !to_delete.is_empty() { - let mut store = self.state.store.lock().unwrap(); + let mut store = self.state.store.lock_recover(); for path in to_delete { eprintln!("breadmill: removing deleted file: {}", path); let _ = store.delete_file(&path); @@ -163,7 +163,7 @@ impl Indexer { } let count = { - let store = self.state.store.lock().unwrap(); + let store = self.state.store.lock_recover(); let n = store.chunk_count(); let _ = store.save_index(&self.state_dir); n @@ -238,7 +238,7 @@ impl Indexer { self.handle_fs_event(&path); } let count = { - let store = self.state.store.lock().unwrap(); + let store = self.state.store.lock_recover(); let n = store.chunk_count(); let _ = store.save_index(&self.state_dir); n @@ -261,7 +261,7 @@ impl Indexer { if !path.is_file() { let path_str = path.to_string_lossy().into_owned(); // File deleted — remove from index - let mut store = self.state.store.lock().unwrap(); + let mut store = self.state.store.lock_recover(); let _ = store.delete_file(&path_str); return; } @@ -309,7 +309,7 @@ impl Indexer { // Check if hash changed (catches content changes without mtime change) { - let store = self.state.store.lock().unwrap(); + let store = self.state.store.lock_recover(); if let Ok(files) = store.all_files() { if files.iter().any(|f| f.path == path_str && f.hash == hash) { return; @@ -322,20 +322,22 @@ impl Indexer { // for natural-language files. let chunks = chunk::chunk_text(&text, 400, 80, 2_000); eprintln!("breadmill: embedding {} ({} chars, {} chunks)", path_str, text.len(), chunks.len()); - let mut embedder_guard = self.state.embedder.lock().unwrap(); if !self.state.model_ready.load(Ordering::Relaxed) { eprintln!("breadmill: model not ready, skipping embed for {}", path_str); return; } - let embedder = match embedder_guard.as_mut() { - Some(e) => e, - None => return, - }; + // Confirm the embedder is actually present before committing to + // clearing this file's old chunks below — same check as before, + // just without holding the embedder lock past this one glance (see + // the per-chunk locking in the loop for why). + if self.state.embedder.lock_recover().is_none() { + return; + } { - let mut store = self.state.store.lock().unwrap(); + let mut store = self.state.store.lock_recover(); let _ = store.delete_file(path_str); // remove old chunks/vectors first } @@ -344,9 +346,18 @@ impl Indexer { for (i, chunk) in chunks.iter().enumerate() { eprintln!("breadmill: embed chunk {}/{} ({} chars) for {}", i + 1, chunks.len(), chunk.text.len(), path_str); - match embedder.embed_document(&chunk.text) { + // Lock the embedder only around this single chunk's embed call — + // this used to be held for the whole file's chunk loop, so one + // large file needing a fresh MIGraphX JIT compile (60-120s for a + // new sequence length) could block every query (serve.rs locks + // this same mutex) for the entire file, not just one chunk. + let embed_result = match self.state.embedder.lock_recover().as_mut() { + Some(embedder) => embedder.embed_document(&chunk.text), + None => break, // model was unloaded mid-scan; stop here + }; + match embed_result { Ok(embedding) => { - let mut store = self.state.store.lock().unwrap(); + let mut store = self.state.store.lock_recover(); // Ensure file row exists before inserting chunks (FK constraint) let _ = store.upsert_file(path_str, mtime, &hash); let _ = store.insert_chunk( @@ -367,7 +378,7 @@ impl Indexer { eprintln!("breadmill: no chunks embedded for {}", path_str); // Record the file so the mtime+hash check skips it on the next startup // rather than re-entering the same embed-fail loop. - let store = self.state.store.lock().unwrap(); + let store = self.state.store.lock_recover(); let _ = store.upsert_file(path_str, mtime, &hash); } else { // Increment live so `status` reflects progress before the full scan ends. diff --git a/breadmill/src/main.rs b/breadmill/src/main.rs index da5814d..08452e8 100644 --- a/breadmill/src/main.rs +++ b/breadmill/src/main.rs @@ -13,10 +13,12 @@ mod indexer; mod power; mod serve; mod store; +mod sync_ext; use embed::{Backend, OrtEmbedder}; use indexer::{Indexer, SharedState}; use store::Store; +use sync_ext::MutexExt; const MODEL_URL: &str = "https://huggingface.co/nomic-ai/nomic-embed-text-v1.5/resolve/main/onnx/model.onnx"; @@ -170,7 +172,7 @@ fn run_daemon( eprintln!("breadmill: loading model..."); match OrtEmbedder::load(&model_path, &tokenizer_path, dim, backend) { Ok(embedder) => { - *state_clone.embedder.lock().unwrap() = Some(embedder); + *state_clone.embedder.lock_recover() = Some(embedder); state_clone.model_ready.store(true, Ordering::Relaxed); eprintln!("breadmill: model loaded"); } diff --git a/breadmill/src/serve.rs b/breadmill/src/serve.rs index cd67e67..a147235 100644 --- a/breadmill/src/serve.rs +++ b/breadmill/src/serve.rs @@ -8,6 +8,7 @@ use std::{ use breadsearch_shared::{Request, Response, StatusInfo}; use crate::indexer::SharedState; +use crate::sync_ext::MutexExt; pub fn run(socket_path: &Path, state: Arc, snippet_len: usize, search_limit: usize) { let _ = std::fs::remove_file(socket_path); @@ -86,7 +87,7 @@ fn dispatch( } let embedding = { - let mut embedder = state.embedder.lock().unwrap(); + let mut embedder = state.embedder.lock_recover(); match embedder.as_mut() { Some(e) => match e.embed_query(&query) { Ok(v) => v, @@ -101,7 +102,7 @@ fn dispatch( }; let limit = limit.min(search_limit).max(1); - let store = state.store.lock().unwrap(); + let store = state.store.lock_recover(); match store.search(&embedding, limit, snippet_len) { Ok(hits) => Response::Hits { hits }, diff --git a/breadmill/src/store.rs b/breadmill/src/store.rs index 077736d..ed1392a 100644 --- a/breadmill/src/store.rs +++ b/breadmill/src/store.rs @@ -57,9 +57,32 @@ impl Store { let index = new_index(&options).map_err(|e| e.to_string())?; if idx_path.exists() { - index - .load(idx_path.to_str().unwrap()) - .map_err(|e| e.to_string())?; + // NOTE: this only catches corruption that usearch's loader + // itself detects and reports as an `Err` (e.g. a recognizable + // but wrong/incompatible header). Verified experimentally: a + // file that doesn't even look like a usearch index at all (pure + // garbage bytes) crashes the *process* with a SIGSEGV inside the + // native loader rather than returning an `Err` — no amount of + // Rust-side `Result`/`catch_unwind` handling can intercept that, + // it's a native-code robustness gap in the usearch library + // itself. This recovery path is still worth having (it's the + // difference between "won't start" and "rebuilds and starts" + // for the errors it *does* catch), but it is not a complete + // guarantee against every possible corrupt file. The atomic + // save below is the real fix for the common case this was + // written for (a crash mid-save) — it now can't produce a + // half-written `vectors.usearch` in the first place. + if let Err(e) = index.load(idx_path.to_str().unwrap()) { + eprintln!( + "breadmill: WARNING: {} failed to load ({}) — treating it as corrupt, \ + discarding it, and rebuilding the index from scratch on the next scan", + idx_path.display(), + e + ); + conn.execute_batch("DELETE FROM chunks; DELETE FROM files;") + .map_err(|e| e.to_string())?; + index.reserve(4096).map_err(|e| e.to_string())?; + } } else { index.reserve(4096).map_err(|e| e.to_string())?; } @@ -218,11 +241,18 @@ impl Store { // ---- persistence -------------------------------------------------------- + /// Saves to a `.tmp` sibling and renames it into place — a same- + /// filesystem rename is atomic, so a crash mid-save leaves only an + /// orphaned `.tmp` file rather than a truncated/corrupt + /// `vectors.usearch` that would otherwise fail to load on next start + /// (see the recovery path in `open`). pub fn save_index(&self, state_dir: &Path) -> Result<(), String> { let idx_path = state_dir.join("vectors.usearch"); + let tmp_path = state_dir.join("vectors.usearch.tmp"); self.index - .save(idx_path.to_str().unwrap()) - .map_err(|e| e.to_string()) + .save(tmp_path.to_str().unwrap()) + .map_err(|e| e.to_string())?; + std::fs::rename(&tmp_path, &idx_path).map_err(|e| e.to_string()) } } @@ -233,3 +263,45 @@ fn truncate_to_chars(s: &str, max_chars: usize) -> String { let truncated: String = s.chars().take(max_chars).collect(); format!("{}…", truncated.trim_end()) } + +#[cfg(test)] +mod tests { + use super::*; + + fn test_dir(name: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!("breadmill-store-test-{name}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + #[test] + fn save_index_leaves_no_tmp_file_behind_and_is_loadable() { + let dir = test_dir("save-atomic"); + let store = Store::open(&dir, 4).unwrap(); + store.save_index(&dir).unwrap(); + + assert!(dir.join("vectors.usearch").exists()); + assert!( + !dir.join("vectors.usearch.tmp").exists(), + "the .tmp staging file should be renamed away, not left behind" + ); + + // A fresh `open` can load what was just saved without error. + Store::open(&dir, 4).unwrap(); + + std::fs::remove_dir_all(&dir).unwrap(); + } + + // No automated test for the corrupt-index recovery branch in `open`: + // the natural way to construct a "corrupt" fixture (writing arbitrary + // garbage to vectors.usearch) was tried and crashes the *test process* + // with a SIGSEGV inside usearch's native loader before our `Result` + // handling ever gets a chance to run — see the comment on that branch + // in `open`. A real usearch file with a deliberately-broken-but-still- + // parseable header could plausibly hit the `Err` path exercised there + // instead, but reverse-engineering that format precisely enough to + // build a safe fixture wasn't worth the risk of another flaky/crashing + // test. The atomic-save test above covers the actual mechanism that + // prevents this scenario from arising in the first place. +} diff --git a/breadmill/src/sync_ext.rs b/breadmill/src/sync_ext.rs new file mode 100644 index 0000000..a891cd7 --- /dev/null +++ b/breadmill/src/sync_ext.rs @@ -0,0 +1,41 @@ +//! Poison-tolerant `Mutex` locking. +//! +//! A panic on any thread while holding `SharedState::store` or +//! `SharedState::embedder` (e.g. an untrapped ONNX Runtime panic — only PDF +//! extraction is `catch_unwind`-guarded, see `extract.rs`) poisons the +//! `Mutex`. Every subsequent plain `.lock().unwrap()` — in the indexer *and* +//! in every query handler in `serve.rs` — would then immediately panic too, +//! silently bricking indexing and search until the daemon is restarted by +//! hand. +//! +//! `Mutex` poisoning exists to flag "the data guarded by this lock might be +//! in an inconsistent state," but every lock scope in this crate is a short, +//! single-step SQLite call or usearch operation — nothing here spans a +//! multi-step invariant across a single `lock()` call — so recovering the +//! guard and logging loudly is a reasonable trade here: continuing to serve +//! (and re-attempting the operation that panicked, on the next file/query) +//! beats a daemon that silently stops answering everything after one panic. + +use std::sync::{Mutex, MutexGuard}; + +pub trait MutexExt { + /// Like `.lock().unwrap()`, but recovers from a poisoned mutex instead + /// of panicking again — logs once per recovery so it's visible in the + /// daemon's own output, not just silently swallowed. + fn lock_recover(&self) -> MutexGuard<'_, T>; +} + +impl MutexExt for Mutex { + fn lock_recover(&self) -> MutexGuard<'_, T> { + match self.lock() { + Ok(guard) => guard, + Err(poisoned) => { + eprintln!( + "breadmill: WARNING: a mutex was poisoned by a panic on another thread; \ + recovering it and continuing instead of cascading the panic" + ); + poisoned.into_inner() + } + } + } +} diff --git a/breadsearch/Cargo.toml b/breadsearch/Cargo.toml index 5569deb..ca78b7a 100644 --- a/breadsearch/Cargo.toml +++ b/breadsearch/Cargo.toml @@ -10,7 +10,7 @@ path = "src/main.rs" [dependencies] breadsearch-shared = { path = "../breadsearch-shared" } -bread-theme = { git = "https://github.com/Breadway/bread-ecosystem", tag = "v0.2.8", features = ["gtk"] } +bread-theme = { git = "https://github.com/Breadway/bread-ecosystem", tag = "v0.2.10", features = ["gtk"] } gtk4 = { version = "0.11", features = ["v4_12"] } gtk4-layer-shell = "0.8" serde_json = "1" diff --git a/breadsearch/src/main.rs b/breadsearch/src/main.rs index 58f4b47..a9812fe 100644 --- a/breadsearch/src/main.rs +++ b/breadsearch/src/main.rs @@ -298,12 +298,19 @@ fn run_ui() { let _ = tx.send(breadsearch_shared::send_request(&req)); }); - // Poll via idle_add_local until the thread delivers its result. - // Unix socket round-trips are sub-millisecond so this fires once. + // Wait for the thread's result on a bounded timer rather than + // an `idle_add_local` — an idle source has no wait condition + // of its own, so GLib re-invokes it on every single main-loop + // iteration, i.e. a busy-spin pinning a full CPU core for as + // long as the daemon takes to answer (normally sub-ms, but + // the daemon's socket has no read timeout of its own, so a + // wedged/slow daemon previously meant an indefinite spin). + // 15ms is imperceptible added latency for a search box and + // caps this at a couple dozen checks per second instead. let rx = Rc::new(rx); let list_t = list_clone.clone(); - glib::idle_add_local(move || { + glib::timeout_add_local(std::time::Duration::from_millis(15), move || { match rx.try_recv() { Ok(result) => { populate_list(&list_t, result);