From 657de3de86a0a1acff74127976b46da979ee5097 Mon Sep 17 00:00:00 2001 From: karl Date: Tue, 13 Oct 2026 02:30:38 +0200 Subject: [PATCH] Watch selected folders and reconcile bounded changes with media freshness guards --- AGENTS.md | 6 + Cargo.lock | 1 + apps/desktop/src-tauri/Cargo.lock | 79 +++ apps/desktop/src-tauri/Cargo.toml | 1 + apps/desktop/src-tauri/src/data_directory.rs | 57 +- apps/desktop/src-tauri/src/main.rs | 105 ++- .../src-tauri/src/semantic_commands.rs | 10 +- apps/desktop/src-tauri/src/watching.rs | 644 ++++++++++++++++++ apps/desktop/src/main.ts | 55 ++ crates/searchi-core/Cargo.toml | 1 + crates/searchi-core/src/lib.rs | 484 +++++++++---- crates/searchi-core/tests/index_options.rs | 1 + .../tests/targeted_reconciliation.rs | 484 +++++++++++++ crates/searchi-media/src/lib.rs | 58 +- crates/searchi-service/src/lib.rs | 2 +- crates/searchi-service/src/media.rs | 15 +- crates/searchi-service/src/store.rs | 6 +- .../searchi-service/src/store/incremental.rs | 15 +- .../tests/incremental_google_cpu.rs | 50 +- docs/AUTOMATIC-INDEXING.md | 89 +++ docs/INCREMENTAL-SEMANTIC.md | 2 +- docs/ROADMAP.md | 6 + 22 files changed, 2022 insertions(+), 149 deletions(-) create mode 100644 apps/desktop/src-tauri/src/watching.rs create mode 100644 crates/searchi-core/tests/targeted_reconciliation.rs create mode 100644 docs/AUTOMATIC-INDEXING.md diff --git a/AGENTS.md b/AGENTS.md index 5064572..bc0d7d4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -51,6 +51,12 @@ readiness without completed release gates and recorded evidence. thread; yield between document encodes. Trace admission separately from native inference. This does not bound decoder, extraction or ANN CPU work. Do not claim preemption of an already executing media encode on a one-thread budget. +- Treat native filesystem events as bounded hints; retain work across busy scans + and verify periodically. Use targeted, cancellable reconciliation for events. + Media hints must advance durable change revisions even with identical stats; + old media chunks stay ineligible until byte-verified indexing restores them. + Capture probe revisions before work and guard metadata writes. Exclude canonical + data/model/runtime paths and keep decoded frames under owned application data. - UI should belong to each OS: native typography/chrome/dialogs, themes, keyboard access, reduced motion and high contrast. Brief transitions must not delay requests. Preserve selection/focus as progressive results arrive. diff --git a/Cargo.lock b/Cargo.lock index bd504df..f098e33 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2267,6 +2267,7 @@ dependencies = [ "lopdf", "quick-xml", "rusqlite", + "searchi-media", "serde", "serde_json", "sha2 0.10.9", diff --git a/apps/desktop/src-tauri/Cargo.lock b/apps/desktop/src-tauri/Cargo.lock index 8c3fd6f..e422654 100644 --- a/apps/desktop/src-tauri/Cargo.lock +++ b/apps/desktop/src-tauri/Cargo.lock @@ -1345,6 +1345,15 @@ dependencies = [ "winapi", ] +[[package]] +name = "fsevent-sys" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2" +dependencies = [ + "libc", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -2133,6 +2142,26 @@ dependencies = [ "cfb", ] +[[package]] +name = "inotify" +version = "0.11.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cc00ea907cab49550b7da656f80ebb97be1b997d931fbcd28d39734e17ce592" +dependencies = [ + "bitflags 2.13.2", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d" +dependencies = [ + "libc", +] + [[package]] name = "inout" version = "0.2.2" @@ -2355,6 +2384,26 @@ dependencies = [ "serde", ] +[[package]] +name = "kqueue" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d763e5b24120b4ddf50de6c92308156765aabfbbccebf401da7cff2d70a41ea" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087" +dependencies = [ + "bitflags 2.13.2", + "libc", +] + [[package]] name = "libappindicator" version = "0.9.0" @@ -2562,6 +2611,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" dependencies = [ "libc", + "log", "wasi", "windows-sys 0.61.2", ] @@ -2648,6 +2698,33 @@ dependencies = [ "memchr", ] +[[package]] +name = "notify" +version = "8.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3" +dependencies = [ + "bitflags 2.13.2", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio", + "notify-types", + "walkdir", + "windows-sys 0.60.2", +] + +[[package]] +name = "notify-types" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a" +dependencies = [ + "bitflags 2.13.2", +] + [[package]] name = "num-conv" version = "0.2.2" @@ -3718,6 +3795,7 @@ dependencies = [ "lopdf", "quick-xml", "rusqlite", + "searchi-media", "serde", "sha2 0.10.9", "walkdir", @@ -3730,6 +3808,7 @@ version = "0.1.0" dependencies = [ "anyhow", "image", + "notify", "searchi-ai", "searchi-assets", "searchi-core", diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index cb0fdfc..3496321 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -15,6 +15,7 @@ tauri-build = { version = "2", features = [] } [dependencies] anyhow = "1" +notify = "=8.2.0" tauri = { version = "2", features = ["protocol-asset"] } tauri-plugin-dialog = "2" tauri-plugin-opener = "2" diff --git a/apps/desktop/src-tauri/src/data_directory.rs b/apps/desktop/src-tauri/src/data_directory.rs index 916ea1c..98a88c3 100644 --- a/apps/desktop/src-tauri/src/data_directory.rs +++ b/apps/desktop/src-tauri/src/data_directory.rs @@ -1,6 +1,44 @@ //! Explicit data-directory isolation for portable use and native release QA. //! Ordinary launches continue to use the operating system's app-data directory. -use std::{ffi::OsString, io, path::PathBuf}; +use std::{ + ffi::OsString, + io, + path::{Path, PathBuf}, +}; + +/// Match canonical watched paths, including Windows verbatim prefixes, even +/// before a first-use model or scratch directory exists. +pub fn canonical_exclusion(path: &Path) -> io::Result { + if !path.is_absolute() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "Excluded directory must be absolute", + )); + } + let mut ancestor = path.to_path_buf(); + let mut suffix = Vec::new(); + loop { + match std::fs::canonicalize(&ancestor) { + Ok(mut resolved) => { + for part in suffix.iter().rev() { + resolved.push(part); + } + return Ok(resolved); + } + Err(error) if error.kind() == io::ErrorKind::NotFound => { + let name = ancestor.file_name().ok_or(error)?; + suffix.push(name.to_os_string()); + if !ancestor.pop() { + return Err(io::Error::new( + io::ErrorKind::NotFound, + "Excluded directory has no existing ancestor", + )); + } + } + Err(error) => return Err(error), + } + } +} pub fn resolve(default: PathBuf, override_path: Option) -> io::Result { let Some(value) = override_path else { @@ -20,6 +58,23 @@ pub fn resolve(default: PathBuf, override_path: Option) -> io::Result< mod tests { use super::*; + #[test] + fn missing_exclusion_matches_canonical_root_without_creating_directories() { + let root = tempfile::tempdir().unwrap(); + let missing = root.path().join("Models").join("MediaScratch"); + let canonical = std::fs::canonicalize(root.path()).unwrap(); + let exclusion = canonical_exclusion(&missing).unwrap(); + assert_eq!(exclusion, canonical.join("Models/MediaScratch")); + assert!(exclusion.starts_with(&canonical)); + assert!(!missing.exists()); + assert_eq!(canonical_exclusion(root.path()).unwrap(), canonical); + } + + #[test] + fn relative_exclusion_is_rejected() { + assert!(canonical_exclusion(Path::new("relative/cache")).is_err()); + } + #[test] fn explicit_directory_isolates_settings_and_index_without_changing_default() { let default = std::env::temp_dir().join("searchi-default"); diff --git a/apps/desktop/src-tauri/src/main.rs b/apps/desktop/src-tauri/src/main.rs index 98defa8..37bd4b7 100644 --- a/apps/desktop/src-tauri/src/main.rs +++ b/apps/desktop/src-tauri/src/main.rs @@ -8,6 +8,7 @@ mod data_directory; mod model_download; mod semantic_commands; mod settings; +mod watching; use ai_commands::{generate_ai, prepare_ai_source, AiSession}; use searchi_core::metrics::{MetricsMode, SearchMetrics, SearchResponse, TraceCollector}; @@ -38,6 +39,8 @@ struct AppState { settings: Arc>, shortcut_status: String, semantic: Arc, + watching: Arc, + index_options: searchi_core::IndexOptions, } #[derive(Serialize)] @@ -85,6 +88,7 @@ struct DesktopStatus { settings: Settings, indexing: bool, shortcut_status: String, + watching: Vec, } #[tauri::command] @@ -98,6 +102,7 @@ fn desktop_status(state: State<'_, AppState>) -> Result { .clone(), indexing: state.indexing.load(Ordering::Acquire), shortcut_status: state.shortcut_status.clone(), + watching: state.watching.status(), }) } @@ -362,15 +367,18 @@ async fn index_folder(root: String, state: State<'_, AppState>) -> Result) -> Result>, + indexing: Arc, + roots: Vec, + options: searchi_core::IndexOptions, +) -> Result { + watching::WatchService::start(roots, options.excluded_paths.clone(), move |request| { + if indexing + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return watching::ScanOutcome::Busy; + } + let guard = IndexGuard(indexing.clone()); + let _ = app.emit("folder-indexing-changed", ()); + let result = (|| -> Result { + let mut index = SearchIndex::open(&database).map_err(|error| error.to_string())?; + let report = match request.paths { + Some(paths) => index.index_paths_cancellable( + &request.root, + &paths, + &options, + &request.cancelled, + ), + None if request.media_hint => index.index_paths_cancellable( + &request.root, + std::slice::from_ref(&request.root), + &options, + &request.cancelled, + ), + None => index.index_folder_cancellable(&request.root, &options, &request.cancelled), + } + .map_err(|error| error.to_string())?; + let changed = report.indexed > 0 || report.removed > 0; + if changed && !request.cancelled.load(Ordering::Acquire) { + index.prewarm().map_err(|error| error.to_string())?; + *lexical + .lock() + .map_err(|_| "Keyword engine is unavailable.")? = index; + } + Ok(changed) + })(); + drop(guard); + let _ = app.emit("folder-indexing-changed", ()); + match result { + Ok(changed) => { + if changed { + let _ = app.emit("search-index-changed", ()); + } + watching::ScanOutcome::Completed + } + Err(error) => watching::ScanOutcome::Failed(error), + } + }) +} + fn main() { tauri::Builder::default() .plugin(tauri_plugin_dialog::init()) @@ -455,7 +523,25 @@ fn main() { "The media decoder is missing from this installation. Repair or reinstall searchi.".into() }) })(); - let semantic = Arc::new(SemanticState::new(model_directory, worker, media_tools)); + let mut ignored = vec![directory.clone(), model_directory.clone()]; + if let Some(parent) = worker.parent().filter(|path| path.is_absolute()) { + ignored.push(parent.to_path_buf()); + } + let index_options = searchi_core::IndexOptions { + excluded_paths: ignored.iter() + .map(|path| data_directory::canonical_exclusion(path)) + .collect::>>()?, + ..Default::default() + }; + let indexing = Arc::new(AtomicBool::new(false)); + let watching = Arc::new(start_folder_watching( + app.handle().clone(), database.clone(), lexical.clone(), indexing.clone(), + settings.value.roots.clone(), index_options.clone(), + ).map_err(std::io::Error::other)?); + let semantic = Arc::new( + SemanticState::new(model_directory, worker, media_tools) + .with_media_scratch(directory.join("MediaScratch")), + ); semantic.begin_automatic_setup(database.clone(), settings.value.cpu_threads); #[cfg(target_os = "macos")] let shortcut = "Shift+Super+Space"; @@ -475,10 +561,12 @@ fn main() { ai: Arc::new(AiSession::default()), database, lexical, - indexing: Arc::new(AtomicBool::new(false)), + indexing, settings: Arc::new(Mutex::new(settings)), shortcut_status, semantic, + watching, + index_options, }); Ok(()) }) @@ -500,8 +588,15 @@ fn main() { hybrid_search, hybrid_search_measured ]) - .run(tauri::generate_context!()) - .expect("Searchi desktop could not start"); + .build(tauri::generate_context!()) + .expect("Searchi desktop could not start") + .run(|app, event| { + if matches!(event, tauri::RunEvent::Exit) { + // Release the worker-held AppHandle and owned watcher before + // application state is torn down. + app.state::().watching.stop(); + } + }); } #[cfg(test)] diff --git a/apps/desktop/src-tauri/src/semantic_commands.rs b/apps/desktop/src-tauri/src/semantic_commands.rs index 96026bc..21dec5e 100644 --- a/apps/desktop/src-tauri/src/semantic_commands.rs +++ b/apps/desktop/src-tauri/src/semantic_commands.rs @@ -29,6 +29,7 @@ pub struct SemanticState { inference_budget: InferenceBudget, lifecycle: Mutex<()>, media_tools: Option, + media_limits: searchi_media::MediaLimits, media_setup_error: Option, } @@ -58,10 +59,17 @@ impl SemanticState { inference_budget: InferenceBudget::new(2), lifecycle: Mutex::new(()), media_tools, + media_limits: searchi_media::MediaLimits::default(), media_setup_error, } } + /// Keep generated video frames below the application's excluded data tree. + pub fn with_media_scratch(mut self, directory: PathBuf) -> Self { + self.media_limits.scratch_directory = Some(directory); + self + } + fn ensure_idle(&self) -> Result<(), String> { if self.embedding.load(Ordering::Acquire) || self.downloading.load(Ordering::Acquire) @@ -335,7 +343,7 @@ pub async fn index_embeddings(state: State<'_, AppState>) -> Result>, + pub cancelled: Arc, + pub media_hint: bool, +} +pub enum ScanOutcome { + Completed, + Busy, + Failed(String), +} +#[derive(Clone, Serialize)] +pub struct WatchStatus { + pub root: PathBuf, + pub attached: bool, + pub pending: bool, + pub error: Option, + pub watch_error: Option, +} +#[derive(Clone, Copy)] +struct Timing { + debounce: Duration, + max_delay: Duration, + periodic: Duration, + retry: Duration, +} +impl Default for Timing { + fn default() -> Self { + Self { + debounce: Duration::from_millis(750), + max_delay: Duration::from_secs(5), + periodic: Duration::from_secs(300), + retry: Duration::from_secs(5), + } + } +} +struct Root { + paths: BTreeSet, + full: bool, + media_hint: bool, + ever_attached: bool, + first: Option, + last: Instant, + next_scan: Instant, + retry_at: Instant, + attach_retry: Instant, + failures: u32, + attached: bool, + error: Option, + watch_error: Option, +} +impl Root { + fn new(now: Instant) -> Self { + Self { + paths: BTreeSet::new(), + full: true, + media_hint: false, + ever_attached: false, + first: Some(now), + last: now, + next_scan: now, + retry_at: now, + attach_retry: now, + failures: 0, + attached: false, + error: None, + watch_error: None, + } + } + fn dirty(&mut self, paths: Option<&[PathBuf]>, now: Instant, media_hint: bool) { + self.media_hint |= media_hint; + self.first.get_or_insert(now); + self.last = now; + match paths { + Some(paths) if !self.full => { + self.paths.extend(paths.iter().cloned()); + if self.paths.len() > MAX_PATHS { + self.paths.clear(); + self.full = true; + } + } + None => { + self.paths.clear(); + self.full = true; + } + _ => {} + } + } +} +struct State { + roots: BTreeMap, + stop: bool, +} +struct Shared { + state: Mutex, + changed: Condvar, + ignored: Vec, + timing: Timing, + cancelled: Arc, +} +/// Own this service for the application lifetime. Dropping joins its worker after +/// the currently executing reconciliation callback finishes. +pub struct WatchService { + shared: Arc, + worker: Mutex>>, +} +impl WatchService { + pub fn start( + roots: Vec, + ignored: Vec, + callback: impl Fn(ScanRequest) -> ScanOutcome + Send + 'static, + ) -> Result { + Self::start_timed(roots, ignored, callback, Timing::default()) + } + fn start_timed( + roots: Vec, + ignored: Vec, + callback: impl Fn(ScanRequest) -> ScanOutcome + Send + 'static, + timing: Timing, + ) -> Result { + validate(&roots)?; + if ignored.len() > 128 || ignored.iter().any(|p| !p.is_absolute()) { + return Err("Watcher exclusions must be bounded absolute paths.".into()); + } + let shared = Arc::new(Shared { + state: Mutex::new(State { + roots: BTreeMap::new(), + stop: false, + }), + changed: Condvar::new(), + ignored, + timing, + cancelled: Arc::new(AtomicBool::new(false)), + }); + let target = shared.clone(); + let worker = thread::Builder::new() + .name("searchi-reconcile".into()) + .spawn(move || run(target, callback)) + .map_err(|e| e.to_string())?; + let service = Self { + shared, + worker: Mutex::new(Some(worker)), + }; + service.replace_roots(roots)?; + Ok(service) + } + /// Stop and join only this owned worker. Call on application exit to break + /// callback ownership of the desktop emitter. Cancellation waits for the + /// current bounded parser operation to reach its next cooperative check. + pub fn stop(&self) { + self.shared.cancelled.store(true, Ordering::Release); + if let Ok(mut state) = self.shared.state.lock() { + state.stop = true; + self.shared.changed.notify_all(); + } + let worker = self.worker.lock().unwrap_or_else(|e| e.into_inner()).take(); + if let Some(worker) = worker { + if worker.thread().id() != thread::current().id() { + let _ = worker.join(); + } + } + } + + pub fn add_root(&self, root: PathBuf) -> Result<(), String> { + let mut s = self + .shared + .state + .lock() + .map_err(|_| "Watcher is unavailable.")?; + if s.roots.contains_key(&root) { + return Ok(()); + } + if s.roots.len() >= MAX_ROOTS || !root.is_absolute() { + return Err("Watcher folders must be absolute and limited to 64.".into()); + } + s.roots.insert(root, Root::new(Instant::now())); + self.shared.changed.notify_all(); + Ok(()) + } + pub fn replace_roots(&self, roots: Vec) -> Result<(), String> { + validate(&roots)?; + let mut s = self + .shared + .state + .lock() + .map_err(|_| "Watcher is unavailable.")?; + s.roots.retain(|p, _| roots.contains(p)); + for p in roots { + s.roots + .entry(p) + .or_insert_with(|| Root::new(Instant::now())); + } + self.shared.changed.notify_all(); + Ok(()) + } + pub fn status(&self) -> Vec { + self.shared + .state + .lock() + .map(|s| { + s.roots + .iter() + .map(|(p, r)| WatchStatus { + root: p.clone(), + attached: r.attached, + pending: r.first.is_some(), + error: r.error.clone().or_else(|| r.watch_error.clone()), + watch_error: r.watch_error.clone(), + }) + .collect() + }) + .unwrap_or_default() + } +} +impl Drop for WatchService { + fn drop(&mut self) { + self.stop(); + } +} +fn validate(roots: &[PathBuf]) -> Result<(), String> { + if roots.len() > MAX_ROOTS || roots.iter().any(|p| !p.is_absolute()) { + return Err("Watcher folders must be absolute and limited to 64.".into()); + } + Ok(()) +} +fn excluded(path: &Path, shared: &Shared) -> bool { + shared.ignored.iter().any(|p| path.starts_with(p)) + || path.components().any(|c| { + matches!( + c.as_os_str().to_str(), + Some(".git" | "node_modules" | "target") + ) + }) +} +fn hint(shared: &Shared, event: notify::Result) { + let mut s = match shared.state.lock() { + Ok(s) => s, + Err(_) => return, + }; + let now = Instant::now(); + match event { + Ok(e) + if !e.need_rescan() + && matches!( + e.kind, + EventKind::Access(_) + | EventKind::Modify(notify::event::ModifyKind::Metadata( + notify::event::MetadataKind::AccessTime + )) + ) => {} + Ok(e) if !e.need_rescan() && !e.paths.is_empty() => { + // Events can contain unbounded path vectors; consume a bounded prefix + // and invalidate all roots if its representation overflowed. + if e.paths.len() > MAX_PATHS { + for r in s.roots.values_mut() { + r.dirty(None, now, true); + } + } else { + for (root, r) in &mut s.roots { + let paths: Vec<_> = e + .paths + .iter() + .filter(|p| p.starts_with(root.as_path()) && !excluded(p, shared)) + .cloned() + .collect(); + if !paths.is_empty() { + r.dirty(Some(&paths), now, true); + } + } + } + } + _ => { + for r in s.roots.values_mut() { + r.dirty(None, now, true); + } + } + } + shared.changed.notify_all(); +} +fn run(shared: Arc, callback: impl Fn(ScanRequest) -> ScanOutcome) { + let events = shared.clone(); + let mut watcher = notify::recommended_watcher(move |event| hint(&events, event)).ok(); + let mut watched = BTreeSet::new(); + loop { + let roots = { + let s = shared.state.lock().unwrap_or_else(|e| e.into_inner()); + if s.stop { + break; + } + s.roots.keys().cloned().collect::>() + }; + for removed in watched + .iter() + .filter(|p| !roots.contains(p)) + .cloned() + .collect::>() + { + if let Some(w) = &mut watcher { + let _ = w.unwatch(&removed); + } + watched.remove(&removed); + } + let now = Instant::now(); + for root in &roots { + if watched.contains(root) && root.is_dir() { + continue; + } + if watched.remove(root) { + if let Some(r) = shared + .state + .lock() + .unwrap_or_else(|e| e.into_inner()) + .roots + .get_mut(root) + { + r.attached = false; + } + if let Some(w) = &mut watcher { + let _ = w.unwatch(root); + } + } + let due = shared + .state + .lock() + .unwrap_or_else(|e| e.into_inner()) + .roots + .get(root) + .is_some_and(|r| now >= r.attach_retry); + if !due { + continue; + } + if watcher.is_none() { + let events = shared.clone(); + watcher = notify::recommended_watcher(move |e| hint(&events, e)).ok(); + } + let result = match &mut watcher { + Some(w) => w + .watch(root, RecursiveMode::Recursive) + .map_err(|e| e.to_string()), + None => Err( + "Native watcher is unavailable; periodic reconciliation remains enabled." + .into(), + ), + }; + let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(r) = s.roots.get_mut(root) { + r.attached = result.is_ok(); + r.attach_retry = if result.is_ok() { + now + } else { + now + shared.timing.retry + }; + if result.is_ok() { + r.watch_error = None; + r.dirty(None, now, r.ever_attached); + r.ever_attached = true; + watched.insert(root.clone()); + } else { + r.watch_error = result.err().map(|e| e.chars().take(512).collect()); + } + } + } + let request = { + let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner()); + if s.stop { + break; + } + let now = Instant::now(); + for r in s.roots.values_mut() { + if now >= r.next_scan { + r.dirty(None, now, false); + r.next_scan = now + shared.timing.periodic; + } + } + s.roots.iter_mut().find_map(|(root, r)| { + let due = r.first.is_some_and(|first| { + now >= r.retry_at + && (now.duration_since(r.last) >= shared.timing.debounce + || now.duration_since(first) >= shared.timing.max_delay) + }); + if !due { + return None; + } + let paths = if r.full { + None + } else { + Some(std::mem::take(&mut r.paths).into_iter().collect()) + }; + r.full = false; + let media_hint = std::mem::take(&mut r.media_hint); + r.first = None; + Some(ScanRequest { + root: root.clone(), + paths, + cancelled: shared.cancelled.clone(), + media_hint, + }) + }) + }; + if let Some(request) = request { + let root = request.root.clone(); + let paths = request.paths.clone(); + let media_hint = request.media_hint; + let result = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| callback(request))) + .unwrap_or_else(|_| { + ScanOutcome::Failed( + "Folder reconciliation failed unexpectedly; retrying.".into(), + ) + }); + let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(r) = s.roots.get_mut(&root) { + let now = Instant::now(); + match result { + ScanOutcome::Completed => { + r.failures = 0; + r.error = None; + r.retry_at = now; + } + ScanOutcome::Busy => { + r.dirty(paths.as_deref(), now, media_hint); + r.retry_at = now + Duration::from_millis(250); + } + ScanOutcome::Failed(e) => { + r.dirty(paths.as_deref(), now, media_hint); + r.failures = r.failures.saturating_add(1); + r.error = Some(e.chars().take(512).collect()); + r.retry_at = now + + shared + .timing + .retry + .saturating_mul(1u32 << r.failures.saturating_sub(1).min(4)) + .min(Duration::from_secs(60)); + } + } + } + } else { + let s = shared.state.lock().unwrap_or_else(|e| e.into_inner()); + if s.stop { + break; + } + let _ = shared.changed.wait_timeout(s, Duration::from_millis(100)); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + fn fast() -> Timing { + Timing { + debounce: Duration::from_millis(40), + max_delay: Duration::from_millis(120), + periodic: Duration::from_secs(30), + retry: Duration::from_millis(20), + } + } + fn receive(rx: &mpsc::Receiver) -> ScanRequest { + rx.recv_timeout(Duration::from_secs(10)) + .expect("watch reconciliation") + } + #[test] + fn native_windows_or_linux_create_update_rename_delete() { + let root = tempfile::tempdir().unwrap(); + let (tx, rx) = mpsc::channel(); + let service = WatchService::start_timed( + vec![root.path().to_owned()], + vec![], + move |r| { + tx.send(r).unwrap(); + ScanOutcome::Completed + }, + fast(), + ) + .unwrap(); + assert!(receive(&rx).paths.is_none()); + let first = root.path().join("first.txt"); + let second = root.path().join("second.txt"); + std::fs::write(&first, "one").unwrap(); + assert!(receive(&rx).paths.unwrap().contains(&first)); + std::fs::write(&first, "two").unwrap(); + assert!(receive(&rx).paths.unwrap().contains(&first)); + std::fs::rename(&first, &second).unwrap(); + let rename = receive(&rx).paths.unwrap(); + assert!(rename.contains(&first) || rename.contains(&second)); + std::fs::remove_file(&second).unwrap(); + assert!(receive(&rx).paths.unwrap().contains(&second)); + assert!(service.status()[0].attached); + } + #[test] + fn overflow_rescan_and_events_during_scan_are_retained() { + let root = tempfile::tempdir().unwrap(); + let (tx, rx) = mpsc::channel(); + let service = WatchService::start_timed( + vec![root.path().to_owned()], + vec![], + move |r| { + tx.send(r).unwrap(); + ScanOutcome::Completed + }, + fast(), + ) + .unwrap(); + let _ = receive(&rx); + let paths = (0..129) + .map(|i| root.path().join(format!("{i}.txt"))) + .collect(); + hint( + &service.shared, + Ok(Event::new(EventKind::Any).add_path(root.path().join("a"))), + ); + let mut e = Event::new(EventKind::Any); + e.paths = paths; + hint(&service.shared, Ok(e)); + let overflow = receive(&rx); + assert!(overflow.paths.is_none()); + assert!(overflow.media_hint); + } + #[test] + fn events_arriving_during_reconciliation_survive_completion() { + let root = tempfile::tempdir().unwrap(); + let (tx, rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let release_rx = Mutex::new(release_rx); + let service = WatchService::start_timed( + vec![root.path().to_owned()], + vec![], + move |r| { + tx.send(r).unwrap(); + release_rx.lock().unwrap().recv().unwrap(); + ScanOutcome::Completed + }, + fast(), + ) + .unwrap(); + let _ = receive(&rx); + let path = root.path().join("later.txt"); + hint( + &service.shared, + Ok(Event::new(EventKind::Any).add_path(path.clone())), + ); + release_tx.send(()).unwrap(); + assert!(receive(&rx).paths.unwrap().contains(&path)); + release_tx.send(()).unwrap(); + drop(service); + } + + #[test] + fn callback_panic_retries_and_recovers() { + let root = tempfile::tempdir().unwrap(); + let (tx, rx) = mpsc::channel(); + let attempts = AtomicBool::new(false); + let service = WatchService::start_timed( + vec![root.path().to_owned()], + vec![], + move |r| { + if !attempts.swap(true, Ordering::AcqRel) { + panic!("owned callback failure"); + } + tx.send(r).unwrap(); + ScanOutcome::Completed + }, + fast(), + ) + .unwrap(); + assert!(receive(&rx).paths.is_none()); + service.stop(); + assert!(!service.status()[0].pending); + } + + #[test] + fn busy_retry_preserves_native_media_hint() { + let root = tempfile::tempdir().unwrap(); + let (tx, rx) = mpsc::channel(); + let service = WatchService::start_timed( + vec![root.path().to_owned()], + vec![], + move |r| { + if r.media_hint { + tx.send(r).unwrap(); + ScanOutcome::Busy + } else { + tx.send(r).unwrap(); + ScanOutcome::Completed + } + }, + fast(), + ) + .unwrap(); + assert!(!receive(&rx).media_hint); + let mut event = Event::new(EventKind::Any); + event.attrs.set_flag(notify::event::Flag::Rescan); + hint(&service.shared, Ok(event)); + let first = receive(&rx); + assert!(first.paths.is_none()); + assert!(first.media_hint); + assert!(receive(&rx).media_hint); + service.stop(); + } + + #[test] + fn periodic_reconciliation_and_shutdown_signal() { + let root = tempfile::tempdir().unwrap(); + let (tx, rx) = mpsc::channel(); + let mut timing = fast(); + timing.periodic = Duration::from_millis(150); + let service = WatchService::start_timed( + vec![root.path().to_owned()], + vec![], + move |r| { + tx.send(r).unwrap(); + ScanOutcome::Completed + }, + timing, + ) + .unwrap(); + let initial = receive(&rx); + assert!(initial.paths.is_none()); + assert!(!initial.media_hint); + let periodic = receive(&rx); + assert!(periodic.paths.is_none()); + assert!(!periodic.media_hint); + drop(service); + assert!(initial.cancelled.load(Ordering::Acquire)); + } +} diff --git a/apps/desktop/src/main.ts b/apps/desktop/src/main.ts index 0b328fd..1e81935 100644 --- a/apps/desktop/src/main.ts +++ b/apps/desktop/src/main.ts @@ -77,6 +77,13 @@ interface DesktopStatus { settings: Settings; indexing: boolean; shortcut_status: string; + watching?: { + root: string; + attached: boolean; + pending: boolean; + error: string | null; + watch_error: string | null; + }[]; } interface SemanticStatus { model: string; @@ -452,6 +459,16 @@ async function refreshDesktopStatus() { const path = document.createElement("span"); path.textContent = displayPath(root); path.className = "path"; + const watched = status.watching?.find((entry) => entry.root === root); + if (watched) { + path.title = watched.error + ? `Automatic updates need attention: ${watched.error}` + : watched.pending + ? "Checking for file changes…" + : watched.attached + ? "Watching for file changes" + : "Checking for changes automatically"; + } const reindex = document.createElement("button"); reindex.type = "button"; reindex.textContent = "Reindex"; @@ -1039,6 +1056,31 @@ function submitSearch(request: SearchRequest, lane: SearchLane) { (lane === "lexical" ? lexicalLane : hybridLane).submit(request); } +function refreshIndexedSearch() { + const previous = latestSearch; + if ( + !previous || + previous.id !== requestId || + query.value.trim() !== previous.query + ) + return; + clearTimeout(hybridTimer); + lexicalLane.clear(); + hybridLane.clear(); + querySemanticFailed = false; + shownLane = "lexical"; + // Keep the original input/first-response timing while rejecting responses + // captured before the index changed, even when the query text stays identical. + const refreshed: SearchRequest = { ...previous, id: ++requestId }; + latestSearch = refreshed; + submitSearch(refreshed, "lexical"); + if (canSearchSemantically()) { + hybridTimer = setTimeout(() => { + if (isCurrent(refreshed)) submitSearch(refreshed, "hybrid"); + }, SEMANTIC_SETTLE_MS); + } +} + const isCurrent = (request: SearchRequest) => request.id === requestId; const lexicalLane = new LatestLane( async (request) => { @@ -1366,6 +1408,19 @@ async function initializeSurface() { void refreshDesktopStatus().catch(showError); queueSemanticRefresh(0); }).catch(showError); + if (isNativeDesktop()) { + void listen("folder-indexing-changed", () => { + void refreshDesktopStatus().catch(showError); + queueSemanticRefresh(0); + }).catch(showError); + void listen("search-index-changed", () => { + folderRevision++; + void refreshStats().catch(showError); + void refreshDesktopStatus().catch(showError); + queueSemanticRefresh(0); + refreshIndexedSearch(); + }).catch(showError); + } // Poll only during a user-triggered job; idle indexing status does no work. setInterval(() => { if (indexingActive) void refreshDesktopStatus().catch(showError); diff --git a/crates/searchi-core/Cargo.toml b/crates/searchi-core/Cargo.toml index 2eadb19..6d0360f 100644 --- a/crates/searchi-core/Cargo.toml +++ b/crates/searchi-core/Cargo.toml @@ -5,6 +5,7 @@ edition.workspace = true license.workspace = true [dependencies] +searchi-media = { path = "../searchi-media" } anyhow.workspace = true serde.workspace = true sha2 = "0.10" diff --git a/crates/searchi-core/src/lib.rs b/crates/searchi-core/src/lib.rs index bd7e8bf..1a0e5f5 100644 --- a/crates/searchi-core/src/lib.rs +++ b/crates/searchi-core/src/lib.rs @@ -17,11 +17,12 @@ use std::{ collections::{BTreeSet, HashMap}, fs, path::{Path, PathBuf}, + sync::atomic::{AtomicBool, Ordering}, time::UNIX_EPOCH, }; use walkdir::WalkDir; -const SCHEMA_VERSION: i64 = 12; +const SCHEMA_VERSION: i64 = 13; const MAX_RESULTS: usize = 1000; #[derive(Debug, Default, Serialize)] @@ -38,12 +39,15 @@ pub struct IndexReport { #[derive(Debug, Clone)] pub struct IndexOptions { pub excluded_directory_names: Vec, + /// Absolute component-normalized directory/file paths, matched by boundary. + pub excluded_paths: Vec, } impl Default for IndexOptions { fn default() -> Self { Self { excluded_directory_names: vec![".git".into(), "node_modules".into(), "target".into()], + excluded_paths: Vec::new(), } } } @@ -262,7 +266,17 @@ impl SearchIndex { if !has_extraction_hash { initialization.execute("ALTER TABLE files ADD COLUMN extracted_sha256 TEXT", [])?; } - initialization.execute_batch("PRAGMA user_version=12;")?; + let has_change_revision: bool = initialization.query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info('files') WHERE name='change_revision')", + [], |row| row.get(0), + )?; + if !has_change_revision { + initialization.execute( + "ALTER TABLE files ADD COLUMN change_revision INTEGER NOT NULL DEFAULT 0", + [], + )?; + } + initialization.execute_batch("PRAGMA user_version=13;")?; initialization.commit()?; } Ok(Self { @@ -295,6 +309,34 @@ impl SearchIndex { duration_ms: u32, size: u64, mtime_ns: i64, + ) -> Result<()> { + let revision = self.media_change_revision(path, size, mtime_ns)?; + self.set_media_metadata_at_revision(path, tags, duration_ms, size, mtime_ns, revision) + } + + /// Capture before probing media. Event hints invalidate same-stat inputs + /// independently of their size and timestamp. + pub fn media_change_revision(&self, path: &str, size: u64, mtime_ns: i64) -> Result { + self.connection + .query_row( + "SELECT change_revision FROM files WHERE path=?1 AND size=?2 AND mtime=?3", + params![path, i64::try_from(size)?, mtime_ns], + |row| row.get(0), + ) + .optional()? + .context("Media metadata file changed since indexing") + } + + /// Publish probed tags only if no targeted media hint arrived since probe + /// admission. The check and tag update share one write transaction. + pub fn set_media_metadata_at_revision( + &mut self, + path: &str, + tags: &str, + duration_ms: u32, + size: u64, + mtime_ns: i64, + expected_revision: i64, ) -> Result<()> { if tags.len() > 32768 { bail!("Media tags exceed 32 KiB bound"); @@ -302,8 +344,8 @@ impl SearchIndex { let tx = self.connection.transaction()?; let id: Option = tx .query_row( - "SELECT id FROM files WHERE path=?1 AND size=?2 AND mtime=?3", - params![path, i64::try_from(size)?, mtime_ns], + "SELECT id FROM files WHERE path=?1 AND size=?2 AND mtime=?3 AND change_revision=?4", + params![path, i64::try_from(size)?, mtime_ns, expected_revision], |r| r.get(0), ) .optional()?; @@ -411,6 +453,110 @@ impl SearchIndex { if !root.is_dir() { bail!("index root must be a directory"); } + self.reconcile_scopes(&root, std::slice::from_ref(&root), options, None, false) + } + + /// Atomically reconcile at most 128 named files or directory subtrees. + /// Paths must already be absolute and component-normalized. Missing paths + /// remove only their matching root-owned records; an unavailable root fails + /// without touching the index. Symlink events/ancestors are rejected. + pub fn index_paths_with_options( + &mut self, + root: impl AsRef, + paths: &[PathBuf], + options: &IndexOptions, + ) -> Result { + self.index_paths_inner(root.as_ref(), paths, options, None) + } + + /// Cancelled work rolls back the complete reconciliation transaction. + pub fn index_folder_cancellable( + &mut self, + root: impl AsRef, + options: &IndexOptions, + cancelled: &AtomicBool, + ) -> Result { + let root = fs::canonicalize(root).context("resolve index root")?; + if !root.is_dir() { + bail!("index root must be a directory"); + } + self.reconcile_scopes( + &root, + std::slice::from_ref(&root), + options, + Some(cancelled), + false, + ) + } + + pub fn index_paths_cancellable( + &mut self, + root: impl AsRef, + paths: &[PathBuf], + options: &IndexOptions, + cancelled: &AtomicBool, + ) -> Result { + self.index_paths_inner(root.as_ref(), paths, options, Some(cancelled)) + } + + fn index_paths_inner( + &mut self, + root: &Path, + paths: &[PathBuf], + options: &IndexOptions, + cancelled: Option<&AtomicBool>, + ) -> Result { + check_index_cancelled(cancelled)?; + if paths.len() > 128 { + bail!("Targeted reconciliation exceeds 128-path batch limit"); + } + let root = fs::canonicalize(root).context("resolve index root")?; + if !root.is_dir() { + bail!("index root must be a directory"); + } + let mut scopes = Vec::new(); + for path in paths { + validate_normalized_absolute(path)?; + let relative = path + .strip_prefix(&root) + .context("Targeted path lies outside index root")?; + let mut ancestor = root.clone(); + for component in relative.components() { + ancestor.push(component.as_os_str()); + match fs::symlink_metadata(&ancestor) { + Ok(metadata) if metadata.file_type().is_symlink() => { + bail!("Targeted path crosses a symlink") + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => break, + Err(error) => return Err(error).context("Inspect targeted path ancestor"), + } + } + scopes.push(path.clone()); + } + scopes.sort(); + scopes.dedup(); + let mut reduced: Vec = Vec::new(); + for scope in scopes { + if !reduced.iter().any(|parent| scope.starts_with(parent)) { + reduced.push(scope); + } + } + self.reconcile_scopes(&root, &reduced, options, cancelled, true) + } + + fn reconcile_scopes( + &mut self, + root: &Path, + scopes: &[PathBuf], + options: &IndexOptions, + cancelled: Option<&AtomicBool>, + media_hints: bool, + ) -> Result { + check_index_cancelled(cancelled)?; + for excluded in &options.excluded_paths { + validate_normalized_absolute(excluded)?; + } let root_key = root.to_string_lossy().into_owned(); let transaction = self.connection.transaction()?; transaction.execute_batch( @@ -418,139 +564,205 @@ impl SearchIndex { DELETE FROM seen_paths;", )?; let mut report = IndexReport::default(); - for entry in WalkDir::new(&root) - .follow_links(false) - .into_iter() - .filter_entry(|entry| { - entry.depth() == 0 - || !entry.file_type().is_dir() - || !options.excluded_directory_names.iter().any(|name| { - entry - .file_name() - .to_string_lossy() - .eq_ignore_ascii_case(name) - }) - }) - { - let entry = entry.context("directory traversal failed; reconciliation rolled back")?; - if !entry.file_type().is_file() { + for scope in scopes { + check_index_cancelled(cancelled)?; + match fs::symlink_metadata(scope) { + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(error).context("Inspect reconciliation scope"), + Ok(metadata) if metadata.file_type().is_symlink() => { + bail!("Reconciliation scope became a symlink") + } + Ok(_) => {} + } + let relative = scope.strip_prefix(root)?; + let relative = if scope.is_file() { + relative.parent().unwrap_or(Path::new("")) + } else { + relative + }; + let excluded = relative.components().any(|component| { + options.excluded_directory_names.iter().any(|name| { + component + .as_os_str() + .to_string_lossy() + .eq_ignore_ascii_case(name) + }) + }); + if excluded + || options + .excluded_paths + .iter() + .any(|path| scope.starts_with(path)) + { continue; } - if let Some(database) = &self.database_path { - let path_key = entry.path().to_string_lossy(); - let db_key = database.to_string_lossy(); - if path_key == db_key - || path_key == format!("{db_key}-wal") - || path_key == format!("{db_key}-shm") - { - continue; - } - } - let path = entry.path().to_string_lossy().into_owned(); - let filename = entry.file_name().to_string_lossy().into_owned(); - transaction.execute("INSERT INTO seen_paths(path) VALUES (?1)", [&path])?; - report.discovered += 1; - let metadata = entry.metadata()?; - let size = i64::try_from(metadata.len()).context("file exceeds supported size")?; - let modified = metadata - .modified()? - .duration_since(UNIX_EPOCH) - .context("file timestamp predates Unix epoch")?; - let mtime = i64::try_from(modified.as_nanos()) - .context("file timestamp exceeds supported range")?; - let extension = entry - .path() - .extension() - .and_then(|value| value.to_str()) - .unwrap_or("") - .to_ascii_lowercase(); - let previous: Option<(i64, i64, String, Option)> = transaction - .query_row( - "SELECT size,mtime,root,extracted_sha256 FROM files WHERE path=?1", - [&path], - |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), - ) - .optional()?; - // Supported inputs are bounded and fingerprinted even when file - // metadata is unchanged. This detects same-stat overwrites while - // unsupported media/giant files retain the metadata-only fast path. - let input = extraction::read_source(entry.path()); - if matches!(&input, Ok(Some(_))) { - let after = fs::metadata(entry.path())?; - if after.len() != metadata.len() || after.modified()? != metadata.modified()? { - bail!("Source changed during extraction read; retry indexing: {path}; reconciliation rolled back"); - } - } - let source_hash = input - .as_ref() - .ok() - .and_then(|input| input.as_ref()) - .map(|input| input.hash.clone()); - if previous - .as_ref() - .is_some_and(|(old_size, old_mtime, old_root, old_hash)| { - *old_size == size - && *old_mtime == mtime - && old_root == &root_key - && input.is_ok() - && old_hash == &source_hash + for entry in WalkDir::new(scope) + .follow_links(false) + .into_iter() + .filter_entry(|entry| { + !options + .excluded_paths + .iter() + .any(|path| entry.path().starts_with(path)) + && (entry.depth() == 0 + || !entry.file_type().is_dir() + || !options.excluded_directory_names.iter().any(|name| { + entry + .file_name() + .to_string_lossy() + .eq_ignore_ascii_case(name) + })) }) { - report.unchanged += 1; - continue; - } - let extracted = input.and_then(|input| match input { - Some(input) => extraction::extract(entry.path(), input), - None => Ok(None), - }); - let (extracted, extraction_error) = match extracted { - Ok(extracted) => (extracted, None), - Err(error) if matches!(extension.as_str(), "pdf" | "docx") => { - report.failures += 1; - let reason: String = format!("{error:#}").chars().take(2048).collect(); - (None, Some(reason)) + check_index_cancelled(cancelled)?; + let entry = + entry.context("directory traversal failed; reconciliation rolled back")?; + if !entry.file_type().is_file() { + continue; } - Err(error) => { - return Err(error) - .with_context(|| format!("read {path}; reconciliation rolled back")) + if let Some(database) = &self.database_path { + let path_key = entry.path().to_string_lossy(); + let db_key = database.to_string_lossy(); + if path_key == db_key + || path_key == format!("{db_key}-wal") + || path_key == format!("{db_key}-shm") + { + continue; + } } - }; - if extracted.is_none() { - report.skipped += 1; - } - let extracted = extracted.unwrap_or_default(); - let content = extracted - .iter() - .map(|passage| passage.text.as_str()) - .collect::>() - .join("\n"); - transaction.execute( - "INSERT INTO files(path,root,filename,content,size,mtime,extension,extraction_error,extracted_sha256) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9) + let path = entry.path().to_string_lossy().into_owned(); + let media_hint = media_hints && searchi_media::classify(entry.path()).is_some(); + let filename = entry.file_name().to_string_lossy().into_owned(); + transaction.execute("INSERT INTO seen_paths(path) VALUES (?1)", [&path])?; + report.discovered += 1; + let metadata = entry.metadata()?; + let size = i64::try_from(metadata.len()).context("file exceeds supported size")?; + let modified = metadata + .modified()? + .duration_since(UNIX_EPOCH) + .context("file timestamp predates Unix epoch")?; + let mtime = i64::try_from(modified.as_nanos()) + .context("file timestamp exceeds supported range")?; + let extension = entry + .path() + .extension() + .and_then(|value| value.to_str()) + .unwrap_or("") + .to_ascii_lowercase(); + let previous: Option<(i64, i64, String, Option)> = transaction + .query_row( + "SELECT size,mtime,root,extracted_sha256 FROM files WHERE path=?1", + [&path], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional()?; + // Supported inputs are bounded and fingerprinted even when file + // metadata is unchanged. This detects same-stat overwrites while + // unsupported media/giant files retain the metadata-only fast path. + let input = extraction::read_source(entry.path()); + // A parser rejection may retain a filename-only record, but a + // filesystem read failure cannot certify this reconciliation. + if input + .as_ref() + .is_err_and(|error| error.chain().any(|cause| cause.is::())) + { + return Err(input.err().expect("Checked source read error")) + .with_context(|| format!("read {path}; reconciliation rolled back")); + } + if matches!(&input, Ok(Some(_))) { + let after = fs::metadata(entry.path())?; + if after.len() != metadata.len() || after.modified()? != metadata.modified()? { + bail!("Source changed during extraction read; retry indexing: {path}; reconciliation rolled back"); + } + } + let source_hash = input + .as_ref() + .ok() + .and_then(|input| input.as_ref()) + .map(|input| input.hash.clone()); + if !media_hint + && previous + .as_ref() + .is_some_and(|(old_size, old_mtime, old_root, old_hash)| { + *old_size == size + && *old_mtime == mtime + && old_root == &root_key + && input.is_ok() + && old_hash == &source_hash + }) + { + report.unchanged += 1; + continue; + } + let extracted = input.and_then(|input| match input { + Some(input) => extraction::extract(entry.path(), input), + None => Ok(None), + }); + let (extracted, extraction_error) = match extracted { + Ok(extracted) => (extracted, None), + Err(error) if matches!(extension.as_str(), "pdf" | "docx") => { + report.failures += 1; + let reason: String = format!("{error:#}").chars().take(2048).collect(); + (None, Some(reason)) + } + Err(error) => { + return Err(error) + .with_context(|| format!("read {path}; reconciliation rolled back")) + } + }; + if extracted.is_none() { + report.skipped += 1; + } + let extracted = extracted.unwrap_or_default(); + let content = extracted + .iter() + .map(|passage| passage.text.as_str()) + .collect::>() + .join("\n"); + transaction.execute( + "INSERT INTO files(path,root,filename,content,size,mtime,extension,extraction_error,extracted_sha256,change_revision) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10) ON CONFLICT(path) DO UPDATE SET root=excluded.root, filename=excluded.filename, content=excluded.content, size=excluded.size, mtime=excluded.mtime, - extension=excluded.extension, extraction_error=excluded.extraction_error, extracted_sha256=excluded.extracted_sha256", - params![path, root_key, filename, content, size, mtime, extension, extraction_error, source_hash], + extension=excluded.extension, extraction_error=excluded.extraction_error, extracted_sha256=excluded.extracted_sha256, + change_revision=files.change_revision+excluded.change_revision", + params![path, root_key, filename, content, size, mtime, extension, extraction_error, source_hash, i64::from(media_hint)], )?; - transaction.execute("DELETE FROM passages WHERE file_path=?1", [&path])?; - for passage in extracted { - transaction.execute( - "INSERT INTO passages(text,file_path,kind,number) VALUES (?1,?2,?3,?4)", - params![ - passage.text, - path, - passage.source.kind, - passage.source.number - ], - )?; + if media_hint { + transaction.execute("DELETE FROM media_metadata WHERE path=?1", [&path])?; + } + transaction.execute("DELETE FROM passages WHERE file_path=?1", [&path])?; + for passage in extracted { + transaction.execute( + "INSERT INTO passages(text,file_path,kind,number) VALUES (?1,?2,?3,?4)", + params![ + passage.text, + path, + passage.source.kind, + passage.source.number + ], + )?; + } + report.indexed += 1; } - report.indexed += 1; } - report.removed = transaction.execute( - "DELETE FROM files WHERE root=?1 - AND NOT EXISTS (SELECT 1 FROM seen_paths WHERE seen_paths.path=files.path)", - [&root_key], - )?; + // Recheck root availability before deleting records. A disconnected + // volume must never look like successful reconciliation of empty scopes. + let root_metadata = + fs::symlink_metadata(root).context("Recheck index root availability")?; + if !root_metadata.is_dir() || root_metadata.file_type().is_symlink() { + bail!("Index root became unavailable"); + } + for scope in scopes { + let key = scope.to_string_lossy(); + let prefix = format!("{}{}", key, std::path::MAIN_SEPARATOR); + report.removed += transaction.execute( + "DELETE FROM files WHERE root=?1 AND (?4 OR path=?2 OR substr(path,1,length(?3))=?3) + AND NOT EXISTS (SELECT 1 FROM seen_paths WHERE seen_paths.path=files.path)", + params![root_key, key, prefix, scope == root], + )?; + } transaction.execute("DELETE FROM seen_paths", [])?; + check_index_cancelled(cancelled)?; transaction.commit()?; self.gram_frequencies.borrow_mut().clear(); Ok(report) @@ -1164,6 +1376,28 @@ fn substring_anchor(characters: &[char], frequencies: &[(i64, String)]) -> Strin characters[start..end].iter().collect() } +fn check_index_cancelled(cancelled: Option<&AtomicBool>) -> Result<()> { + if cancelled.is_some_and(|flag| flag.load(Ordering::Acquire)) { + bail!("Index reconciliation cancelled; transaction rolled back"); + } + Ok(()) +} + +fn validate_normalized_absolute(path: &Path) -> Result<()> { + if !path.is_absolute() + || path.components().any(|c| { + matches!( + c, + std::path::Component::ParentDir | std::path::Component::CurDir + ) + }) + || path.components().collect::().as_os_str() != path.as_os_str() + { + bail!("Path must be absolute and component-normalized"); + } + Ok(()) +} + #[cfg(test)] mod tests { #[test] diff --git a/crates/searchi-core/tests/index_options.rs b/crates/searchi-core/tests/index_options.rs index 44e6d9a..968a118 100644 --- a/crates/searchi-core/tests/index_options.rs +++ b/crates/searchi-core/tests/index_options.rs @@ -12,6 +12,7 @@ fn exclusions_database_and_unchanged_metadata() -> Result<()> { fs::write(root.path().join("visible.txt"), "original content")?; let mut index = SearchIndex::open(root.path().join("index.db"))?; let options = IndexOptions { + excluded_paths: Vec::new(), excluded_directory_names: vec![ ".git".into(), "node_modules".into(), diff --git a/crates/searchi-core/tests/targeted_reconciliation.rs b/crates/searchi-core/tests/targeted_reconciliation.rs new file mode 100644 index 0000000..800df31 --- /dev/null +++ b/crates/searchi-core/tests/targeted_reconciliation.rs @@ -0,0 +1,484 @@ +//! Atomic bounded event reconciliation, independent of platform watcher delivery. +use anyhow::Result; +use searchi_core::{IndexOptions, SearchIndex}; +use std::{fs, path::PathBuf, sync::atomic::AtomicBool}; + +#[test] +fn same_stat_overwrite_and_unrelated_files_are_preserved() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + let file = root.join("changed.txt"); + fs::write(&file, "oldword alpha")?; + fs::write(root.join("other.txt"), "unrelated beta")?; + let root = fs::canonicalize(root)?; + let file = fs::canonicalize(file)?; + let mut index = SearchIndex::open(temp.path().join("index.db"))?; + index.index_folder(&root)?; + let mtime = fs::metadata(&file)?.modified()?; + fs::write(&file, "newword alpha")?; + fs::File::options() + .write(true) + .open(&file)? + .set_times(fs::FileTimes::new().set_modified(mtime))?; + let report = index.index_paths_with_options(&root, &[file], &IndexOptions::default())?; + assert_eq!(report.discovered, 1); + assert_eq!(report.indexed, 1); + assert_eq!(index.search("newword", 10)?.len(), 1); + assert!(index.search("oldword", 10)?.is_empty()); + assert_eq!(index.search("unrelated", 10)?.len(), 1); + Ok(()) +} + +#[test] +fn rename_subtree_deletion_and_literal_boundaries_preserve_other_roots() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + let child = root.join("a_%"); + fs::create_dir(&child)?; + let sibling = root.join("a_%suffix"); + fs::create_dir(&sibling)?; + fs::write(child.join("old.txt"), "firstrecord")?; + fs::write(sibling.join("keep.txt"), "siblingrecord")?; + let other = temp.path().join("other"); + fs::create_dir(&other)?; + fs::write(other.join("keep.txt"), "otherrootrecord")?; + let root = fs::canonicalize(root)?; + let child = root.join("a_%"); + let old = child.join("old.txt"); + let new = child.join("renamed.txt"); + let mut index = SearchIndex::open(temp.path().join("index.db"))?; + index.index_folder(&root)?; + index.index_folder(&other)?; + fs::rename(&old, &new)?; + let report = index.index_paths_with_options( + &root, + &[old, new.clone(), new], + &IndexOptions::default(), + )?; + assert_eq!(report.indexed, 1); + assert_eq!(report.removed, 1); + assert_eq!(index.search("firstrecord", 10)?[0].filename, "renamed.txt"); + fs::remove_dir_all(&child)?; + let report = index.index_paths_with_options(&root, &[child], &IndexOptions::default())?; + assert_eq!(report.removed, 1); + assert_eq!(index.search("siblingrecord", 10)?.len(), 1); + assert_eq!(index.search("otherrootrecord", 10)?.len(), 1); + Ok(()) +} + +#[test] +fn directory_events_reconcile_only_subtree_and_prune_exclusions() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + fs::create_dir(root.join("nested"))?; + fs::create_dir(root.join("target"))?; + fs::write(root.join("nested/one.txt"), "oldnested")?; + fs::write(root.join("top.txt"), "toporiginal")?; + let root = fs::canonicalize(root)?; + let mut index = SearchIndex::open(root.join("index.db"))?; + index.index_folder(&root)?; + fs::remove_file(root.join("nested/one.txt"))?; + fs::write(root.join("nested/two.txt"), "newnested")?; + fs::write(root.join("top.txt"), "topchanged")?; + fs::write(root.join("target/excluded.txt"), "excludedcontent")?; + let report = index.index_paths_with_options( + &root, + &[ + root.join("nested"), + root.join("nested/two.txt"), + root.join("target"), + ], + &IndexOptions::default(), + )?; + assert_eq!(report.discovered, 1); + assert_eq!(report.removed, 1); + assert_eq!(index.search("newnested", 10)?.len(), 1); + assert_eq!(index.search("toporiginal", 10)?.len(), 1); + assert!(index.search("topchanged", 10)?.is_empty()); + assert!(index.search("excludedcontent", 10)?.is_empty()); + assert!(index.search("index.db", 10)?.is_empty()); + Ok(()) +} + +#[test] +fn invalid_batches_and_failed_reads_roll_back_all_changes() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + fs::write(root.join("a.txt"), "beforetransaction")?; + fs::write(root.join("z.txt"), "validlast")?; + let root = fs::canonicalize(root)?; + let mut index = SearchIndex::open(temp.path().join("index.db"))?; + index.index_folder(&root)?; + fs::write(root.join("a.txt"), "aftertransaction")?; + #[cfg(windows)] + let locked = { + use std::os::windows::fs::OpenOptionsExt; + fs::OpenOptions::new() + .read(true) + .share_mode(0) + .open(root.join("z.txt"))? + }; + #[cfg(unix)] + let permissions = { + use std::os::unix::fs::PermissionsExt; + let previous = fs::metadata(root.join("z.txt"))?.permissions(); + fs::set_permissions(root.join("z.txt"), fs::Permissions::from_mode(0))?; + previous + }; + let failed = index.index_paths_with_options( + &root, + &[root.join("a.txt"), root.join("z.txt")], + &IndexOptions::default(), + ); + #[cfg(windows)] + drop(locked); + #[cfg(unix)] + fs::set_permissions(root.join("z.txt"), permissions)?; + assert!(failed.is_err()); + assert_eq!(index.search("beforetransaction", 10)?.len(), 1); + assert!(index.search("aftertransaction", 10)?.is_empty()); + assert!(index + .index_paths_with_options( + &root, + &[temp.path().join("outside.txt")], + &IndexOptions::default() + ) + .is_err()); + assert!(index + .index_paths_with_options( + &root, + &[PathBuf::from("relative.txt")], + &IndexOptions::default() + ) + .is_err()); + assert!(index + .index_paths_with_options( + &root, + &[root.join("../outside.txt")], + &IndexOptions::default() + ) + .is_err()); + assert!(index + .index_paths_with_options( + &root, + &vec![root.join("a.txt"); 129], + &IndexOptions::default() + ) + .is_err()); + Ok(()) +} + +#[test] +fn cancellation_and_unavailable_root_preserve_records() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + fs::write(root.join("file.txt"), "preservedrecord")?; + let root = fs::canonicalize(root)?; + let mut index = SearchIndex::open(temp.path().join("index.db"))?; + index.index_folder(&root)?; + fs::remove_file(root.join("file.txt"))?; + let cancelled = AtomicBool::new(true); + assert!(index + .index_paths_cancellable( + &root, + &[root.join("file.txt")], + &IndexOptions::default(), + &cancelled + ) + .is_err()); + assert!(index + .index_folder_cancellable(&root, &IndexOptions::default(), &cancelled) + .is_err()); + assert_eq!(index.search("preservedrecord", 10)?.len(), 1); + fs::remove_dir(&root)?; + assert!(index + .index_paths_with_options(&root, &[root.join("file.txt")], &IndexOptions::default()) + .is_err()); + assert_eq!(index.search("preservedrecord", 10)?.len(), 1); + Ok(()) +} + +#[cfg(unix)] +#[test] +fn symlink_ancestor_cannot_index_outside_content() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + let outside = temp.path().join("outside"); + fs::create_dir(&outside)?; + fs::write(outside.join("secret.txt"), "secretoutside")?; + let root = fs::canonicalize(root)?; + std::os::unix::fs::symlink(&outside, root.join("link"))?; + let mut index = SearchIndex::open(temp.path().join("index.db"))?; + assert!(index + .index_paths_with_options( + &root, + &[root.join("link/secret.txt")], + &IndexOptions::default() + ) + .is_err()); + assert!(index.search("secretoutside", 10)?.is_empty()); + Ok(()) +} + +#[test] +fn explicit_exclusions_remove_prior_records_without_prefix_leakage() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + fs::create_dir(root.join("cache"))?; + fs::create_dir(root.join("cache-extra"))?; + fs::write(root.join("cache/hidden.txt"), "previouslyvisible")?; + fs::write(root.join("cache-extra/keep.txt"), "preservedneighbor")?; + let root = fs::canonicalize(root)?; + let mut index = SearchIndex::open(temp.path().join("index.db"))?; + index.index_folder(&root)?; + let mut options = IndexOptions::default(); + options.excluded_paths.push(root.join("cache")); + let report = index.index_paths_with_options(&root, &[root.join("cache")], &options)?; + assert_eq!(report.removed, 1); + assert!(index.search("previouslyvisible", 10)?.is_empty()); + assert_eq!(index.search("preservedneighbor", 10)?.len(), 1); + assert_eq!( + index.index_folder_with_options(&root, &options)?.discovered, + 1 + ); + options.excluded_paths.push(PathBuf::from("relative")); + assert!(index.index_folder_with_options(&root, &options).is_err()); + Ok(()) +} + +#[test] +fn independent_reader_remains_available_and_sees_atomic_update() -> Result<()> { + use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }; + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + for number in 0..64 { + fs::write(root.join(format!("{number}.txt")), "oldversion")?; + } + let root = fs::canonicalize(root)?; + let db = temp.path().join("index.db"); + let mut writer = SearchIndex::open(&db)?; + writer.index_folder(&root)?; + let reader = SearchIndex::open(&db)?; + for number in 0..64 { + fs::write( + root.join(format!("{number}.txt")), + format!("newversion {}", "content ".repeat(4000)), + )?; + } + let finished = Arc::new(AtomicBool::new(false)); + let done = finished.clone(); + std::thread::scope(|scope| -> Result<()> { + let task = scope.spawn(move || { + let result = writer.index_paths_with_options( + &root, + std::slice::from_ref(&root), + &IndexOptions::default(), + ); + done.store(true, Ordering::Release); + result + }); + let mut observations = 0; + while !finished.load(Ordering::Acquire) { + let count = reader.search("oldversion", 100)?.len(); + assert!( + count == 64 || count == 0, + "Reader must never see partially committed records" + ); + observations += 1; + } + task.join().unwrap()?; + assert!(observations > 0); + assert_eq!(reader.search("newversion", 100)?.len(), 64); + Ok(()) + })?; + Ok(()) +} + +#[test] +fn cancellation_after_write_transaction_begins_rolls_back_batch() -> Result<()> { + use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, + time::{Duration, Instant}, + }; + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + for number in 0..128 { + fs::write(root.join(format!("{number}.txt")), "oldcancelrecord")?; + } + let root = fs::canonicalize(root)?; + let db = temp.path().join("index.db"); + let mut writer = SearchIndex::open(&db)?; + writer.index_folder(&root)?; + let reader = SearchIndex::open(&db)?; + for number in 0..128 { + fs::write( + root.join(format!("{number}.txt")), + format!("newcancelrecord {}", "new content ".repeat(8000)), + )?; + } + let probe = rusqlite::Connection::open(&db)?; + probe.busy_timeout(Duration::ZERO)?; + let cancelled = Arc::new(AtomicBool::new(false)); + let flag = cancelled.clone(); + std::thread::scope(|scope| -> Result<()> { + let indexing = scope + .spawn(move || writer.index_folder_cancellable(&root, &IndexOptions::default(), &flag)); + let deadline = Instant::now() + Duration::from_secs(5); + let mut observed_write = false; + while Instant::now() < deadline { + match probe.execute_batch("BEGIN IMMEDIATE") { + Ok(()) => probe.execute_batch("ROLLBACK")?, + Err(rusqlite::Error::SqliteFailure(error, _)) + if error.code == rusqlite::ErrorCode::DatabaseBusy => + { + observed_write = true; + cancelled.store(true, Ordering::Release); + break; + } + Err(error) => return Err(error.into()), + } + std::thread::sleep(Duration::from_millis(1)); + } + let result = indexing.join().unwrap(); + assert!( + observed_write, + "Must observe the writer transaction before cancellation" + ); + assert!(result.is_err()); + assert_eq!(reader.search("oldcancelrecord", 200)?.len(), 128); + assert!(reader.search("newcancelrecord", 200)?.is_empty()); + Ok(()) + })?; + Ok(()) +} + +#[test] +fn same_stat_media_hint_advances_revision_and_removes_only_its_tags() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + fs::write(root.join("changed.wav"), [1u8, 2, 3, 4])?; + fs::write(root.join("unrelated.wav"), [5u8, 6, 7, 8])?; + let root = fs::canonicalize(root)?; + let file = root.join("changed.wav"); + let db = temp.path().join("index.db"); + let mut index = SearchIndex::open(&db)?; + index.index_folder(&root)?; + let path = file.to_string_lossy().into_owned(); + let (size, mtime) = index.file_signature(&path)?.unwrap(); + index.set_media_metadata(&path, "stale artist metadata", 1000, size, mtime)?; + let original_mtime = fs::metadata(&file)?.modified()?; + fs::write(&file, [4u8, 3, 2, 1])?; + fs::OpenOptions::new() + .write(true) + .open(&file)? + .set_times(fs::FileTimes::new().set_modified(original_mtime))?; + assert_eq!( + index.index_folder(&root)?.unchanged, + 2, + "Periodic metadata scan retains fast path" + ); + let connection = rusqlite::Connection::open(&db)?; + let revisions = || -> Result> { + let mut statement = + connection.prepare("SELECT filename,change_revision FROM files ORDER BY filename")?; + let rows = statement.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?; + Ok(rows.collect::>()?) + }; + assert_eq!( + revisions()?, + vec![("changed.wav".into(), 0), ("unrelated.wav".into(), 0)] + ); + assert_eq!(index.search("stale artist", 10)?.len(), 1); + let report = index.index_paths_with_options(&root, &[file], &IndexOptions::default())?; + assert_eq!(report.indexed, 1); + assert_eq!(report.unchanged, 0); + assert_eq!( + revisions()?, + vec![("changed.wav".into(), 1), ("unrelated.wav".into(), 0)] + ); + assert!(index.search("stale artist", 10)?.is_empty()); + assert_eq!(index.search("changed.wav", 10)?.len(), 1); + let report = index.index_paths_with_options( + &root, + std::slice::from_ref(&root), + &IndexOptions::default(), + )?; + assert_eq!(report.indexed, 2); + assert_eq!( + revisions()?, + vec![("changed.wav".into(), 2), ("unrelated.wav".into(), 1)] + ); + Ok(()) +} + +#[test] +fn schema_twelve_migrates_change_revision_without_reextracting_records() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + fs::write(root.join("document.txt"), "preserved migration text")?; + let db = temp.path().join("index.db"); + let mut index = SearchIndex::open(&db)?; + index.index_folder(&root)?; + drop(index); + let connection = rusqlite::Connection::open(&db)?; + connection + .execute_batch("ALTER TABLE files DROP COLUMN change_revision; PRAGMA user_version=12;")?; + drop(connection); + let mut index = SearchIndex::open(&db)?; + assert_eq!(index.search("migration text", 10)?.len(), 1); + assert_eq!(index.index_folder(&root)?.unchanged, 1); + let connection = rusqlite::Connection::open(&db)?; + assert_eq!( + connection.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))?, + 13 + ); + assert_eq!( + connection.query_row("SELECT change_revision FROM files", [], |row| row + .get::<_, i64>(0))?, + 0 + ); + Ok(()) +} + +#[test] +fn metadata_probe_revision_rejects_tags_after_same_stat_hint() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("root"); + fs::create_dir(&root)?; + fs::write(root.join("song.wav"), [1u8, 2, 3, 4])?; + let root = fs::canonicalize(root)?; + let file = root.join("song.wav"); + let path = file.to_string_lossy().into_owned(); + let mut index = SearchIndex::open(temp.path().join("index.db"))?; + index.index_folder(&root)?; + let (size, mtime) = index.file_signature(&path)?.unwrap(); + let before = index.media_change_revision(&path, size, mtime)?; + index.index_paths_with_options(&root, &[file], &IndexOptions::default())?; + assert!(index + .set_media_metadata_at_revision(&path, "stale probe tags", 1000, size, mtime, before) + .is_err()); + assert!(index.search("stale probe", 10)?.is_empty()); + let current = index.media_change_revision(&path, size, mtime)?; + assert_eq!(current, before + 1); + index.set_media_metadata_at_revision(&path, "fresh probe tags", 1000, size, mtime, current)?; + assert_eq!(index.search("fresh probe", 10)?.len(), 1); + assert!(index.media_change_revision(&path, size + 1, mtime).is_err()); + Ok(()) +} diff --git a/crates/searchi-media/src/lib.rs b/crates/searchi-media/src/lib.rs index 067e9af..81fc185 100644 --- a/crates/searchi-media/src/lib.rs +++ b/crates/searchi-media/src/lib.rs @@ -93,6 +93,29 @@ mod tests { Ok(()) } + #[test] + fn owned_frame_scratch_validates_parent_and_cleans_batch() -> Result<()> { + let root = tempfile::tempdir()?; + let parent = root.path().join("owned scratch"); + let limits = MediaLimits { + scratch_directory: Some(parent.clone()), + ..Default::default() + }; + let batch = frame_scratch(&limits)?; + assert_eq!(batch.path().parent(), Some(parent.as_path())); + let batch_path = batch.path().to_owned(); + fs::write(batch.path().join("frame.png"), b"owned temporary bytes")?; + drop(batch); + assert!(!batch_path.exists()); + assert!(parent.is_dir()); + let relative = MediaLimits { + scratch_directory: Some(PathBuf::from("relative")), + ..Default::default() + }; + assert!(frame_scratch(&relative).is_err()); + Ok(()) + } + #[test] fn identity_rejects_changes_and_oversized_input() -> Result<()> { let temp = tempfile::tempdir()?; @@ -170,10 +193,19 @@ mod tests { ]) .arg(&video); run(command, Duration::from_secs(30), 4096, 65536)?; - let limits = MediaLimits::default(); + let scratch = temp.path().join("owned scratch"); + let limits = MediaLimits { + scratch_directory: Some(scratch.clone()), + ..Default::default() + }; let probe = probe(&tools, &video, &limits)?; assert!(probe.has_video && probe.has_audio); let frames = decode_frames(&tools, &video, &probe, 2000, false, &limits)?; + assert!(frames.directory().starts_with(&scratch)); + assert!(frames + .frames + .iter() + .all(|frame| frame.path.starts_with(&scratch))); assert_eq!(frames.frames.len(), 1); assert!((1900..=2100).contains(&frames.frames[0].timestamp_ms)); let audio = decode_audio( @@ -312,6 +344,10 @@ pub struct MediaLimits { pub audio_overlap_ms: u32, pub max_audio_windows: usize, pub command_timeout: Duration, + /// Owned decoded-frame scratch parent. Desktop callers place this below + /// excluded application data to avoid indexing their own temporary frames. + /// None preserves the normal operating-system temporary directory. + pub scratch_directory: Option, } impl Default for MediaLimits { fn default() -> Self { @@ -325,6 +361,7 @@ impl Default for MediaLimits { audio_overlap_ms: 2000, max_audio_windows: 600, command_timeout: Duration::from_secs(45), + scratch_directory: None, } } } @@ -615,6 +652,23 @@ impl FrameBatch { self.directory.path() } } +fn frame_scratch(limits: &MediaLimits) -> Result { + match &limits.scratch_directory { + Some(parent) => { + ensure!( + parent.is_absolute(), + "Media scratch directory must be absolute" + ); + fs::create_dir_all(parent).context("Create owned media scratch directory")?; + tempfile::Builder::new() + .prefix("searchi-frames-") + .tempdir_in(parent) + .context("Create owned frame batch") + } + None => tempfile::tempdir().context("Create frame batch"), + } +} + pub fn decode_frames( tools: &FfmpegTools, path: &Path, @@ -623,7 +677,7 @@ pub fn decode_frames( include_scene: bool, limits: &MediaLimits, ) -> Result { - let directory = tempfile::tempdir()?; + let directory = frame_scratch(limits)?; let mut command = Command::new(&tools.ffmpeg); command.args(["-protocol_whitelist", "file,pipe"]); command.args(["-nostdin","-hide_banner","-loglevel","info","-threads","1","-hwaccel","none","-copyts","-ss"]) diff --git a/crates/searchi-service/src/lib.rs b/crates/searchi-service/src/lib.rs index 134ada1..869d202 100644 --- a/crates/searchi-service/src/lib.rs +++ b/crates/searchi-service/src/lib.rs @@ -272,7 +272,7 @@ impl HybridSearch { }); } ensure!( - (6..=7).contains(&version), + (6..=8).contains(&version), "Unsupported semantic schema version {version}" ); let (generation,fingerprint,revision,current):(Option,Option,Option,i64)=connection.query_row( diff --git a/crates/searchi-service/src/media.rs b/crates/searchi-service/src/media.rs index 778e576..1be7800 100644 --- a/crates/searchi-service/src/media.rs +++ b/crates/searchi-service/src/media.rs @@ -141,6 +141,11 @@ impl HybridSearch { task: &SourceTask, work: &mut BuildWork, ) -> Result<()> { + let change_revision = self.lexical.media_change_revision( + &path.to_string_lossy(), + identity.bytes, + identity.modified_ns, + )?; if let Some(count) = self .store .reuse_media_source(generation, task, id, identity)? @@ -152,7 +157,7 @@ impl HybridSearch { if kind != MediaKind::Image { // Metadata remains part of music search after path remapping. let metadata = probe(tools, path, limits)?; - self.refresh_media_metadata(path, identity, &metadata)?; + self.refresh_media_metadata(path, identity, &metadata, change_revision)?; } identity.ensure_unchanged()?; work.report.chunks += count; @@ -169,6 +174,7 @@ impl HybridSearch { tools, limits, identity, + change_revision, &mut work.report, )?; self.store @@ -182,6 +188,7 @@ impl HybridSearch { path: &Path, identity: &MediaIdentity, metadata: &MediaProbe, + change_revision: i64, ) -> Result<()> { let tags = metadata .tags @@ -189,12 +196,13 @@ impl HybridSearch { .map(|(k, v)| format!("{k}: {v}")) .collect::>() .join("\n"); - self.lexical.set_media_metadata( + self.lexical.set_media_metadata_at_revision( &path.to_string_lossy(), &tags, metadata.duration_ms, identity.bytes, identity.modified_ns, + change_revision, ) } @@ -208,6 +216,7 @@ impl HybridSearch { tools: &FfmpegTools, limits: &MediaLimits, identity: &MediaIdentity, + change_revision: i64, report: &mut EmbeddingReport, ) -> Result<()> { let mut ordinal = 0; @@ -228,7 +237,7 @@ impl HybridSearch { report.chunks += 1; } else { let metadata = probe(tools, path, limits)?; - self.refresh_media_metadata(path, identity, &metadata)?; + self.refresh_media_metadata(path, identity, &metadata, change_revision)?; if metadata.has_video && kind == MediaKind::Video { let mut seen = std::collections::HashSet::new(); let mut scenes = 0; diff --git a/crates/searchi-service/src/store.rs b/crates/searchi-service/src/store.rs index 00ad1a2..c60356a 100644 --- a/crates/searchi-service/src/store.rs +++ b/crates/searchi-service/src/store.rs @@ -111,7 +111,7 @@ impl Store { |row| row.get(0), )?; ensure!( - (1..=7).contains(&version), + (1..=8).contains(&version), "Unsupported semantic schema version {version}" ); initialization.execute_batch(" @@ -273,7 +273,7 @@ impl Store { ) }; let bytes: Vec = vector.iter().flat_map(|v| v.to_le_bytes()).collect(); - self.connection.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime) VALUES(?1,?2,?3,?4,?5,?6,?7,0,0,?8,?9,?10,?11,?12)",params![generation,-file_id,ordinal as i64,path,kind,timestamp,identity.sha256,snippet,bytes,source_type,identity.bytes as i64,identity.modified_ns])?; + self.connection.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,file_change_revision) VALUES(?1,?2,?3,?4,?5,?6,?7,0,0,?8,?9,?10,?11,?12,(SELECT change_revision FROM files WHERE path=?4))",params![generation,-file_id,ordinal as i64,path,kind,timestamp,identity.sha256,snippet,bytes,source_type,identity.bytes as i64,identity.modified_ns])?; Ok(()) } @@ -991,7 +991,7 @@ fn collect_eligibility( }; let mut statement=connection.prepare(&format!("SELECT c.rowid,c.passage_id,c.content_hash,p.text,c.path,c.source_type FROM semantic_chunks c LEFT JOIN passages p ON c.source_type='text' AND p.rowid=c.passage_id AND p.file_path=c.path AND p.kind=c.kind AND p.number=c.number - JOIN files f ON f.path=c.path WHERE c.generation=?1 AND ((c.source_type='text' AND p.rowid IS NOT NULL) OR (c.source_type!='text' AND c.file_size=f.size AND c.file_mtime=f.mtime)) + JOIN files f ON f.path=c.path WHERE c.generation=?1 AND ((c.source_type='text' AND p.rowid IS NOT NULL) OR (c.source_type!='text' AND c.file_size=f.size AND c.file_mtime=f.mtime AND c.file_change_revision=f.change_revision)) AND (?2 IS NULL OR substr(f.path,1,length(?2))=?2) AND (?3 IS NULL OR f.extension=?3) AND (?4 IS NULL OR f.size>=?4) AND (?5 IS NULL OR f.size<=?5) AND (?6 IS NULL OR f.mtime/1000000>=?6) AND (?7 IS NULL OR f.mtime/1000000<=?7) diff --git a/crates/searchi-service/src/store/incremental.rs b/crates/searchi-service/src/store/incremental.rs index 0837685..9f4df1b 100644 --- a/crates/searchi-service/src/store/incremental.rs +++ b/crates/searchi-service/src/store/incremental.rs @@ -22,6 +22,11 @@ pub(super) fn migrate(connection: &Connection) -> Result<()> { "INTEGER NOT NULL DEFAULT 0", ), ("semantic_chunks", "task_key", "TEXT"), + ( + "semantic_chunks", + "file_change_revision", + "INTEGER NOT NULL DEFAULT 0", + ), ] { let exists: bool = connection.query_row( "SELECT EXISTS(SELECT 1 FROM pragma_table_info(?1) WHERE name=?2)", @@ -43,7 +48,7 @@ pub(super) fn migrate(connection: &Connection) -> Result<()> { PRIMARY KEY(generation,task_key)); CREATE INDEX IF NOT EXISTS semantic_task_reuse ON semantic_tasks(source_hash,title_hash,policy,source_type,state); CREATE INDEX IF NOT EXISTS semantic_chunk_task ON semantic_chunks(generation,task_key); - UPDATE semantic_meta SET schema_version=7 WHERE singleton=1;")?; + UPDATE semantic_meta SET schema_version=8 WHERE singleton=1;")?; Ok(()) } @@ -294,8 +299,8 @@ impl Store { for (key, donor) in mappings { tx.execute("INSERT INTO semantic_tasks(generation,task_key,path,source_type,source_hash,title_hash,policy,passage_id,state,attempts,expected_chunks,message) SELECT ?1,d.task_key,d.path,d.source_type,d.source_hash,d.title_hash,d.policy,d.passage_id,'done',old.attempts,old.expected_chunks,NULL FROM semantic_discovery_tasks d JOIN semantic_prune_tasks old ON old.task_key=?3 WHERE d.task_key=?2",params![generation,key,donor])?; - tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key) - SELECT ?1,d.passage_id,c.ordinal,d.path,c.kind,c.number,c.content_hash,c.start_byte,c.end_byte,c.snippet,c.vector,c.source_type,c.file_size,c.file_mtime,d.task_key FROM semantic_prune_chunks c JOIN semantic_discovery_tasks d ON d.task_key=?2 WHERE c.task_key=?3",params![generation,key,donor])?; + tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key,file_change_revision) + SELECT ?1,d.passage_id,c.ordinal,d.path,c.kind,c.number,c.content_hash,c.start_byte,c.end_byte,c.snippet,c.vector,c.source_type,c.file_size,c.file_mtime,d.task_key,COALESCE((SELECT change_revision FROM files WHERE path=d.path),c.file_change_revision) FROM semantic_prune_chunks c JOIN semantic_discovery_tasks d ON d.task_key=?2 WHERE c.task_key=?3",params![generation,key,donor])?; } tx.execute( "DELETE FROM semantic_media_errors WHERE generation=?1", @@ -530,8 +535,8 @@ impl Store { "DELETE FROM semantic_chunks WHERE generation=?1 AND task_key=?2", params![generation, task.key], )?; - tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key) - SELECT ?1,?2,ordinal,?3,COALESCE(?4,kind),COALESCE(?5,number),content_hash,start_byte,end_byte,snippet,vector,source_type,COALESCE(?6,file_size),COALESCE(?7,file_mtime),?8 FROM semantic_reuse_chunks",params![generation,task.passage_id,task.path,location.map(|s|s.kind.as_str()),location.map(|s|s.number),identity.map(|i|i.bytes as i64),identity.map(|i|i.modified_ns),task.key])?; + tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key,file_change_revision) + SELECT ?1,?2,ordinal,?3,COALESCE(?4,kind),COALESCE(?5,number),content_hash,start_byte,end_byte,snippet,vector,source_type,COALESCE(?6,file_size),COALESCE(?7,file_mtime),?8,COALESCE((SELECT change_revision FROM files WHERE path=?3),file_change_revision) FROM semantic_reuse_chunks",params![generation,task.passage_id,task.path,location.map(|s|s.kind.as_str()),location.map(|s|s.number),identity.map(|i|i.bytes as i64),identity.map(|i|i.modified_ns),task.key])?; let changed=tx.execute("UPDATE semantic_tasks SET state='done',expected_chunks=?3,path=?4,passage_id=?5,message=NULL WHERE generation=?1 AND task_key=?2 AND source_hash=?6 AND title_hash=?7 AND policy=?8 AND source_type=?9",params![generation,task.key,count as i64,task.path,task.passage_id,task.source_hash,task.title_hash,task.policy,task.source_type])?; ensure!(changed == 1, "Missing source task during reuse"); if task.source_type == "media" { diff --git a/crates/searchi-service/tests/incremental_google_cpu.rs b/crates/searchi-service/tests/incremental_google_cpu.rs index fc4099a..5d621e2 100644 --- a/crates/searchi-service/tests/incremental_google_cpu.rs +++ b/crates/searchi-service/tests/incremental_google_cpu.rs @@ -321,14 +321,39 @@ fn genuine_google_mixed_media_reuse_checks_bytes_despite_identical_file_metadata .open(&image)? .set_times(fs::FileTimes::new().set_modified(metadata.modified()?))?; assert_eq!(fs::metadata(&image)?.modified()?, metadata.modified()?); - // Core intentionally sees unchanged size/mtime: semantic reuse must still - // hash actual media bytes before accepting a completion manifest. + // Full metadata reconciliation intentionally retains its media fast path. + // A native event must durably invalidate old media even with identical stats. lexical.index_folder(&root)?; + let canonical_root = fs::canonicalize(&root)?; + let image_hint = fs::canonicalize(&image)?; + lexical.index_paths_with_options( + &canonical_root, + std::slice::from_ref(&image_hint), + &Default::default(), + )?; + assert!(HybridSearch::embedding_status(&db)?.stale); + let options = SearchOptions { + extension: Some("bmp".into()), + ..Default::default() + }; + assert!( + service.search("a red square", 10, &options)?.is_empty(), + "The hinted old image must be hidden until actual byte-verified reindexing" + ); service.index_embeddings_with_media(&tools, &limits)?; let changed = calls.lock().unwrap().clone(); assert_eq!(changed.documents, initial.documents); assert_eq!(changed.audio, initial.audio); assert_eq!(changed.images, initial.images + 1); + assert!(!HybridSearch::embedding_status(&db)?.stale); + let response = service.search("a green square", 10, &options)?; + assert_eq!(response[0].filename, "color.bmp"); + let connection = Connection::open(&db)?; + let revisions: (i64,i64) = connection.query_row( + "SELECT f.change_revision,c.file_change_revision FROM files f JOIN semantic_chunks c ON c.path=f.path WHERE f.path=?1 AND c.generation=(SELECT active_generation FROM semantic_meta WHERE singleton=1)", + [image_hint.to_string_lossy().as_ref()], |row| Ok((row.get(0)?,row.get(1)?)), + )?; + assert_eq!(revisions, (1, 1)); let latest = chunks(&db)?; for chunk in first.iter().filter(|chunk| !chunk.0.ends_with("color.bmp")) { assert!( @@ -348,6 +373,27 @@ fn genuine_google_mixed_media_reuse_checks_bytes_despite_identical_file_metadata .unwrap() .6 ); + // Unchanged bytes still need fresh eligibility after a hint, but the SHA- + // verified completed source should reuse vectors and remap revision tokens. + let audio_hint = fs::canonicalize(root.join("tone.wav"))?; + lexical.index_paths_with_options( + &canonical_root, + &[image_hint.clone(), audio_hint.clone()], + &Default::default(), + )?; + assert!(service.search("a green square", 10, &options)?.is_empty()); + let reused = service.index_embeddings_with_media(&tools, &limits)?; + assert_eq!(reused.reused_sources, 5); + assert_eq!(*calls.lock().unwrap(), changed); + assert_eq!( + service.search("a green square", 10, &options)?[0].filename, + "color.bmp" + ); + let audio_revisions: (i64,i64) = connection.query_row( + "SELECT f.change_revision,c.file_change_revision FROM files f JOIN semantic_chunks c ON c.path=f.path WHERE f.path=?1 AND c.generation=(SELECT active_generation FROM semantic_meta WHERE singleton=1)", + [audio_hint.to_string_lossy().as_ref()], |row| Ok((row.get(0)?,row.get(1)?)), + )?; + assert_eq!(audio_revisions, (1, 1)); Ok(()) } diff --git a/docs/AUTOMATIC-INDEXING.md b/docs/AUTOMATIC-INDEXING.md new file mode 100644 index 0000000..aa57e1c --- /dev/null +++ b/docs/AUTOMATIC-INDEXING.md @@ -0,0 +1,89 @@ +# Automatic folder updates + +The desktop now registers saved folders at startup and reconciles them in the +background. Native notifications use pinned `notify` 8.2.0. They are hints: +[notify documents](https://docs.rs/notify/8.2.0/notify/) missed events, network +filesystem limitations and platform-specific editor behavior. Periodic scans +remain necessary; a native watch failure keeps polling and reports the issue. + +## Scheduling and bounds + +- At most 64 roots and 128 retained paths per root; overflow becomes a full + reconciliation rather than an unbounded event queue. +- A 750 ms quiet period coalesces bursts, with a five-second maximum delay. +- Startup reconciles saved folders; ordinary verification runs every five minutes. +- Busy manual indexing retains queued work. Failed scans retry with backoff from + five seconds to one minute. Events arriving during a scan remain pending. +- Read/access-time events are ignored; native rescan flags still take priority. +- Missing roots keep existing records and retry attachment. Successful + reattachment after a prior watch requests verification of missed media events. +- Callback panics become a retryable failure. Shutdown signals cancellation, + wakes the owned thread and joins it without holding its state mutex. + +Scans own separate SQLite connections. They do not hold the foreground search +mutex during traversal or extraction. A changed scan prepares its reader and +briefly swaps it into the desktop. UI events refresh counts and the active query, +reject results from before the change, and request normal semantic indexing. +The original input/first-response timing survives an automatic query refresh. + +## Targeted reconciliation + +`SearchIndex::index_paths_with_options` handles bounded absolute normalized paths. +Files update individually; directory hints reconcile only that subtree. Missing +paths remove only records within their root and path boundary. Parent/child hints +are coalesced. Outside paths, traversal and symlink ancestors are rejected. +Read/traversal failure or cancellation rolls back the complete batch. + +The cancellable folder and path variants check between entries and before commit. +An individual bounded parser input can finish before cancellation is observed. +Parser process isolation and bounded shutdown time still require further work. + +Absolute exclusions complement directory-name exclusions. The desktop excludes +its data, model and worker directories from both scans and notifications. +Decoded video frames use owned scratch directories under application data, +so temporary frames cannot feed back into the index when a home folder is watched. + +## Media freshness + +Core schema 13 adds `files.change_revision`, initially zero for existing records. +Targeted media hints increment it even when file size and timestamp are unchanged, +and remove old media tags. Full native-rescan/overflow hints take the targeted +root path; ordinary periodic scans retain the metadata fast path for media. + +Semantic schema 8 records the matching `file_change_revision` on media chunks. +An old media vector is ineligible after a hint until genuine byte-verified indexing +completes. Unchanged media can reuse completed vectors with the current revision; +changed bytes require new Google inference. Metadata probing captures the revision +before work and writes only if it still matches, preventing a stale probe from +restoring tags after a concurrent event. Generation activation remains guarded +by the original corpus revision and actual source SHA validation. + +Periodic scans hash supported bounded document inputs, but do not hash every +large media file. A same-stat media overwrite with no observed native hint remains +a verification limitation; explicit semantic rebuilds verify actual media bytes. + +## Executed validation and remaining gates + +Windows native watcher tests observed real create/update/rename/delete events. +The final Windows workspace run passed 123 tests (19 explicitly ignored), and +the desktop suite passed 28 tests. Frontend build, 39 tests, formatting and strict +workspace/desktop Clippy passed. The real CPU FFmpeg decoder regression passed +with frame paths under the configured owned scratch parent. +Coordinator tests cover overflow, rescan, busy retention, concurrent events, +periodic scheduling, panic recovery and shutdown. Core tests cover same-stat +documents/media, subtree and boundary deletion, exclusions, migrations, rollback, +concurrent readers and cancellation after a write transaction starts. + +The real Google mixed-media CPU regression passed in 83.70 seconds. It verified +that a same-stat image hint hides old vectors, changed image bytes cause one new +image encode, unchanged hinted image/audio sources reuse vectors, and current +revision tokens restore eligibility. This is correctness evidence on a tiny +fixture, not relevance quality or search/indexing throughput evidence. + +Linux and macOS native watcher execution, root disappearance/reappearance on +real removable/network storage, native UI refresh interaction, representative +large-corpus contention and power-loss behavior remain open. Full reconciliation +still holds a write transaction during extraction and can contend with semantic +writes; targeted batches reduce that work but do not establish a universal +latency bound. Resource admission currently covers neural inference, not all +filesystem, decoder, parser or ANN work. diff --git a/docs/INCREMENTAL-SEMANTIC.md b/docs/INCREMENTAL-SEMANTIC.md index 4c8ef8e..52c384d 100644 --- a/docs/INCREMENTAL-SEMANTIC.md +++ b/docs/INCREMENTAL-SEMANTIC.md @@ -1,7 +1,7 @@ # Incremental semantic indexing and recovery Status: completed-source reuse and compatible staging recovery implemented in -schema 7. Unchanged vectors are copied with current citations; the ANN graph +schema 8 (completed-source tasks were introduced in schema 7). Unchanged vectors are copied with current citations; the ANN graph is rebuilt. Incomplete source groups are discarded before retry. The previous ready generation and its reader policy remain available on failures. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index c7e47d1..5f1dae6 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -40,6 +40,12 @@ Status: implementation started; no production release yet. Windows VM tests use - [ ] Folder selection, exclusion controls, resource budgets and indexing status. - [ ] Global shortcut, opening/revealing source files, PDF page navigation. - [ ] Watchers and reconciliation across Windows, macOS and Linux. + - [x] Native notification coordinator, bounded targeted reconciliation, + periodic verification and CPU-tested Windows create/update/rename/delete. + - [x] Durable same-stat media hint invalidation, guarded metadata writes and + real Google reindex/reuse regression; owned decoder scratch avoids feedback. + - [ ] Native Linux/macOS events, actual UI refresh and representative storage + failure/contention checks; see [automatic indexing](AUTOMATIC-INDEXING.md). ## 4. Media - [x] Genuine Google image, audio and video embeddings integrated with the shared