use super::paths::{resolve, tree}; use super::*; use std::collections::HashSet; use tokio::sync::Notify; const MAX_ROOTS: usize = 128; const MAX_QUEUED_EVENTS: usize = 256; const MAX_QUEUED_PATHS: usize = 4096; const CREATE: u8 = 0; const MODIFY: u8 = 1; const REMOVE: u8 = 2; const RENAME: u8 = 3; const RESCAN: u8 = 4; type WatchEvent = (u8, Vec); #[derive(Clone)] struct Root { target: PathTarget, path: PathBuf, directory: bool, recursive: bool, // A file root is watched through its parent directory, so replacing the // file (the usual atomic save) or recreating it stays observable at the // same path instead of following the old inode. watch_path: PathBuf, } #[derive(Clone, Debug, PartialEq, Eq)] struct QueuedEvent { kind: u8, paths: Vec, ends: Vec, } #[derive(Default)] struct WatchState { queue: VecDeque, closed: bool, terminal: Option<(io::ErrorKind, String)>, lossy: bool, live: Vec, // Root endings discarded with an overflowed queue, applied when the // replacement rescan is delivered instead. pending_ends: Vec, epoch: u32, } struct Shared { caps: Arc, roots: Vec, state: Mutex, notify: Notify, } struct WatchResource { shared: Arc, #[cfg_attr( not(any( target_os = "linux", target_os = "android", target_os = "macos", target_os = "windows" )), allow(dead_code) )] native: Mutex>, } impl WatchResource { fn close(&self) { { let mut state = self.shared.state.lock().unwrap(); state.closed = true; state.queue.clear(); state.lossy = false; state.epoch = state.epoch.wrapping_add(1); } self.shared.notify.notify_waiters(); let native = self.native.lock().unwrap().take(); drop(native); self.shared.notify.notify_waiters(); } } struct Watchers { caps: Arc, resources: Mutex>>, next: AtomicU64, } impl Drop for Watchers { fn drop(&mut self) { for resource in self .resources .get_mut() .unwrap() .drain() .map(|(_, resource)| resource) { resource.close(); } } } impl Watchers { fn resource(&self, id: u64) -> io::Result> { self.resources .lock() .unwrap() .get(&id) .cloned() .ok_or_else(bad_resource) } fn create(&self, labels: Vec, recursive: bool) -> io::Result { if labels.is_empty() || labels.len() > MAX_ROOTS { return Err(io::Error::new( io::ErrorKind::InvalidInput, "Expected between one and 128 filesystem watch paths", )); } let roots = roots(&self.caps, labels, recursive)?; let live = vec![true; roots.len()]; let shared = Arc::new(Shared { caps: self.caps.clone(), roots, state: Mutex::new(WatchState { live, ..WatchState::default() }), notify: Notify::new(), }); let native = register(&shared)?; check_roots_unchanged(&shared)?; let id = self .next .fetch_update(Ordering::AcqRel, Ordering::Acquire, |id| { (id <= super::MAX_ID).then_some(id + 1) }) .map_err(|_| io::Error::other("Filesystem watcher identifiers exhausted"))?; self.resources.lock().unwrap().insert( id, Arc::new(WatchResource { shared, native: Mutex::new(Some(native)), }), ); Ok(id) } fn close(&self, id: u64, quiet: bool) -> io::Result { let resource = self.resources.lock().unwrap().remove(&id); match resource { Some(resource) => { resource.close(); Ok(true) } None if quiet => Ok(true), None => Err(bad_resource()), } } } fn roots(caps: &Capabilities, labels: Vec, recursive: bool) -> io::Result> { let mut roots = Vec::new(); let mut seen = HashSet::new(); for label in labels { let target = resolve(caps, &label, false, false)?; let path = target.as_path().to_owned(); if !seen.insert(path.clone()) { continue; } let (directory, name) = parent(&path)?; let metadata = directory.symlink_metadata(&name)?; if metadata.is_symlink() { return Err(io::Error::new( io::ErrorKind::WouldBlock, "Filesystem watch root changed while it was registered", )); } let (directory, watch_path) = if metadata.is_dir() { (true, path.clone()) } else if metadata.is_file() { let parent = path.parent().ok_or_else(|| { io::Error::new(io::ErrorKind::InvalidInput, "Watched file has no parent") })?; (false, parent.to_owned()) } else { return Err(io::Error::new( io::ErrorKind::Unsupported, "Only regular files and directories can be watched", )); }; if directory && recursive { tree(caps, &target, &label, false)?; } roots.push(Root { target, path, directory, recursive: directory && recursive, watch_path, }); } Ok(roots) } fn check_roots_unchanged(shared: &Shared) -> io::Result<()> { for root in &shared.roots { let current = PathTarget::for_creation(&root.path).map_err(|_| { io::Error::new( io::ErrorKind::WouldBlock, "Filesystem watch root changed while it was registered", ) })?; if current.as_path() != root.path { return Err(io::Error::new( io::ErrorKind::WouldBlock, "Filesystem watch root changed while it was registered", )); } let (directory, name) = parent(&root.path)?; let metadata = directory.symlink_metadata(&name)?; if metadata.is_symlink() || metadata.is_dir() != root.directory { return Err(io::Error::new( io::ErrorKind::WouldBlock, "Filesystem watch root changed while it was registered", )); } } Ok(()) } #[cfg(any( target_os = "linux", target_os = "android", target_os = "macos", target_os = "windows" ))] fn register(shared: &Arc) -> io::Result { use notify::Watcher; let weak = Arc::downgrade(shared); let handler = move |result: notify::Result| { if let Some(shared) = weak.upgrade() { shared.handle(result); } }; let config = notify::Config::default().with_follow_symlinks(false); let mut watcher = ::new(handler, config).map_err(watch_error)?; // Roots can share a registration path (a watched file and its parent // directory); the recursive registration must win when both are present. let mut registrations: HashMap = HashMap::new(); for root in &shared.roots { let recursive = root.directory && root.recursive; registrations .entry(root.watch_path.clone()) .and_modify(|existing| *existing |= recursive) .or_insert(recursive); } for (path, recursive) in registrations { let mode = if recursive { notify::RecursiveMode::Recursive } else { notify::RecursiveMode::NonRecursive }; watcher.watch(&path, mode).map_err(watch_error)?; } Ok(watcher) } #[cfg(not(any( target_os = "linux", target_os = "android", target_os = "macos", target_os = "windows" )))] fn register(_shared: &Arc) -> io::Result { Err(io::Error::new( io::ErrorKind::Unsupported, "Filesystem watching is not supported on this platform", )) } fn watch_error(error: notify::Error) -> io::Error { let message = error.to_string(); match error.kind { notify::ErrorKind::PathNotFound | notify::ErrorKind::WatchNotFound => { io::Error::new(io::ErrorKind::NotFound, message) } notify::ErrorKind::Io(inner) => io::Error::new(inner.kind(), message), notify::ErrorKind::InvalidConfig(_) => io::Error::new(io::ErrorKind::InvalidInput, message), notify::ErrorKind::MaxFilesWatch => io::Error::new(io::ErrorKind::WouldBlock, message), notify::ErrorKind::Generic(_) => io::Error::other(message), } } impl Shared { fn handle(self: &Arc, result: notify::Result) { match result { Ok(event) => { if event.need_rescan() { self.enqueue(QueuedEvent { kind: RESCAN, paths: self.event_roots(&event.paths), ends: Vec::new(), }); return; } let Some(kind) = map_kind(&event.kind, event.paths.len()) else { return; }; if event.paths.is_empty() { return; } let mut paths = Vec::new(); let mut unknown = false; for path in event.paths { if self.root_index(&path).is_some() || kind == RENAME { paths.push(path); } else if self.in_scope(&path) { // The backend also sees siblings of a watched file and, // on some platforms, descendants of a non-recursive // directory. Those are in scope but not reportable. continue; } else { unknown = true; } } if unknown && kind != RENAME { self.enqueue(QueuedEvent { kind: RESCAN, paths: self.roots.iter().map(|root| root.path.clone()).collect(), ends: Vec::new(), }); return; } if paths.is_empty() { return; } let ends = self.event_ends(kind, &paths); self.enqueue(QueuedEvent { kind, paths, ends }); } Err(error) => { let error = watch_error(error); let mut state = self.state.lock().unwrap(); if !state.closed && state.terminal.is_none() { state.terminal = Some((error.kind(), error.to_string())); drop(state); self.notify.notify_waiters(); } } } } fn event_roots(&self, paths: &[PathBuf]) -> Vec { let mut roots = Vec::new(); for path in paths { if let Some(index) = self.root_index(path) { let root = &self.roots[index]; if !roots.contains(&root.path) { roots.push(root.path.clone()); } } } if roots.is_empty() { roots = self.roots.iter().map(|root| root.path.clone()).collect(); } roots } fn event_ends(&self, kind: u8, paths: &[PathBuf]) -> Vec { let mut ends = Vec::new(); if kind != REMOVE && kind != RENAME { return ends; } for (index, root) in self.roots.iter().enumerate() { if !root.directory { // A watched file is anchored at its path, not its inode, and // may be removed and recreated without ending the watch. continue; } let ends_at_root = if kind == REMOVE { paths.iter().any(|path| path == &root.path) } else { paths.first() == Some(&root.path) }; if ends_at_root { ends.push(index); } } ends } fn root_index(&self, path: &Path) -> Option { self.roots .iter() .enumerate() .filter(|(_, root)| { if !root.directory { return path == root.path; } if path == root.path { return true; } let Ok(relative) = path.strip_prefix(&root.path) else { return false; }; root.recursive || relative.components().count() == 1 }) .max_by_key(|(_, root)| root.path.components().count()) .map(|(index, _)| index) } fn in_scope(&self, path: &Path) -> bool { self.roots .iter() .any(|root| path.starts_with(&root.watch_path)) } fn enqueue(&self, event: QueuedEvent) { let mut state = self.state.lock().unwrap(); if state.closed { return; } if state.lossy { state.pending_ends.extend(event.ends.iter().copied()); return; } if event.kind == RESCAN { if let Some(existing) = state.queue.iter_mut().find(|queued| queued.kind == RESCAN) { for path in event.paths { if !existing.paths.contains(&path) { existing.paths.push(path); } } drop(state); self.notify.notify_waiters(); return; } } else if event.kind != RENAME && state.queue.back().is_some_and(|last| { last.kind == event.kind && last.ends.is_empty() && event.ends.is_empty() }) { let queued_paths = state .queue .iter() .map(|queued| queued.paths.len()) .sum::(); let new_paths = { let last = state.queue.back().unwrap(); event .paths .iter() .filter(|path| !last.paths.contains(path)) .count() }; if queued_paths + new_paths <= MAX_QUEUED_PATHS { let last = state.queue.back_mut().unwrap(); for path in event.paths { if !last.paths.contains(&path) { last.paths.push(path); } } drop(state); self.notify.notify_waiters(); return; } } let queued_paths = state .queue .iter() .map(|queued| queued.paths.len()) .sum::(); if state.queue.len() >= MAX_QUEUED_EVENTS || queued_paths + event.paths.len() > MAX_QUEUED_PATHS { let paths = self .roots .iter() .enumerate() .filter(|(index, _)| state.live.get(*index).copied().unwrap_or(false)) .map(|(_, root)| root.path.clone()) .collect::>(); let mut discarded: Vec = state .queue .drain(..) .flat_map(|queued| queued.ends) .collect(); discarded.extend(event.ends.iter().copied()); state.pending_ends.extend(discarded); state.queue.push_back(QueuedEvent { kind: RESCAN, paths, ends: Vec::new(), }); state.lossy = true; } else { state.queue.push_back(event); } drop(state); self.notify.notify_waiters(); } async fn next(&self) -> io::Result> { loop { let notified = self.notify.notified(); tokio::pin!(notified); notified.as_mut().enable(); let candidate = { let mut state = self.state.lock().unwrap(); if state.closed { return Ok(None); } if let Some(mut event) = state.queue.pop_front() { if event.kind == RESCAN { state.lossy = false; event.ends.append(&mut state.pending_ends); } Some(event) } else if let Some((kind, message)) = state.terminal.take() { state.closed = true; return Err(io::Error::new(kind, message)); } else if state.live.iter().all(|live| !*live) { state.closed = true; return Ok(None); } else { None } }; if let Some(candidate) = candidate { let ends = candidate.ends.clone(); let event = self.authorize(candidate); { let mut state = self.state.lock().unwrap(); for index in ends { if let Some(live) = state.live.get_mut(index) { *live = false; } } } if let Some(event) = event { return Ok(Some(event)); } if !self.any_root_authorized() { let mut state = self.state.lock().unwrap(); state.closed = true; return Err(denied( &self .roots .first() .map(|root| root.path.display().to_string()) .unwrap_or_default(), false, )); } continue; } notified.await; } } fn authorize(&self, event: QueuedEvent) -> Option { if event.kind == RESCAN { let mut paths = Vec::new(); for path in event.paths { let Some(index) = self.root_index(&path) else { continue; }; if !self.state.lock().unwrap().live[index] { continue; } let root = &self.roots[index]; if self.caps.check_read(&root.target).is_err() { continue; } if let Ok(path) = path.into_os_string().into_string() && !paths.contains(&path) { paths.push(path); } } return (!paths.is_empty()).then_some((RESCAN, paths)); } let mut paths = Vec::new(); for path in &event.paths { if !self.path_authorized(path) { continue; } match path.clone().into_os_string().into_string() { Ok(path) => { if !paths.contains(&path) { paths.push(path); } } Err(_) => return self.rescan_for(event.paths.first()), } } if paths.is_empty() { return None; } let kind = if event.kind == RENAME && paths.len() == 1 { let retained = &paths[0]; if event.paths.first().and_then(|from| from.to_str()) == Some(retained.as_str()) { REMOVE } else { CREATE } } else { event.kind }; Some((kind, paths)) } fn rescan_for(&self, _path: Option<&PathBuf>) -> Option { let mut paths = Vec::new(); for root in &self.roots { if self.caps.check_read(&root.target).is_err() { continue; } if let Ok(path) = root.path.clone().into_os_string().into_string() { paths.push(path); } } (!paths.is_empty()).then_some((RESCAN, paths)) } fn path_authorized(&self, path: &Path) -> bool { let Some(index) = self.root_index(path) else { return false; }; let root = &self.roots[index]; if !self.state.lock().unwrap().live[index] { return false; } let mut target = root.target.clone(); if path != root.path { let Ok(relative) = path.strip_prefix(&root.path) else { return false; }; for component in relative.components() { let Component::Normal(name) = component else { return false; }; let Ok(child) = target.child(name) else { return false; }; target = child; } } self.caps.check_read(&target).is_ok() } fn any_root_authorized(&self) -> bool { let state = self.state.lock().unwrap(); self.roots.iter().enumerate().any(|(index, root)| { state.live.get(index).copied().unwrap_or(false) && self.caps.check_read(&root.target).is_ok() }) } async fn wait_epoch(&self, epoch: u32) -> bool { loop { let notified = self.notify.notified(); tokio::pin!(notified); notified.as_mut().enable(); { let state = self.state.lock().unwrap(); if state.closed || state.epoch != epoch { return false; } } notified.await; } } fn bump_epoch(&self) -> u32 { let epoch = { let mut state = self.state.lock().unwrap(); state.epoch = state.epoch.wrapping_add(1); state.epoch }; self.notify.notify_waiters(); epoch } } fn map_kind(kind: ¬ify::EventKind, path_count: usize) -> Option { use notify::event::{EventKind, ModifyKind, RenameMode}; match kind { EventKind::Access(_) => None, EventKind::Create(_) => Some(CREATE), EventKind::Modify(ModifyKind::Name(RenameMode::Both)) if path_count == 2 => Some(RENAME), EventKind::Modify(ModifyKind::Name(RenameMode::From)) => Some(REMOVE), EventKind::Modify(ModifyKind::Name(RenameMode::To)) => Some(CREATE), EventKind::Modify(_) => Some(MODIFY), EventKind::Remove(_) => Some(REMOVE), EventKind::Any | EventKind::Other => Some(MODIFY), } } pub(super) fn ops(files: Arc) -> Vec { let watchers = Arc::new(Watchers { caps: files.caps.clone(), resources: Mutex::new(HashMap::new()), next: AtomicU64::new(1), }); let create = watchers.clone(); let next = watchers.clone(); let next_unref = watchers.clone(); let keepalive = watchers.clone(); let reference = watchers.clone(); let close = watchers.clone(); vec![ Op::new( "op_file_watch", move |(paths, recursive): (Vec, bool)| { Ok::<_, Infallible>(response(create.create(paths, recursive))) }, ), Op::asynchronous("op_file_watch_next", move |id: u64| { let resource = next.resource(id); async move { Ok::<_, Infallible>(receive_watch(resource).await) } }), Op::asynchronous_unreferenced("op_file_watch_next_unref", move |id: u64| { let resource = next_unref.resource(id); async move { Ok::<_, Infallible>(receive_watch(resource).await) } }), Op::asynchronous("op_file_watch_keepalive", move |(id, epoch): (u64, u32)| { let resource = keepalive.resource(id); async move { let result = match resource { Ok(resource) => Ok(resource.shared.wait_epoch(epoch).await), Err(error) => Err(error), }; Ok::<_, Infallible>(response(result)) } }), Op::new("op_file_watch_ref", move |id: u64| { let result = reference .resource(id) .map(|resource| resource.shared.bump_epoch()); Ok::<_, Infallible>(response(result)) }), Op::new("op_file_watch_close", move |(id, quiet): (u64, bool)| { Ok::<_, Infallible>(response(close.close(id, quiet))) }), ] } async fn receive_watch(resource: io::Result>) -> Response> { match resource { Ok(resource) => response(resource.shared.next().await), Err(error) => response(Err(error)), } } #[cfg(test)] mod tests { use super::*; #[test] fn maps_notify_kinds_to_public_kinds() { use notify::event::{CreateKind, EventKind, ModifyKind, RemoveKind, RenameMode}; assert_eq!( map_kind(&EventKind::Create(CreateKind::File), 1), Some(CREATE) ); assert_eq!( map_kind(&EventKind::Remove(RemoveKind::File), 1), Some(REMOVE) ); assert_eq!( map_kind(&EventKind::Modify(ModifyKind::Any), 1), Some(MODIFY) ); assert_eq!( map_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::Both)), 2), Some(RENAME) ); assert_eq!( map_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::Both)), 1), Some(MODIFY) ); assert_eq!( map_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::From)), 1), Some(REMOVE) ); assert_eq!( map_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::To)), 1), Some(CREATE) ); } fn shared(root: &Path) -> Arc { Arc::new(Shared { caps: Capabilities::new(avril_caps::Initial::new(false), root.to_path_buf()), roots: vec![Root { target: PathTarget::resolve(root).unwrap(), path: root.to_path_buf(), directory: true, recursive: true, watch_path: root.to_path_buf(), }], state: Mutex::new(WatchState { live: vec![true], ..WatchState::default() }), notify: Notify::new(), }) } #[test] fn coalesces_adjacent_events_without_reordering_kinds() { let directory = tempfile::tempdir().unwrap(); let root = directory.path(); let shared = shared(root); shared.enqueue(QueuedEvent { kind: MODIFY, paths: vec![root.join("a")], ends: Vec::new(), }); shared.enqueue(QueuedEvent { kind: MODIFY, paths: vec![root.join("b"), root.join("a")], ends: Vec::new(), }); shared.enqueue(QueuedEvent { kind: CREATE, paths: vec![root.join("c")], ends: Vec::new(), }); let state = shared.state.lock().unwrap(); assert_eq!(state.queue.len(), 2); assert_eq!(state.queue[0].paths, [root.join("a"), root.join("b")]); assert_eq!(state.queue[1].kind, CREATE); } #[test] fn queue_overflow_collapses_to_one_rescan() { let directory = tempfile::tempdir().unwrap(); let root = directory.path(); let shared = shared(root); for index in 0..=MAX_QUEUED_PATHS { shared.enqueue(QueuedEvent { kind: MODIFY, paths: vec![root.join(index.to_string())], ends: Vec::new(), }); } { let state = shared.state.lock().unwrap(); assert!(state.lossy); assert_eq!(state.queue.len(), 1); assert_eq!(state.queue[0].kind, RESCAN); assert_eq!(state.queue[0].paths, [root.to_path_buf()]); } shared.enqueue(QueuedEvent { kind: CREATE, paths: vec![root.join("late")], ends: Vec::new(), }); assert_eq!(shared.state.lock().unwrap().queue.len(), 1); } #[test] fn overflow_preserves_pending_root_ends() { let directory = tempfile::tempdir().unwrap(); let root = directory.path(); let shared = shared(root); shared.enqueue(QueuedEvent { kind: REMOVE, paths: vec![root.to_path_buf()], ends: vec![0], }); for index in 0..=MAX_QUEUED_PATHS { shared.enqueue(QueuedEvent { kind: MODIFY, paths: vec![root.join(index.to_string())], ends: Vec::new(), }); } let state = shared.state.lock().unwrap(); assert_eq!(state.queue[0].kind, RESCAN); assert_eq!(state.pending_ends, [0]); } #[test] fn rename_with_an_unauthorized_destination_becomes_remove() { let directory = tempfile::tempdir().unwrap(); let root = directory.path(); let shared = shared(root); let secret = root.join("secret"); shared .caps .pledge(vec![( avril_caps::Category::Read, None, vec![secret.display().to_string()], )]) .unwrap(); let event = shared.authorize(QueuedEvent { kind: RENAME, paths: vec![root.join("old"), secret.join("new")], ends: Vec::new(), }); assert_eq!( event, Some((REMOVE, vec![root.join("old").display().to_string()])) ); } }