645 lines
21 KiB
Rust
645 lines
21 KiB
Rust
//! Native notifications are bounded hints; periodic reconciliation is authoritative.
|
|
use notify::{Event, EventKind, RecursiveMode, Watcher};
|
|
use serde::Serialize;
|
|
use std::{
|
|
collections::{BTreeMap, BTreeSet},
|
|
path::{Path, PathBuf},
|
|
sync::{
|
|
atomic::{AtomicBool, Ordering},
|
|
Arc, Condvar, Mutex,
|
|
},
|
|
thread::{self, JoinHandle},
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
const MAX_ROOTS: usize = 64;
|
|
const MAX_PATHS: usize = 128;
|
|
|
|
pub struct ScanRequest {
|
|
pub root: PathBuf,
|
|
pub paths: Option<Vec<PathBuf>>,
|
|
pub cancelled: Arc<AtomicBool>,
|
|
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<String>,
|
|
pub watch_error: Option<String>,
|
|
}
|
|
#[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<PathBuf>,
|
|
full: bool,
|
|
media_hint: bool,
|
|
ever_attached: bool,
|
|
first: Option<Instant>,
|
|
last: Instant,
|
|
next_scan: Instant,
|
|
retry_at: Instant,
|
|
attach_retry: Instant,
|
|
failures: u32,
|
|
attached: bool,
|
|
error: Option<String>,
|
|
watch_error: Option<String>,
|
|
}
|
|
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<PathBuf, Root>,
|
|
stop: bool,
|
|
}
|
|
struct Shared {
|
|
state: Mutex<State>,
|
|
changed: Condvar,
|
|
ignored: Vec<PathBuf>,
|
|
timing: Timing,
|
|
cancelled: Arc<AtomicBool>,
|
|
}
|
|
/// Own this service for the application lifetime. Dropping joins its worker after
|
|
/// the currently executing reconciliation callback finishes.
|
|
pub struct WatchService {
|
|
shared: Arc<Shared>,
|
|
worker: Mutex<Option<JoinHandle<()>>>,
|
|
}
|
|
impl WatchService {
|
|
pub fn start(
|
|
roots: Vec<PathBuf>,
|
|
ignored: Vec<PathBuf>,
|
|
callback: impl Fn(ScanRequest) -> ScanOutcome + Send + 'static,
|
|
) -> Result<Self, String> {
|
|
Self::start_timed(roots, ignored, callback, Timing::default())
|
|
}
|
|
fn start_timed(
|
|
roots: Vec<PathBuf>,
|
|
ignored: Vec<PathBuf>,
|
|
callback: impl Fn(ScanRequest) -> ScanOutcome + Send + 'static,
|
|
timing: Timing,
|
|
) -> Result<Self, String> {
|
|
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<PathBuf>) -> 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<WatchStatus> {
|
|
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<Event>) {
|
|
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<Shared>, 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::<Vec<_>>()
|
|
};
|
|
for removed in watched
|
|
.iter()
|
|
.filter(|p| !roots.contains(p))
|
|
.cloned()
|
|
.collect::<Vec<_>>()
|
|
{
|
|
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>) -> 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));
|
|
}
|
|
}
|