diff --git a/src/app.rs b/src/app.rs index 77275a9..c41c4fc 100644 --- a/src/app.rs +++ b/src/app.rs @@ -25,17 +25,30 @@ use std::net::SocketAddr; use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::time::Duration; use hydra::{ Application, ApplicationConfig, ChildSpec, Dest, ExitReason, From as HydraFrom, GenServer, GenServerOptions, Pid, SupervisionStrategy, Supervisor, SupervisorOptions, }; +use notify::{RecursiveMode, Watcher}; use serde::{Deserialize, Serialize}; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; use crate::index::NoteIndex; use crate::vault::{self, Note}; +use crate::watcher::run_debouncer; -const INDEX_WRITER_NAME: &str = "index_writer"; +pub(crate) const INDEX_WRITER_NAME: &str = "index_writer"; +pub(crate) const VAULT_SCANNER_NAME: &str = "vault_scanner"; +pub(crate) const FILESYSTEM_WATCHER_NAME: &str = "filesystem_watcher"; +pub(crate) const HTTP_SERVER_NAME: &str = "http_server"; + +/// Default debounce window for the filesystem watcher. Per [[ADR-003 +/// Filesystem Change Tracking]] -- 100 ms is the documented target, tunable +/// once real save patterns are observed. +const WATCH_DEBOUNCE_WINDOW: Duration = Duration::from_millis(100); /// Snapshot of per-process restart counters. Each role increments its own /// counter on every `init` *after the first*, so the count represents @@ -137,6 +150,7 @@ impl Application for Anwesen { } .child_spec(), FilesystemWatcher { + vault: self.vault.clone(), counters: self.counters.clone(), } .child_spec(), @@ -184,12 +198,12 @@ impl VaultScanner { fn child_spec(self) -> ChildSpec { let counters = self.counters.clone(); let vault = self.vault.clone(); - ChildSpec::new("vault_scanner").start(move || { + ChildSpec::new(VAULT_SCANNER_NAME).start(move || { VaultScanner { vault: vault.clone(), counters: counters.clone(), } - .start_link(GenServerOptions::new().name("vault_scanner")) + .start_link(GenServerOptions::new().name(VAULT_SCANNER_NAME)) }) } @@ -247,38 +261,89 @@ impl GenServer for VaultScanner { } } -// -- filesystem_watcher (stub for ANW-16) ----------------------------------- +// -- filesystem_watcher ----------------------------------------------------- +/// The watcher accepts no inbound messages today -- its work is the +/// `notify` stream plus the debouncer task. The single `Noop` variant +/// satisfies Hydra's `Receivable` bound; once a real call/cast surface is +/// useful (e.g. for tests) it can be added. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum FilesystemWatcherMessage { - /// Placeholder so the enum has a variant -- replaced by real watch - /// events in [ANW-16](https://crvrs.youtrack.cloud/issue/ANW-16). Noop, } #[derive(Clone)] pub struct FilesystemWatcher { + vault: PathBuf, counters: Arc, } impl FilesystemWatcher { fn child_spec(self) -> ChildSpec { let counters = self.counters.clone(); - ChildSpec::new("filesystem_watcher").start(move || { - FilesystemWatcher { + let vault = self.vault.clone(); + ChildSpec::new(FILESYSTEM_WATCHER_NAME).start(move || { + FilesystemWatcherState { + vault: vault.clone(), counters: counters.clone(), + watcher: None, + debouncer: None, } - .start_link(GenServerOptions::new().name("filesystem_watcher")) + .start_link(GenServerOptions::new().name(FILESYSTEM_WATCHER_NAME)) }) } } -impl GenServer for FilesystemWatcher { +/// Runtime state for the watcher process. Holds the live +/// [`notify::RecommendedWatcher`] (must outlive event delivery) and the +/// [`JoinHandle`] for the debouncer task; both are torn down when the +/// process is dropped on restart. +struct FilesystemWatcherState { + vault: PathBuf, + counters: Arc, + watcher: Option, + debouncer: Option>, +} + +impl Drop for FilesystemWatcherState { + fn drop(&mut self) { + if let Some(h) = self.debouncer.take() { + h.abort(); + } + // `watcher` drops on its own, which is enough to stop the inotify + // binding; the debouncer task then sees the channel close. + } +} + +impl GenServer for FilesystemWatcherState { type Message = FilesystemWatcherMessage; async fn init(&mut self) -> Result<(), ExitReason> { let restart = self.counters.filesystem_watcher.record_init(); - tracing::info!(restart, "filesystem_watcher: init (stub; ANW-16)"); + + let (tx, rx) = mpsc::unbounded_channel::>(); + let mut watcher = notify::recommended_watcher(move |res| { + // Send is non-blocking on an unbounded channel; ignore the + // SendError that arises only when the receiver has been dropped + // (process shutdown). + let _ = tx.send(res); + }) + .map_err(|e| ExitReason::from(format!("filesystem_watcher: recommended_watcher: {e}")))?; + watcher + .watch(&self.vault, RecursiveMode::Recursive) + .map_err(|e| ExitReason::from(format!("filesystem_watcher: watch: {e}")))?; + + let handle = tokio::spawn(run_debouncer(rx, self.vault.clone(), WATCH_DEBOUNCE_WINDOW)); + + self.watcher = Some(watcher); + self.debouncer = Some(handle); + + tracing::info!( + restart, + vault = %self.vault.display(), + debounce = ?WATCH_DEBOUNCE_WINDOW, + "filesystem_watcher: init" + ); Ok(()) } @@ -304,15 +369,28 @@ pub enum IndexWriterMessage { /// Discard the current index and reindex the given notes. Sent by /// `vault_scanner` at startup and after `rescan_now`. Rebuild(Vec), - /// Insert-or-replace one note. Sent by `filesystem_watcher` - /// ([ANW-16](https://crvrs.youtrack.cloud/issue/ANW-16)) on - /// Create / Move-In / Modify / Close-Write events. + /// Apply one debounce-window's worth of upserts and deletes in a single + /// Tantivy commit. Sent by `filesystem_watcher`. Picks up sie's ANW-12 + /// follow-up #3 (commit batching) before ANW-16's watcher can turn + /// per-save events into a hot commit loop. + Batch(IndexBatch), + /// Insert-or-replace one note. Retained for direct callers; the watcher + /// uses `Batch`. Upsert(Box), - /// Drop one note from the index. Sent by `filesystem_watcher` on - /// Delete / Move-Out events. + /// Drop one note from the index. Retained for direct callers; the + /// watcher uses `Batch`. Delete(String), } +/// One batched index update. Deletes apply first so a "delete-then-upsert" +/// sequence is unambiguous; per-path coalescing in +/// [`crate::watcher::coalesce`] already keeps at most one action per path. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct IndexBatch { + pub upserts: Vec, + pub deletes: Vec, +} + #[derive(Clone)] pub struct IndexWriter { counters: Arc, @@ -334,7 +412,7 @@ impl IndexWriter { /// Runtime state for the [`IndexWriter`] process. Held in a separate struct /// so the `Clone`-friendly child-spec form (which doesn't carry the live /// `NoteIndex`) stays simple. -struct IndexWriterState { +pub(crate) struct IndexWriterState { counters: Arc, /// Created lazily in `init` so a Tantivy construction failure surfaces /// as an `ExitReason` and triggers a supervisor restart, rather than @@ -369,6 +447,14 @@ impl GenServer for IndexWriterState { } tracing::info!(notes = count, "index_writer: rebuilt"); } + IndexWriterMessage::Batch(batch) => { + let (u, d) = (batch.upserts.len(), batch.deletes.len()); + if let Err(e) = index.apply_batch(&batch.upserts, &batch.deletes) { + tracing::error!(error = %e, "index_writer: batch apply failed"); + return Err(ExitReason::from(format!("batch: {e}"))); + } + tracing::info!(upserts = u, deletes = d, "index_writer: batch applied"); + } IndexWriterMessage::Upsert(note) => { let path = note.path.clone(); if let Err(e) = index.upsert(¬e) { @@ -416,12 +502,12 @@ impl HttpServer { fn child_spec(self) -> ChildSpec { let bind = self.bind; let counters = self.counters.clone(); - ChildSpec::new("http_server").start(move || { + ChildSpec::new(HTTP_SERVER_NAME).start(move || { HttpServer { bind, counters: counters.clone(), } - .start_link(GenServerOptions::new().name("http_server")) + .start_link(GenServerOptions::new().name(HTTP_SERVER_NAME)) }) } } diff --git a/src/index.rs b/src/index.rs index 2f376c5..ed91ac1 100644 --- a/src/index.rs +++ b/src/index.rs @@ -120,6 +120,25 @@ impl NoteIndex { Ok(()) } + /// Apply a batch of upserts and deletes in one Tantivy commit. The + /// watcher's debouncer dispatches batches per debounce window so a busy + /// editor save loop does not produce one commit per event. + /// + /// # Errors + /// Returns the underlying Tantivy error if any add or the final commit + /// fails. + pub fn apply_batch(&mut self, upserts: &[Note], deletes: &[String]) -> Result<()> { + for path in deletes { + self.writer.delete_term(self.path_term(path)); + } + for note in upserts { + self.writer.delete_term(self.path_term(¬e.path)); + self.add_document(note)?; + } + self.writer.commit().context("commit batch")?; + Ok(()) + } + /// Number of indexed documents from a fresh reader. Read-only; intended /// for tests and the `/health` endpoint ([ANW-8]). /// diff --git a/src/lib.rs b/src/lib.rs index 99b04d1..7aaa5f2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,3 +7,4 @@ pub mod app; pub mod index; pub mod vault; +pub mod watcher; diff --git a/src/vault.rs b/src/vault.rs index d702edd..dcf916a 100644 --- a/src/vault.rs +++ b/src/vault.rs @@ -114,7 +114,7 @@ pub fn scan(vault_root: &Path) -> ScanResult { if !is_markdown(abs_path) { continue; } - match read_one(vault_root, abs_path) { + match scan_one(vault_root, abs_path) { Ok(note) => notes.push(note), Err(kind) => issues.push(ScanIssue { path: abs_path.to_path_buf(), @@ -134,10 +134,20 @@ fn is_dot_prefixed(name: &std::ffi::OsStr) -> bool { name.to_str().is_some_and(|s| s.starts_with('.')) } -fn read_one(vault_root: &Path, abs_path: &Path) -> Result { +/// Read a single Markdown file off disk and produce its [`Note`]. Used by +/// [`scan`] and by the filesystem watcher's per-event handler. +/// +/// # Errors +/// Returns a [`ScanIssueKind`] when the file cannot be read, the body is not +/// valid UTF-8, the path itself isn't UTF-8, or the frontmatter YAML fails to +/// parse. +pub fn scan_one(vault_root: &Path, abs_path: &Path) -> Result { let raw_bytes = std::fs::read(abs_path)?; let metadata = std::fs::metadata(abs_path)?; - let size = metadata.len(); + // Size from the bytes we actually hashed -- avoids the one-frame drift + // possible between `read` and `metadata` if the file is rewritten under + // us. ETag and size now reflect the same snapshot. + let size = raw_bytes.len() as u64; let last_modified: DateTime = metadata.modified()?.into(); let etag = format!("\"{}\"", blake3::hash(&raw_bytes).to_hex()); diff --git a/src/watcher.rs b/src/watcher.rs new file mode 100644 index 0000000..fefc56d --- /dev/null +++ b/src/watcher.rs @@ -0,0 +1,400 @@ +//! Filesystem-event mapping + debouncer for [ANW-16]. +//! +//! The [`FilesystemWatcher`](crate::app::FilesystemWatcher) `GenServer` owns a +//! [`notify::RecommendedWatcher`] over the vault root. Each native event is +//! pushed into a Tokio channel; [`run_debouncer`] drains the channel, +//! classifies events into [`WatchAction`]s, coalesces a 100 ms window's +//! worth into one [`Batch`], and casts the batch to the `index_writer` +//! named process for a single Tantivy commit (folds in sie's ANW-12 +//! follow-up #3). +//! +//! See [[ADR-003 Filesystem Change Tracking]] for the event model. + +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +use hydra::{Dest, GenServer}; +use notify::Event; +use notify::EventKind; +use notify::event::{AccessKind, AccessMode, Flag, ModifyKind, RenameMode}; +use tokio::sync::mpsc::UnboundedReceiver; + +use crate::app::{INDEX_WRITER_NAME, IndexBatch, IndexWriterMessage, IndexWriterState}; +use crate::app::{VAULT_SCANNER_NAME, VaultScanner, VaultScannerMessage}; +use crate::vault; + +/// One path-scoped action derived from a native filesystem event. Always +/// carries a vault-relative, forward-slash-normalized path string -- the +/// same form [`vault::Note.path`] uses. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WatchAction { + Upsert(String), + Delete(String), +} + +/// One debouncer window's worth of coalesced changes. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct WatchBatch { + pub upserts: Vec, + pub deletes: Vec, +} + +/// Classify one [`Event`] into zero or more [`WatchAction`]s. Dot-segments +/// anywhere in the path and non-`.md` files are dropped here so the +/// downstream batch is already filtered. +#[must_use] +pub fn map_event(event: &Event, vault_root: &Path) -> Vec { + if is_overflow(event) { + // Overflow is a vault-wide signal, not a per-path action -- the + // caller dispatches `rescan_now` to `vault_scanner` instead. + return Vec::new(); + } + let mut actions = Vec::new(); + let action_kind = classify(event.kind); + match action_kind { + Some(EventAction::Upsert) => { + for p in &event.paths { + if let Some(rel) = vault_relative(vault_root, p) { + actions.push(WatchAction::Upsert(rel)); + } + } + } + Some(EventAction::Delete) => { + for p in &event.paths { + if let Some(rel) = vault_relative(vault_root, p) { + actions.push(WatchAction::Delete(rel)); + } + } + } + Some(EventAction::Rename) => { + // Notify packs (from, to) in event.paths in that order. + if let Some(from) = event.paths.first() + && let Some(rel) = vault_relative(vault_root, from) + { + actions.push(WatchAction::Delete(rel)); + } + if let Some(to) = event.paths.get(1) + && let Some(rel) = vault_relative(vault_root, to) + { + actions.push(WatchAction::Upsert(rel)); + } + } + None => {} + } + actions +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum EventAction { + Upsert, + Delete, + Rename, +} + +fn classify(kind: EventKind) -> Option { + match kind { + EventKind::Create(_) + | EventKind::Modify( + ModifyKind::Data(_) + | ModifyKind::Metadata(_) + | ModifyKind::Any + | ModifyKind::Name(RenameMode::To), + ) + | EventKind::Access(AccessKind::Close(AccessMode::Write)) => Some(EventAction::Upsert), + EventKind::Modify(ModifyKind::Name(RenameMode::From)) | EventKind::Remove(_) => { + Some(EventAction::Delete) + } + EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => Some(EventAction::Rename), + // Access(non-close), Modify(Name(Any|Other)), Other, Any -> drop. + _ => None, + } +} + +/// True if the event signals an inotify queue overflow ([[ADR-003 Filesystem +/// Change Tracking]] / notify's [`Flag::Rescan`]). +#[must_use] +pub fn is_overflow(event: &Event) -> bool { + event.flag() == Some(Flag::Rescan) +} + +/// Collapse a sequence of [`WatchAction`]s into a single batch. The last +/// action per path wins (delete-then-upsert ends up as upsert, and so on). +/// The batch keeps deterministic order by sorting paths inside each list. +#[must_use] +pub fn coalesce(actions: impl IntoIterator) -> WatchBatch { + #[derive(Clone, Copy)] + enum Last { + Upsert, + Delete, + } + let mut state: BTreeMap = BTreeMap::new(); + for a in actions { + match a { + WatchAction::Upsert(p) => { + state.insert(p, Last::Upsert); + } + WatchAction::Delete(p) => { + state.insert(p, Last::Delete); + } + } + } + let mut upserts = Vec::new(); + let mut deletes = Vec::new(); + for (path, last) in state { + match last { + Last::Upsert => upserts.push(path), + Last::Delete => deletes.push(path), + } + } + WatchBatch { upserts, deletes } +} + +/// Filter and normalize an absolute notify path. Returns the vault-relative +/// forward-slash path if the entry is a `.md` file outside any dot-directory; +/// returns `None` otherwise (so the caller drops the event). +fn vault_relative(vault_root: &Path, abs: &Path) -> Option { + let rel = abs.strip_prefix(vault_root).ok()?; + if rel.extension().is_none_or(|e| e != "md") { + return None; + } + for component in rel.components() { + let s = component.as_os_str().to_str()?; + if s.starts_with('.') { + return None; + } + } + Some(rel.to_str()?.replace('\\', "/")) +} + +/// Resolve a vault-relative path back to an absolute one for re-reading. +#[must_use] +pub fn absolute_path(vault_root: &Path, rel: &str) -> PathBuf { + vault_root.join(rel) +} + +/// Drain notify events from `rx`, coalesce in `window`-length batches, and +/// dispatch each batch to the `index_writer` and (on overflow) a +/// `rescan_now` cast to `vault_scanner`. +/// +/// Loops until the channel closes -- that only happens when the watcher +/// process is being shut down and its sender side is dropped. +pub async fn run_debouncer( + mut rx: UnboundedReceiver>, + vault_root: PathBuf, + window: Duration, +) { + loop { + let Some(first) = rx.recv().await else { + break; + }; + let mut events = vec![first]; + let deadline = Instant::now() + window; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + match tokio::time::timeout(remaining, rx.recv()).await { + Ok(Some(e)) => events.push(e), + Ok(None) | Err(_) => break, + } + } + let mut overflow = false; + let mut actions = Vec::new(); + for entry in events { + match entry { + Ok(ev) => { + if is_overflow(&ev) { + overflow = true; + } + actions.extend(map_event(&ev, &vault_root)); + } + Err(err) => { + tracing::warn!(error = %err, "filesystem_watcher: notify error"); + } + } + } + if overflow { + tracing::warn!("filesystem_watcher: overflow -- dispatching rescan_now"); + VaultScanner::cast( + Dest::from(VAULT_SCANNER_NAME), + VaultScannerMessage::RescanNow, + ); + } + let batch = coalesce(actions); + if batch.upserts.is_empty() && batch.deletes.is_empty() { + continue; + } + let index_batch = build_index_batch(&vault_root, batch); + IndexWriterState::cast( + Dest::from(INDEX_WRITER_NAME), + IndexWriterMessage::Batch(index_batch), + ); + } +} + +fn build_index_batch(vault_root: &Path, batch: WatchBatch) -> IndexBatch { + let mut upserts = Vec::with_capacity(batch.upserts.len()); + for rel in batch.upserts { + let abs = absolute_path(vault_root, &rel); + match vault::scan_one(vault_root, &abs) { + Ok(note) => upserts.push(note), + Err(kind) => { + tracing::warn!( + path = %abs.display(), + error = %kind, + "filesystem_watcher: re-read failed; skipping upsert" + ); + } + } + } + IndexBatch { + upserts, + deletes: batch.deletes, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use notify::event::{CreateKind, RemoveKind}; + + fn vault() -> PathBuf { + PathBuf::from("/v") + } + + fn ev(kind: EventKind, paths: &[&str]) -> Event { + let mut e = Event::new(kind); + for p in paths { + e = e.add_path(PathBuf::from(p)); + } + e + } + + #[test] + fn create_md_under_vault_is_upsert() { + let e = ev(EventKind::Create(CreateKind::File), &["/v/a.md"]); + assert_eq!( + map_event(&e, &vault()), + vec![WatchAction::Upsert("a.md".into())] + ); + } + + #[test] + fn non_md_file_is_dropped() { + let e = ev(EventKind::Create(CreateKind::File), &["/v/a.txt"]); + assert!(map_event(&e, &vault()).is_empty()); + } + + #[test] + fn dot_directory_descendants_are_dropped() { + let e = ev(EventKind::Create(CreateKind::File), &["/v/.obsidian/x.md"]); + assert!(map_event(&e, &vault()).is_empty()); + let e = ev( + EventKind::Create(CreateKind::File), + &["/v/sub/.hidden/y.md"], + ); + assert!(map_event(&e, &vault()).is_empty()); + } + + #[test] + fn path_outside_vault_is_dropped() { + let e = ev(EventKind::Create(CreateKind::File), &["/elsewhere/a.md"]); + assert!(map_event(&e, &vault()).is_empty()); + } + + #[test] + fn modify_data_becomes_upsert() { + let e = ev( + EventKind::Modify(ModifyKind::Data(notify::event::DataChange::Content)), + &["/v/a.md"], + ); + assert_eq!( + map_event(&e, &vault()), + vec![WatchAction::Upsert("a.md".into())] + ); + } + + #[test] + fn close_write_becomes_upsert() { + let e = ev( + EventKind::Access(AccessKind::Close(AccessMode::Write)), + &["/v/a.md"], + ); + assert_eq!( + map_event(&e, &vault()), + vec![WatchAction::Upsert("a.md".into())] + ); + } + + #[test] + fn rename_from_is_delete() { + let e = ev( + EventKind::Modify(ModifyKind::Name(RenameMode::From)), + &["/v/a.md"], + ); + assert_eq!( + map_event(&e, &vault()), + vec![WatchAction::Delete("a.md".into())] + ); + } + + #[test] + fn rename_to_is_upsert() { + let e = ev( + EventKind::Modify(ModifyKind::Name(RenameMode::To)), + &["/v/b.md"], + ); + assert_eq!( + map_event(&e, &vault()), + vec![WatchAction::Upsert("b.md".into())] + ); + } + + #[test] + fn rename_both_emits_delete_then_upsert() { + let e = ev( + EventKind::Modify(ModifyKind::Name(RenameMode::Both)), + &["/v/a.md", "/v/b.md"], + ); + assert_eq!( + map_event(&e, &vault()), + vec![ + WatchAction::Delete("a.md".into()), + WatchAction::Upsert("b.md".into()) + ] + ); + } + + #[test] + fn remove_is_delete() { + let e = ev(EventKind::Remove(RemoveKind::File), &["/v/a.md"]); + assert_eq!( + map_event(&e, &vault()), + vec![WatchAction::Delete("a.md".into())] + ); + } + + #[test] + fn coalesce_last_action_per_path_wins() { + let actions = vec![ + WatchAction::Upsert("a.md".into()), + WatchAction::Upsert("a.md".into()), + WatchAction::Delete("b.md".into()), + WatchAction::Upsert("b.md".into()), + WatchAction::Delete("c.md".into()), + ]; + let batch = coalesce(actions); + assert_eq!(batch.upserts, vec!["a.md".to_string(), "b.md".to_string()]); + assert_eq!(batch.deletes, vec!["c.md".to_string()]); + } + + #[test] + fn overflow_event_returns_empty_actions() { + // Caller dispatches rescan_now; map_event itself produces no per-path actions. + let e = Event::new(EventKind::Other).set_flag(Flag::Rescan); + assert!(is_overflow(&e)); + assert!(map_event(&e, &vault()).is_empty()); + } +}