breadmill/breadsearch: narrow embedder lock scope, recover from poisoned mutexes, atomic index save + corrupt rebuild, fix GUI busy-poll, bump theme pin
- indexer.rs: embedder mutex is now locked only around each chunk's embed_document() call instead of the whole file's chunk loop, so a slow MIGraphX JIT compile on one file no longer blocks every query for minutes - sync_ext.rs (new)/indexer.rs/main.rs/serve.rs: added MutexExt::lock_recover(), replacing every .lock().unwrap() so a panic on one thread (poisoning the mutex) logs a warning and recovers instead of cascading into every future lock().unwrap() call - store.rs: save_index now writes to a .tmp sibling and renames atomically into place; Store::open recovers from an index that fails to load by wiping it and the SQLite files/chunks tables so the next scan rebuilds from scratch, instead of refusing to start at all. Note: this recovery only catches corruption usearch's loader reports as Err — verified experimentally that sufficiently garbled input segfaults the process instead, which no Rust-side handling can catch; the atomic save is what actually prevents the realistic mid-crash corruption case from arising - breadsearch/src/main.rs: GUI query-result wait switched from glib::idle_add_local (re-invoked every main-loop tick, pegging a core for the whole wait) to a 15ms glib::timeout_add_local poll - breadsearch/Cargo.toml: bread-theme pin bumped v0.2.8 -> v0.2.10 to match the rest of the family
This commit is contained in:
parent
3d83bd747e
commit
d01a3841d9
8 changed files with 164 additions and 30 deletions
2
Cargo.lock
generated
2
Cargo.lock
generated
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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<Store>,
|
||||
|
|
@ -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<String, (i64, String)> = {
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<SharedState>, 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 },
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
}
|
||||
|
|
|
|||
41
breadmill/src/sync_ext.rs
Normal file
41
breadmill/src/sync_ext.rs
Normal file
|
|
@ -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<T> {
|
||||
/// 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<T> MutexExt<T> for Mutex<T> {
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue