use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::path::{Path, PathBuf};
use std::time::SystemTime;
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct FsDelta {
pub changed: Vec<PathBuf>,
pub removed: Vec<PathBuf>,
}
impl FsDelta {
pub fn merge(&mut self, next: FsDelta) {
for path in next.changed {
self.removed.retain(|existing| existing != &path);
if !self.changed.contains(&path) {
self.changed.push(path);
}
}
for path in next.removed {
self.changed.retain(|existing| existing != &path);
if !self.removed.contains(&path) {
self.removed.push(path);
}
}
}
}
#[derive(Debug, Default)]
pub struct DeltaCoalescer {
pending: Option<FsDelta>,
}
impl DeltaCoalescer {
pub fn observe(&mut self, polled: Option<FsDelta>) -> Option<FsDelta> {
match polled {
Some(delta) => {
match &mut self.pending {
Some(pending) => pending.merge(delta),
None => self.pending = Some(delta),
}
None
}
None => self.pending.take(),
}
}
pub fn retry(&mut self, delta: FsDelta) {
match &mut self.pending {
Some(pending) => pending.merge(delta),
None => self.pending = Some(delta),
}
}
}
#[derive(Debug)]
pub struct FsWatcher {
root: PathBuf,
entries: HashMap<PathBuf, u64>,
spare: HashMap<PathBuf, u64>,
}
impl FsWatcher {
pub fn new(root: impl Into<PathBuf>) -> Self {
let root = root.into();
let mut entries = HashMap::new();
let mut rel = PathBuf::new();
collect(&root, &mut rel, &mut |path, signature| {
entries.insert(path.to_path_buf(), signature);
});
Self {
root,
entries,
spare: HashMap::new(),
}
}
pub fn poll(&mut self) -> Option<FsDelta> {
let mut next = std::mem::take(&mut self.spare);
next.clear();
next.reserve(self.entries.len());
let mut changed: Vec<PathBuf> = Vec::new();
{
let entries = &mut self.entries;
let mut rel = PathBuf::new();
collect(
&self.root,
&mut rel,
&mut |path, signature| match entries.remove_entry(path) {
Some((key, old)) if old == signature => {
next.insert(key, signature);
}
Some((key, _)) => {
changed.push(key.clone());
next.insert(key, signature);
}
None => {
let key = path.to_path_buf();
changed.push(key.clone());
next.insert(key, signature);
}
},
);
}
let mut removed: Vec<PathBuf> = self.entries.drain().map(|(path, _)| path).collect();
self.spare = std::mem::replace(&mut self.entries, next);
if changed.is_empty() && removed.is_empty() {
return None;
}
changed.sort();
removed.sort();
Some(FsDelta { changed, removed })
}
}
fn signature(meta: &std::fs::Metadata) -> u64 {
let mtime = meta
.modified()
.ok()
.and_then(|t| t.duration_since(SystemTime::UNIX_EPOCH).ok())
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
let mut hasher = DefaultHasher::new();
meta.is_dir().hash(&mut hasher);
meta.len().hash(&mut hasher);
mtime.hash(&mut hasher);
hasher.finish()
}
fn collect<F: FnMut(&Path, u64)>(dir: &Path, rel: &mut PathBuf, visit: &mut F) {
let Ok(read) = std::fs::read_dir(dir) else {
return;
};
for entry in read.flatten() {
let Ok(meta) = entry.metadata() else {
continue;
};
rel.push(entry.file_name());
visit(rel.as_path(), signature(&meta));
if meta.is_dir() {
collect(&entry.path(), rel, visit);
}
rel.pop();
}
}
#[cfg(test)]
mod tests {
#[allow(clippy::wildcard_imports)]
use super::*;
use std::fs;
use tempfile::TempDir;
fn delta(changed: &[&str], removed: &[&str]) -> FsDelta {
FsDelta {
changed: changed.iter().map(PathBuf::from).collect(),
removed: removed.iter().map(PathBuf::from).collect(),
}
}
#[test]
fn a_fresh_watcher_reports_no_change() {
let tmp = TempDir::new().unwrap();
fs::write(tmp.path().join("a.md"), "one").unwrap();
let mut watcher = FsWatcher::new(tmp.path());
assert!(watcher.poll().is_none(), "no change since construction");
assert!(watcher.poll().is_none(), "still no change");
}
#[test]
fn adding_a_file_is_detected_once() {
let tmp = TempDir::new().unwrap();
let mut watcher = FsWatcher::new(tmp.path());
fs::write(tmp.path().join("a.md"), "one").unwrap();
let delta = watcher.poll().expect("new file detected");
assert!(delta.changed.contains(&PathBuf::from("a.md")));
assert_eq!(delta.removed, [] as [std::path::PathBuf; 0]);
assert!(watcher.poll().is_none(), "reported only once");
}
#[test]
fn removing_a_file_is_detected() {
let tmp = TempDir::new().unwrap();
let file = tmp.path().join("a.md");
fs::write(&file, "one").unwrap();
let mut watcher = FsWatcher::new(tmp.path());
fs::remove_file(&file).unwrap();
let delta = watcher.poll().expect("deletion detected");
assert!(delta.removed.contains(&PathBuf::from("a.md")));
assert!(!delta.changed.contains(&PathBuf::from("a.md")));
}
#[test]
fn changing_a_file_size_is_detected() {
let tmp = TempDir::new().unwrap();
let file = tmp.path().join("a.md");
fs::write(&file, "one").unwrap();
let mut watcher = FsWatcher::new(tmp.path());
fs::write(&file, "a much longer body").unwrap();
let delta = watcher.poll().expect("content change detected");
assert!(delta.changed.contains(&PathBuf::from("a.md")));
assert_eq!(delta.removed, [] as [std::path::PathBuf; 0]);
}
#[test]
fn nested_segment_paths_are_relative_to_the_root() {
let tmp = TempDir::new().unwrap();
fs::create_dir_all(tmp.path().join("alice")).unwrap();
let mut watcher = FsWatcher::new(tmp.path());
fs::write(tmp.path().join("alice").join("a.md"), "one").unwrap();
let delta = watcher.poll().expect("change inside a segment detected");
assert!(delta.changed.contains(&PathBuf::from("alice/a.md")));
assert!(
!delta.changed.iter().any(|path| path.is_absolute()),
"every reported path is root-relative: {:?}",
delta.changed
);
}
#[test]
fn legacy_flat_files_are_reported_by_basename() {
let tmp = TempDir::new().unwrap();
let mut watcher = FsWatcher::new(tmp.path());
fs::write(tmp.path().join("a.md"), "one").unwrap();
let delta = watcher.poll().unwrap();
assert!(delta.changed.contains(&PathBuf::from("a.md")));
assert!(
!delta
.changed
.iter()
.any(|path| path.components().count() != 1)
);
}
#[test]
fn a_rename_is_a_remove_and_an_add() {
let tmp = TempDir::new().unwrap();
fs::write(tmp.path().join("a.md"), "one").unwrap();
let mut watcher = FsWatcher::new(tmp.path());
fs::rename(tmp.path().join("a.md"), tmp.path().join("b.md")).unwrap();
let delta = watcher.poll().expect("rename detected");
assert!(delta.removed.contains(&PathBuf::from("a.md")));
assert!(delta.changed.contains(&PathBuf::from("b.md")));
assert!(!delta.changed.contains(&PathBuf::from("a.md")));
}
#[test]
fn missing_root_is_not_an_error() {
let tmp = TempDir::new().unwrap();
let mut watcher = FsWatcher::new(tmp.path().join("does-not-exist"));
assert!(watcher.poll().is_none());
}
#[test]
fn merge_keeps_the_later_observation() {
let mut pending = delta(&["a.md"], &[]);
pending.merge(delta(&["b.md"], &["a.md"]));
assert_eq!(pending.changed, vec![PathBuf::from("b.md")]);
assert_eq!(pending.removed, vec![PathBuf::from("a.md")]);
}
#[test]
fn merge_deduplicates() {
let mut pending = delta(&["a.md"], &["b.md"]);
pending.merge(delta(&["a.md"], &["b.md"]));
assert_eq!(pending.changed, vec![PathBuf::from("a.md")]);
assert_eq!(pending.removed, vec![PathBuf::from("b.md")]);
}
#[test]
fn a_burst_coalesces_into_one_apply() {
let mut coalescer = DeltaCoalescer::default();
assert!(coalescer.observe(Some(delta(&["a.md"], &[]))).is_none());
assert!(coalescer.observe(Some(delta(&["b.md"], &[]))).is_none());
let merged = coalescer
.observe(None)
.expect("the settled tick applies once");
assert_eq!(
merged.changed,
vec![PathBuf::from("a.md"), PathBuf::from("b.md")]
);
assert!(coalescer.observe(None).is_none(), "no second apply");
}
#[test]
fn a_retried_delta_is_applied_again_on_the_next_quiet_tick() {
let mut coalescer = DeltaCoalescer::default();
assert!(coalescer.observe(Some(delta(&["a.md"], &[]))).is_none());
let failed = coalescer.observe(None).unwrap();
coalescer.retry(failed);
let retried = coalescer
.observe(None)
.expect("a retried delta is still pending");
assert_eq!(retried.changed, vec![PathBuf::from("a.md")]);
}
}