rig_tools/
watch.rs

1//! Watching a workspace's files: [`Watch`], what
2//! [`Workspace::watch`](crate::workspace::Workspace::watch) gives.
3//! [`Local`](crate::workspace::Local) hears of changes from the system (inotify, FSEvents, …); the default
4//! looks every second.
5
6use std::collections::{BTreeMap, BTreeSet, HashMap};
7use std::fmt;
8use std::fs;
9use std::io;
10use std::path::{Path, PathBuf};
11use std::pin::Pin;
12use std::sync::Arc;
13use std::sync::mpsc::{Receiver, RecvTimeoutError};
14use std::task::{Context, Poll};
15use std::thread;
16use std::time::{Duration, Instant, SystemTime};
17
18use futures::Stream;
19use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender, unbounded};
20use ignore::Match;
21use ignore::gitignore::{Gitignore, GitignoreBuilder};
22use notify::{EventKind, RecursiveMode, Watcher as _};
23use rig_core::tool::{ContentHash, content_hash};
24
25use crate::MAX_FILE_BYTES;
26use crate::workspace::{Listing, Stat, Workspace};
27
28/// How often the default watch looks.
29const EVERY: Duration = Duration::from_secs(1);
30/// How long a batch waits for the next event before it is sent.
31const QUIET: Duration = Duration::from_millis(50);
32/// The longest a batch waits, however many events come.
33const LONGEST: Duration = Duration::from_millis(500);
34
35/// Batches of paths created, changed or removed under a watched root, each
36/// batch in path order, as a [`Stream`]. Dropping it stops the watch.
37pub struct Watch {
38    /// The batches, as they come.
39    batches: UnboundedReceiver<Vec<PathBuf>>,
40    /// What feeds them, stopped when dropped.
41    _keep: Box<dyn Send>,
42}
43
44impl Watch {
45    /// A watch whose batches come on `batches`, keeping `keep` (such as the
46    /// system watcher that feeds them) until it is dropped: for a
47    /// workspace of your own that hears of changes its own way.
48    pub fn new(batches: UnboundedReceiver<Vec<PathBuf>>, keep: impl Send + 'static) -> Self {
49        Self {
50            batches,
51            _keep: Box::new(keep),
52        }
53    }
54}
55
56impl Stream for Watch {
57    type Item = Vec<PathBuf>;
58
59    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
60        Pin::new(&mut self.batches).poll_next(context)
61    }
62}
63
64impl fmt::Debug for Watch {
65    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
66        f.write_str("Watch")
67    }
68}
69
70/// What a look saw of one file: its size and change time, and its hash
71/// when the workspace cannot tell the time.
72type Mark = (u64, Option<SystemTime>, Option<ContentHash>);
73
74/// The default watch: looks at what `workspace` lists under `root` every
75/// second, on a thread of its own, and sends what changed.
76pub(crate) fn poll<W: Workspace + ?Sized>(
77    workspace: Arc<W>,
78    root: &Path,
79    recursive: bool,
80    listing: Listing,
81) -> io::Result<Watch> {
82    let (sender, batches) = unbounded();
83    let root = root.to_path_buf();
84    let mut before = look(&*workspace, &root, recursive, listing);
85    thread::Builder::new()
86        .name("rig-watch".to_owned())
87        .spawn(move || {
88            while !sender.is_closed() {
89                thread::sleep(EVERY);
90                let now = look(&*workspace, &root, recursive, listing);
91                let changed = differences(&before, &now);
92                before = now;
93                if !changed.is_empty() && sender.unbounded_send(changed).is_err() {
94                    break;
95                }
96            }
97        })?;
98    Ok(Watch::new(batches, ()))
99}
100
101/// The files under `root` that `workspace` lists with `listing`, each with
102/// its [`Mark`].
103fn look<W: Workspace + ?Sized>(
104    workspace: &W,
105    root: &Path,
106    recursive: bool,
107    listing: Listing,
108) -> BTreeMap<PathBuf, Mark> {
109    let files = workspace.files(root, None, listing).unwrap_or_default();
110    files
111        .into_iter()
112        .filter(|path| recursive || path == root || path.parent() == Some(root))
113        .filter_map(|path| match workspace.stat(&path) {
114            Ok(Stat::File { size, modified, .. }) => {
115                let hash = modified.is_none().then(|| {
116                    let bytes = workspace.read(&path, MAX_FILE_BYTES).unwrap_or_default();
117                    content_hash(&bytes.bytes)
118                });
119                Some((path, (size, modified, hash)))
120            }
121            _ => None,
122        })
123        .collect()
124}
125
126/// The paths created, changed or removed from `before` to `after`, in path
127/// order.
128fn differences(before: &BTreeMap<PathBuf, Mark>, after: &BTreeMap<PathBuf, Mark>) -> Vec<PathBuf> {
129    let changed = after
130        .iter()
131        .filter(|(path, now)| before.get(path.as_path()) != Some(now))
132        .map(|(path, _)| path);
133    let removed = before
134        .keys()
135        .filter(|path| !after.contains_key(path.as_path()));
136    let paths: BTreeSet<&PathBuf> = changed.chain(removed).collect();
137    paths.into_iter().cloned().collect()
138}
139
140/// [`Local`](crate::workspace::Local)'s watch: the system's file events
141/// under `root`, filtered as [`Workspace::files`] lists with `listing`,
142/// and sent in batches. A `root` that is a file is watched through its
143/// directory, so a write that replaces it is seen; one that does not exist
144/// yet is looked at every second instead.
145pub(crate) fn notified(root: &Path, recursive: bool, listing: Listing) -> io::Result<Watch> {
146    let Ok(meta) = fs::metadata(root) else {
147        return poll(Arc::new(crate::workspace::Local), root, recursive, listing);
148    };
149    let (events, raw) = std::sync::mpsc::channel::<Raw>();
150    let mut watcher = notify::recommended_watcher(move |event: notify::Result<notify::Event>| {
151        let Ok(event) = event else {
152            return;
153        };
154        if matches!(event.kind, EventKind::Access(_)) {
155            return;
156        }
157        let created = matches!(event.kind, EventKind::Create(_));
158        for path in event.paths {
159            events.send(Raw { path, created }).ok();
160        }
161    })
162    .map_err(io::Error::other)?;
163    let file = (!meta.is_dir()).then(|| root.to_path_buf());
164    let (target, mode) = match file.as_deref().and_then(Path::parent) {
165        Some(directory) => (directory, RecursiveMode::NonRecursive),
166        None if recursive => (root, RecursiveMode::Recursive),
167        None => (root, RecursiveMode::NonRecursive),
168    };
169    watcher.watch(target, mode).map_err(io::Error::other)?;
170    let filter = Filter {
171        root: root.to_path_buf(),
172        file,
173        recursive,
174        listing,
175        ignores: Ignores::new(root),
176    };
177    let (sender, batches) = unbounded();
178    thread::Builder::new()
179        .name("rig-watch".to_owned())
180        .spawn(move || send_batches(&raw, &sender, filter))?;
181    Ok(Watch::new(batches, watcher))
182}
183
184/// A path the system's watcher named, and whether it was created.
185pub(crate) struct Raw {
186    pub(crate) path: PathBuf,
187    pub(crate) created: bool,
188}
189
190/// Every file under `dir`, when it is a directory, into `files`; symbolic
191/// links are not followed.
192fn files_in(dir: &Path, files: &mut Vec<PathBuf>) {
193    let Ok(entries) = fs::read_dir(dir) else {
194        return;
195    };
196    for entry in entries.flatten() {
197        let path = entry.path();
198        match entry.file_type() {
199            Ok(kind) if kind.is_dir() => files_in(&path, files),
200            Ok(kind) if kind.is_file() => files.push(path),
201            _ => {}
202        }
203    }
204}
205
206/// Gathers the paths `raw` names into batches, each sent once no event came
207/// for [`QUIET`] or it waited [`LONGEST`], until the watcher or the
208/// [`Watch`] is gone. A directory created under a recursive watch is
209/// watched only after the watcher has told of it, so the files already in
210/// it when its batch ends, by then watched, join the batch.
211fn send_batches(raw: &Receiver<Raw>, sender: &UnboundedSender<Vec<PathBuf>>, mut filter: Filter) {
212    while let Ok(first) = raw.recv() {
213        let mut paths = BTreeSet::new();
214        let mut directories = Vec::new();
215        let deadline = Instant::now() + LONGEST;
216        let mut next = Ok(first);
217        while let Ok(Raw { path, created }) = next {
218            if created && filter.recursive && filter.file.is_none() && path.is_dir() {
219                directories.push(path);
220            } else if filter.keeps(&path) {
221                paths.insert(path);
222            }
223            let left = deadline.saturating_duration_since(Instant::now());
224            next = match left.is_zero() {
225                true => Err(RecvTimeoutError::Timeout),
226                false => raw.recv_timeout(QUIET.min(left)),
227            };
228        }
229        let mut found = Vec::new();
230        for directory in &directories {
231            files_in(directory, &mut found);
232        }
233        paths.extend(found.into_iter().filter(|path| filter.keeps(path)));
234        if !paths.is_empty() && sender.unbounded_send(paths.into_iter().collect()).is_err() {
235            return;
236        }
237        if matches!(next, Err(RecvTimeoutError::Disconnected)) {
238            return;
239        }
240    }
241}
242
243/// Which event paths a watch passes on.
244struct Filter {
245    /// The watched root.
246    root: PathBuf,
247    /// The watched file, when the root is one.
248    file: Option<PathBuf>,
249    /// Whether directories under the root count.
250    recursive: bool,
251    /// Which files count.
252    listing: Listing,
253    /// The ignore rules, read as they are needed.
254    ignores: Ignores,
255}
256
257impl Filter {
258    /// Whether `path` is one the watch reports: under the root (one level
259    /// down unless recursive), no directory, and a file
260    /// [`Workspace::files`] would list with the watch's listing.
261    fn keeps(&mut self, path: &Path) -> bool {
262        if let Some(file) = &self.file {
263            return path == file;
264        }
265        let Ok(below) = path.strip_prefix(&self.root) else {
266            return false;
267        };
268        if path
269            .file_name()
270            .is_some_and(|name| name == ".gitignore" || name == ".ignore")
271        {
272            self.ignores.forget(path.parent());
273        }
274        if below.as_os_str().is_empty()
275            || (!self.recursive && below.components().count() > 1)
276            || fs::metadata(path).is_ok_and(|meta| meta.is_dir())
277        {
278            return false;
279        }
280        self.listing == Listing::All || (self.listing.keeps(below) && !self.ignores.ignored(path))
281    }
282}
283
284/// The ignore rules [`Workspace::files`] follows on this machine (each
285/// directory's `.ignore`, and in a git work tree its `.gitignore` files,
286/// `.git/info/exclude` and the user's global excludes), read per directory
287/// when first needed.
288struct Ignores {
289    /// The highest directory whose rules count: the work tree's top, or
290    /// the root outside one.
291    top: PathBuf,
292    /// The watched root: the rules decide about what is below it, never
293    /// about the root itself or a directory it is in.
294    root: PathBuf,
295    /// Whether the root is in a git work tree.
296    git: bool,
297    /// The repository's and the user's excludes, the weakest rules.
298    excludes: Vec<Gitignore>,
299    /// Each directory's own rules.
300    directories: HashMap<PathBuf, Gitignore>,
301}
302
303impl Ignores {
304    /// The rules for a watch of `root`.
305    fn new(root: &Path) -> Self {
306        let top = root.ancestors().find(|dir| dir.join(".git").exists());
307        let mut excludes = Vec::new();
308        if let Some(top) = top {
309            excludes.push(Gitignore::new(top.join(".git/info/exclude")).0);
310            excludes.push(Gitignore::global().0);
311        }
312        Self {
313            top: top.unwrap_or(root).to_path_buf(),
314            root: root.to_path_buf(),
315            git: top.is_some(),
316            excludes,
317            directories: HashMap::new(),
318        }
319    }
320
321    /// Reads `directory`'s rules again when next needed.
322    fn forget(&mut self, directory: Option<&Path>) {
323        if let Some(directory) = directory {
324            self.directories.remove(directory);
325        }
326    }
327
328    /// Whether `path`, a file, or a directory it is in below the root, is
329    /// ignored.
330    fn ignored(&mut self, path: &Path) -> bool {
331        let Ok(below) = path.strip_prefix(&self.top) else {
332            return false;
333        };
334        let parts = below.components().count();
335        let above = self
336            .root
337            .strip_prefix(&self.top)
338            .map_or(0, |root| root.components().count());
339        let mut at = self.top.clone();
340        for (index, part) in below.components().enumerate() {
341            let directory = at.clone();
342            at.push(part);
343            if index >= above && self.decides(&directory, &at, index + 1 < parts) {
344                return true;
345            }
346        }
347        false
348    }
349
350    /// Whether `candidate`, an entry of `directory`, is ignored by the
351    /// rules of `directory` or a directory above it: the deepest rule that
352    /// matches decides.
353    fn decides(&mut self, directory: &Path, candidate: &Path, is_dir: bool) -> bool {
354        for dir in directory.ancestors() {
355            let rules = self.rules(dir);
356            let below = candidate.strip_prefix(dir).unwrap_or(candidate);
357            match rules.matched(below, is_dir) {
358                Match::Ignore(_) => return true,
359                Match::Whitelist(_) => return false,
360                Match::None => {}
361            }
362            if dir == self.top {
363                break;
364            }
365        }
366        let below = candidate.strip_prefix(&self.top).unwrap_or(candidate);
367        self.excludes
368            .iter()
369            .map(|rules| rules.matched(below, is_dir))
370            .find(|found| !found.is_none())
371            .is_some_and(|found| found.is_ignore())
372    }
373
374    /// `dir`'s own rules, read once.
375    fn rules(&mut self, dir: &Path) -> &Gitignore {
376        let git = self.git;
377        self.directories
378            .entry(dir.to_path_buf())
379            .or_insert_with(|| {
380                // The later file wins where both match: `.ignore` over `.gitignore`.
381                let mut builder = GitignoreBuilder::new(dir);
382                if git {
383                    builder.add(dir.join(".gitignore"));
384                }
385                builder.add(dir.join(".ignore"));
386                builder.build().unwrap_or_else(|_| Gitignore::empty())
387            })
388    }
389}
390
391#[cfg(test)]
392mod tests;
rig_tools/watch.rs

rig_tools/
watch.rs
