tor_config/
file_watcher.rs

1//! Code to watch configuration files for any changes.
2//!
3// TODO: perhaps this shouldn't live in tor-config? But it doesn't seem substantial enough to have
4// its own crate, and it can't live in e.g. tor-basic-utils, because it depends on tor-rtcompat.
5
6use std::collections::hash_map::Entry;
7use std::collections::{HashMap, HashSet};
8use std::io;
9use std::marker::PhantomData;
10use std::path::{Path, PathBuf};
11use std::pin::Pin;
12use std::sync::{Arc, Mutex};
13use std::task::{Context, Poll};
14
15use tor_rtcompat::Runtime;
16
17use amplify::Getters;
18use notify::{EventKind, Watcher};
19use postage::watch;
20
21use futures::{Stream, StreamExt as _};
22
23/// `Result` whose `Err` is [`FileWatcherBuildError`].
24pub type Result<T> = std::result::Result<T, FileWatcherBuildError>;
25
26cfg_if::cfg_if! {
27    if #[cfg(any(target_os = "linux", target_os = "android", target_os = "windows"))] {
28        /// The concrete type of the underlying watcher.
29        type NotifyWatcher = notify::RecommendedWatcher;
30    } else {
31        /// The concrete type of the underlying watcher.
32        type NotifyWatcher = notify::PollWatcher;
33    }
34}
35
36/// A wrapper around a `notify::Watcher` to watch a set of parent
37/// directories in order to learn about changes in some specific files that they
38/// contain.
39///
40/// The `Watcher` implementation in `notify` has a weakness: it gives sensible
41/// results when you're watching directories, but if you start watching
42/// non-directory files, it won't notice when those files get replaced.  That's
43/// a problem for users who want to change their configuration atomically by
44/// making new files and then moving them into place over the old ones.
45///
46/// For more background on the issues with `notify`, see
47/// <https://github.com/notify-rs/notify/issues/165> and
48/// <https://github.com/notify-rs/notify/pull/166>.
49///
50/// ## Limitations
51///
52/// On backends using kqueue, this uses a polling watcher
53/// to work around a bug in the `notify` crate[^1].
54/// This introduces a perceivable delay,
55/// and can be very expensive for large file trees.
56///
57/// [^1]: See <https://github.com/notify-rs/notify/issues/644>
58#[derive(Getters)]
59#[must_use = "A dropped FileWatcher exits immediately"]
60pub struct FileWatcher {
61    /// An underlying `notify` watcher that tells us about directory changes.
62    // this field is kept only so the watcher is not dropped
63    #[getter(skip)]
64    _watcher: NotifyWatcher,
65    /// The list of directories that we're currently watching.
66    watching_dirs: HashSet<PathBuf>,
67}
68
69impl FileWatcher {
70    /// Create a `FileWatcherBuilder`
71    pub fn builder<R: Runtime>(runtime: R) -> FileWatcherBuilder<R> {
72        FileWatcherBuilder::new(runtime)
73    }
74}
75
76/// Event possibly triggering a configuration reload
77//
78// WARNING!
79//
80// Simply adding new, more specific, events, to this struct, would be wrong.
81// This is because internally, we transmit the events via a postage::watch,
82// which means that receivers might not receive all events!
83#[derive(Debug, Clone, PartialEq)]
84#[non_exhaustive]
85pub enum Event {
86    /// Some files may have been modified.
87    ///
88    /// This is semantically equivalent to `Rescan`, since in neither case
89    /// do we say *which* files may have been changed.
90    FileChanged,
91    /// Some filesystem events may have been missed.
92    Rescan,
93}
94
95/// Builder used to configure a [`FileWatcher`] before it starts watching for changes.
96pub struct FileWatcherBuilder<R: Runtime> {
97    /// The runtime.  We used to use this.
98    ///
99    /// TODO get rid of this, but after we decide whether to keep using postage::watch.
100    /// See the Warning note on Event.
101    #[allow(dead_code)]
102    runtime: PhantomData<R>,
103    /// The list of directories that we're currently watching.
104    ///
105    /// Each directory has a set of filters that indicates whether a given notify::Event
106    /// is relevant or not.
107    watching_dirs: HashMap<PathBuf, HashSet<DirEventFilter>>,
108}
109
110/// A filter for deciding what to do with a notify::Event pertaining
111/// to files that are relative to one of the directories we are watching.
112///
113// Private, as this is an implementation detail.
114// If/when we decide to make this public, this might need revisiting.
115#[derive(Clone, Debug, Hash, PartialEq, Eq)]
116enum DirEventFilter {
117    /// Notify the caller about the event, if the file has the specified extension.
118    MatchesExtension(String),
119    /// Notify the caller about the event, if the file has the specified path.
120    MatchesPath(PathBuf),
121}
122
123impl DirEventFilter {
124    /// Check whether this filter accepts `path`.
125    fn accepts_path(&self, path: &Path) -> bool {
126        match self {
127            DirEventFilter::MatchesExtension(ext) => path
128                .extension()
129                .and_then(|ext| ext.to_str())
130                .map(|e| e == ext.as_str())
131                .unwrap_or_default(),
132            DirEventFilter::MatchesPath(p) => p == path,
133        }
134    }
135}
136
137impl<R: Runtime> FileWatcherBuilder<R> {
138    /// Create a `FileWatcherBuilder`
139    pub fn new(_runtime: R) -> Self {
140        FileWatcherBuilder {
141            runtime: PhantomData,
142            watching_dirs: HashMap::new(),
143        }
144    }
145
146    /// Add a single path to the list of things to watch.
147    ///
148    /// The event receiver will be notified if the path is created, modified, renamed, or removed.
149    ///
150    /// If the path is a directory, its contents will **not** be watched.
151    /// To watch the contents of a directory, use [`watch_dir`](FileWatcherBuilder::watch_dir).
152    ///
153    /// Idempotent: does nothing if we're already watching that path.
154    pub fn watch_path<P: AsRef<Path>>(&mut self, path: P) -> Result<()> {
155        self.watch_just_parents(path.as_ref())?;
156        Ok(())
157    }
158
159    /// Add a directory (but not any subdirs) to the list of things to watch.
160    ///
161    /// The event receiver will be notified whenever a file with the specified `extension`
162    /// is created within this directory, or if an existing file with this extension
163    /// is modified, renamed, or removed.
164    /// Changes to files that have a different extension are ignored.
165    ///
166    /// Idempotent.
167    pub fn watch_dir<P: AsRef<Path>, S: AsRef<str>>(
168        &mut self,
169        path: P,
170        extension: S,
171    ) -> Result<()> {
172        let path = self.watch_just_parents(path.as_ref())?;
173        self.watch_just_abs_dir(
174            &path,
175            DirEventFilter::MatchesExtension(extension.as_ref().into()),
176        );
177        Ok(())
178    }
179
180    /// Add the parents of `path` to the list of things to watch.
181    ///
182    /// Returns the absolute path of `path`.
183    ///
184    /// Idempotent.
185    fn watch_just_parents(&mut self, path: &Path) -> Result<PathBuf> {
186        // Make the path absolute (without necessarily making it canonical).
187        //
188        // We do this because `notify` reports all of its events in terms of
189        // absolute paths, so if we were to tell it to watch a directory by its
190        // relative path, we'd get reports about the absolute paths of the files
191        // in that directory.
192        let cwd = std::env::current_dir()
193            .map_err(|e| FileWatcherBuildError::CurrentDirectory(Arc::new(e)))?;
194        let path = cwd.join(path);
195        debug_assert!(path.is_absolute());
196
197        // See what directory we should watch in order to watch this file.
198        let watch_target = match path.parent() {
199            // The file has a parent, so watch that.
200            Some(parent) => parent,
201            // The file has no parent.  Given that it's absolute, that means
202            // that we're looking at the root directory.  There's nowhere to go
203            // "up" from there.
204            None => path.as_ref(),
205        };
206
207        // Note this file as one that we're watching, so that we can see changes
208        // to it later on.
209        self.watch_just_abs_dir(watch_target, DirEventFilter::MatchesPath(path.clone()));
210
211        Ok(path)
212    }
213
214    /// Add just this (absolute) directory to the list of things to watch.
215    ///
216    /// Does not watch any of the parents.
217    ///
218    /// Idempotent.
219    fn watch_just_abs_dir(&mut self, watch_target: &Path, filter: DirEventFilter) {
220        match self.watching_dirs.entry(watch_target.to_path_buf()) {
221            Entry::Occupied(mut o) => {
222                let _: bool = o.get_mut().insert(filter);
223            }
224            Entry::Vacant(v) => {
225                let _ = v.insert(HashSet::from([filter]));
226            }
227        }
228    }
229
230    /// Build a `FileWatcher` and start sending events to `tx`.
231    ///
232    /// On startup, the watcher sends a [`Rescan`](Event::Rescan) event.
233    /// This helps mitigate the event loss that occurs if the watched files are modified between
234    /// the time they are initially loaded and the time when the watcher is set up.
235    pub fn start_watching(self, tx: FileEventSender) -> Result<FileWatcher> {
236        let watching_dirs = self.watching_dirs.clone();
237        let event_sender = move |event: notify::Result<notify::Event>| {
238            let event = handle_event(event, &watching_dirs);
239            if let Some(event) = event {
240                // NB!  This can lose events!  See the internal warning comment on `Event`
241                *tx.0.lock().expect("poisoned").borrow_mut() = event;
242            }
243        };
244
245        cfg_if::cfg_if! {
246            if #[cfg(any(target_os = "linux", target_os = "android", target_os = "windows"))] {
247                let config = notify::Config::default();
248            } else {
249                /// The polling frequency, for use with the `PollWatcher`.
250                #[cfg(not(any(test, feature = "testing")))]
251                const WATCHER_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5);
252
253                #[cfg(any(test, feature = "testing"))]
254                const WATCHER_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(10);
255
256                let config = notify::Config::default()
257                    .with_poll_interval(WATCHER_POLL_INTERVAL);
258
259                // When testing, compare the contents of the files too, not just their mtime
260                // Otherwise, because the polling backend detects changes based on mtime,
261                // if the test creates/writes files too fast,
262                // it will fail to notice changes (this can happen, for example, on a tmpfs).
263                #[cfg(any(test, feature = "testing"))]
264                let config = config.with_compare_contents(true);
265            }
266        }
267
268        let mut watcher = NotifyWatcher::new(event_sender, config).map_err(Arc::new)?;
269
270        let watching_dirs: HashSet<_> = self.watching_dirs.keys().cloned().collect();
271        for dir in &watching_dirs {
272            watcher
273                .watch(dir, notify::RecursiveMode::NonRecursive)
274                .map_err(Arc::new)?;
275        }
276
277        Ok(FileWatcher {
278            _watcher: watcher,
279            watching_dirs,
280        })
281    }
282}
283
284/// Map a `notify` event to the [`Event`] type returned by [`FileWatcher`].
285fn handle_event(
286    event: notify::Result<notify::Event>,
287    watching_dirs: &HashMap<PathBuf, HashSet<DirEventFilter>>,
288) -> Option<Event> {
289    let watching = |f: &PathBuf| {
290        // For paths with no parent (i.e. root), the watcher is added for the path itself,
291        // so we do the same here.
292        let parent = f.parent().unwrap_or_else(|| f.as_ref());
293
294        // Find the filters that apply to this directory
295        match watching_dirs
296            .iter()
297            .find_map(|(dir, filters)| (dir == parent).then_some(filters))
298        {
299            Some(filters) => {
300                // This event is interesting, if any of the filters apply.
301                filters.iter().any(|filter| filter.accepts_path(f.as_ref()))
302            }
303            None => false,
304        }
305    };
306
307    // filter events we don't want and map to event code
308    match event {
309        Ok(event) => {
310            if event.need_rescan() {
311                Some(Event::Rescan)
312            } else if ignore_event_kind(&event.kind) {
313                None
314            } else if event.paths.iter().any(watching) {
315                Some(Event::FileChanged)
316            } else {
317                None
318            }
319        }
320        Err(error) => {
321            if error.paths.iter().any(watching) {
322                Some(Event::FileChanged)
323            } else {
324                None
325            }
326        }
327    }
328}
329
330/// Check whether this is a kind of [`notify::Event`] that we want to ignore.
331///
332/// Returns `true` for
333///   * events that trigger on non-mutating file accesses
334///   * catch-all events (used by `notify` for unsupported/unknown events)
335///   * "other" meta-events
336fn ignore_event_kind(kind: &EventKind) -> bool {
337    use EventKind::*;
338    matches!(kind, Access(_) | Any | Other)
339}
340
341/// The sender half of a watch channel used by a [`FileWatcher`] for sending [`Event`]s.
342///
343/// For use with [`FileWatcherBuilder::start_watching`].
344///
345/// **Important**: to avoid contention, avoid sharing clones of the same `FileEventSender`
346/// with multiple [`FileWatcherBuilder`]s. This type is [`Clone`] to support creating new
347/// [`FileWatcher`]s from an existing [`channel`], which enables existing receivers to receive
348/// events from new `FileWatcher`s (any old `FileWatcher`s are supposed to be discarded).
349#[derive(Clone)]
350pub struct FileEventSender(Arc<Mutex<watch::Sender<Event>>>);
351
352/// The receiver half of a watch channel used for receiving [`Event`]s sent by a [`FileWatcher`].
353#[derive(Clone)]
354pub struct FileEventReceiver(watch::Receiver<Event>);
355
356impl Stream for FileEventReceiver {
357    type Item = Event;
358
359    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
360        self.0.poll_next_unpin(cx)
361    }
362}
363
364impl FileEventReceiver {
365    /// Try to read a message from the stream, without blocking.
366    ///
367    /// Returns `Some` if a message is ready.
368    /// Returns `None` if the stream is open, but no messages are available,
369    /// or if the stream is closed.
370    pub fn try_recv(&mut self) -> Option<Event> {
371        use postage::prelude::Stream;
372
373        self.0.try_recv().ok()
374    }
375}
376
377/// Create a new channel for use with a [`FileWatcher`].
378//
379// Note: the [`FileEventSender`] and [`FileEventReceiver`]  wrappers exist
380// so we don't expose the channel's underlying type
381// in our public API.
382pub fn channel() -> (FileEventSender, FileEventReceiver) {
383    let (tx, rx) = watch::channel_with(Event::Rescan);
384    (
385        FileEventSender(Arc::new(Mutex::new(tx))),
386        FileEventReceiver(rx),
387    )
388}
389
390/// An error coming from a [`FileWatcherBuilder`].
391#[derive(Debug, Clone, thiserror::Error)]
392#[non_exhaustive]
393pub enum FileWatcherBuildError {
394    /// Invalid current working directory.
395    ///
396    /// This error can happen if the current directory does not exist,
397    /// or if we don't have the necessary permissions to access it.
398    #[error("Invalid current working directory")]
399    CurrentDirectory(#[source] Arc<io::Error>),
400
401    /// Encountered a problem while creating a `Watcher`.
402    #[error("Problem creating Watcher")]
403    Notify(#[from] Arc<notify::Error>),
404}
405
406#[cfg(test)]
407mod test {
408    // @@ begin test lint list maintained by maint/add_warning @@
409    #![allow(clippy::bool_assert_comparison)]
410    #![allow(clippy::clone_on_copy)]
411    #![allow(clippy::dbg_macro)]
412    #![allow(clippy::mixed_attributes_style)]
413    #![allow(clippy::print_stderr)]
414    #![allow(clippy::print_stdout)]
415    #![allow(clippy::single_char_pattern)]
416    #![allow(clippy::unwrap_used)]
417    #![allow(clippy::unchecked_time_subtraction)]
418    #![allow(clippy::useless_vec)]
419    #![allow(clippy::needless_pass_by_value)]
420    #![allow(clippy::string_slice)] // See arti#2571
421    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
422
423    use super::*;
424    use notify::event::{AccessKind, ModifyKind};
425    use test_temp_dir::{TestTempDir, test_temp_dir};
426
427    /// Write `data` to file `name` within `dir`.
428    fn write_file(dir: &TestTempDir, name: &str, data: &[u8]) -> PathBuf {
429        let path = dir.as_path_untracked().join(name);
430        std::fs::write(&path, data).unwrap();
431        path
432    }
433
434    /// Return an event that has the Rescan flag set
435    fn rescan_event() -> notify::Event {
436        let event = notify::Event::new(notify::EventKind::Any);
437        event.set_flag(notify::event::Flag::Rescan)
438    }
439
440    /// Assert that at least one FileChanged event is received.
441    async fn assert_file_changed(rx: &mut FileEventReceiver) {
442        assert_eq!(rx.next().await, Some(Event::FileChanged));
443
444        // The write might trigger more than one event
445        while let Some(ev) = rx.try_recv() {
446            assert_eq!(ev, Event::FileChanged);
447        }
448    }
449
450    /// Set the `EventKind` of `event` to an uninteresting `EventKind`
451    /// and assert that it is ignored by `handle_event`.
452    fn assert_ignored(event: &notify::Event, watching: &HashMap<PathBuf, HashSet<DirEventFilter>>) {
453        for kind in [EventKind::Access(AccessKind::Any), EventKind::Other] {
454            let ignored_event = event.clone().set_kind(kind);
455            assert_eq!(handle_event(Ok(ignored_event.clone()), watching), None);
456            // ...but if the rescan flag is set, the event is *not* ignored
457            let event = ignored_event.set_flag(notify::event::Flag::Rescan);
458            assert_eq!(handle_event(Ok(event), watching), Some(Event::Rescan));
459        }
460    }
461
462    #[test]
463    fn notify_event_handler() {
464        let mut event = notify::Event::new(notify::EventKind::Modify(ModifyKind::Any));
465
466        let mut watching_dirs = Default::default();
467        assert_eq!(handle_event(Ok(event.clone()), &watching_dirs), None);
468        assert_eq!(
469            handle_event(Ok(rescan_event()), &watching_dirs),
470            Some(Event::Rescan)
471        );
472
473        // Watch some directories
474        watching_dirs.insert(
475            "/foo/baz".into(),
476            HashSet::from([DirEventFilter::MatchesExtension("auth".into())]),
477        );
478        assert_eq!(handle_event(Ok(event.clone()), &watching_dirs), None);
479        assert_eq!(
480            handle_event(Ok(rescan_event()), &watching_dirs),
481            Some(Event::Rescan)
482        );
483
484        event = event.add_path("/foo/bar/alice.authh".into());
485        assert_eq!(handle_event(Ok(event.clone()), &watching_dirs), None);
486
487        event = event.add_path("/foo/bar/alice.auth".into());
488        assert_eq!(handle_event(Ok(event.clone()), &watching_dirs), None);
489
490        event = event.add_path("/foo/baz/bob.auth".into());
491        assert_eq!(
492            handle_event(Ok(event.clone()), &watching_dirs),
493            Some(Event::FileChanged)
494        );
495
496        // The same event, but with an irrelevant kind, gets ignored:
497        assert_ignored(&event, &watching_dirs);
498
499        // Watch some files within /foo/bar
500        watching_dirs.insert(
501            "/foo/bar".into(),
502            HashSet::from([DirEventFilter::MatchesPath("/foo/bar/abc".into())]),
503        );
504
505        assert_eq!(
506            handle_event(Ok(event.clone()), &watching_dirs),
507            Some(Event::FileChanged)
508        );
509        assert_eq!(
510            handle_event(Ok(rescan_event()), &watching_dirs),
511            Some(Event::Rescan)
512        );
513
514        // The same event, but with an irrelevant kind, gets ignored:
515        assert_ignored(&event, &watching_dirs);
516
517        // Watch some other files
518        let event = notify::Event::new(notify::EventKind::Modify(ModifyKind::Any))
519            .add_path("/a/b/c/d".into());
520        let watching_dirs = [(
521            "/a/b/c/".into(),
522            HashSet::from([DirEventFilter::MatchesPath("/a/b/c/d".into())]),
523        )]
524        .into_iter()
525        .collect();
526        assert_eq!(
527            handle_event(Ok(event), &watching_dirs),
528            Some(Event::FileChanged)
529        );
530        assert_eq!(
531            handle_event(Ok(rescan_event()), &watching_dirs),
532            Some(Event::Rescan)
533        );
534
535        // Errors can also trigger an event
536        let err = notify::Error::path_not_found();
537        assert_eq!(handle_event(Err(err), &watching_dirs), None);
538        let mut err = notify::Error::path_not_found();
539        err = err.add_path("/a/b/c/d".into());
540        assert_eq!(
541            handle_event(Err(err), &watching_dirs),
542            Some(Event::FileChanged)
543        );
544    }
545
546    #[test]
547    fn watch_dirs() {
548        tor_rtcompat::test_with_one_runtime!(|rt| async move {
549            let temp_dir = test_temp_dir!();
550            let (tx, mut rx) = channel();
551            // Watch for changes in .foo files from temp_dir
552            let mut builder = FileWatcher::builder(rt.clone());
553            builder
554                .watch_dir(temp_dir.as_path_untracked(), "foo")
555                .unwrap();
556            let watcher = builder.start_watching(tx).unwrap();
557
558            // On startup, the watcher sends a Event::Rescan event.
559            // This is because the watcher is often set up after loading
560            // the files or directories it is watching.
561            assert_eq!(rx.try_recv(), Some(Event::Rescan));
562            assert_eq!(rx.try_recv(), None);
563
564            // Write a file with extension "foo".
565            write_file(&temp_dir, "bar.foo", b"hello");
566
567            assert_eq!(rx.next().await, Some(Event::FileChanged));
568
569            drop(watcher);
570            // The write might trigger more than one event
571            while let Some(ev) = rx.next().await {
572                assert_eq!(ev.clone(), Event::FileChanged);
573            }
574        });
575    }
576
577    #[test]
578    fn watch_file_path() {
579        tor_rtcompat::test_with_one_runtime!(|rt| async move {
580            let temp_dir = test_temp_dir!();
581            let (tx, mut rx) = channel();
582            // Watch for changes to hello.txt
583            let path = write_file(&temp_dir, "hello.txt", b"hello");
584            let mut builder = FileWatcher::builder(rt.clone());
585            builder.watch_path(&path).unwrap();
586            let _watcher = builder.start_watching(tx).unwrap();
587
588            // On startup, the watcher sends a Event::Rescan event.
589            assert_eq!(rx.try_recv(), Some(Event::Rescan));
590            assert_eq!(rx.try_recv(), None);
591
592            // Write to hello.txt
593            let _: PathBuf = write_file(&temp_dir, "hello.txt", b"good-bye");
594
595            assert_file_changed(&mut rx).await;
596
597            // Remove hello.txt
598            std::fs::remove_file(&path).unwrap();
599            assert_file_changed(&mut rx).await;
600
601            // Create a new file
602            let tmp_hello = write_file(&temp_dir, "hello.tmp", b"new hello");
603            // Copy it over to the watched hello.txt location
604            std::fs::rename(&tmp_hello, &path).unwrap();
605            assert_file_changed(&mut rx).await;
606        });
607    }
608
609    #[test]
610    fn watch_dir_path() {
611        tor_rtcompat::test_with_one_runtime!(|rt| async move {
612            let temp_dir1 = tempfile::TempDir::new().unwrap();
613            let (tx, mut rx) = channel();
614            // Watch temp_dir for changes
615            let mut builder = FileWatcher::builder(rt.clone());
616            builder.watch_path(temp_dir1.path()).unwrap();
617
618            let _watcher = builder.start_watching(tx).unwrap();
619
620            // On startup, the watcher sends a Event::Rescan event.
621            assert_eq!(rx.try_recv(), Some(Event::Rescan));
622            assert_eq!(rx.try_recv(), None);
623
624            // Writing a file to this directory shouldn't trigger an event
625            std::fs::write(temp_dir1.path().join("hello.txt"), b"hello").unwrap();
626            assert_eq!(rx.try_recv(), None);
627
628            // Move temp_dir1 to temp_dir2
629            let temp_dir2 = tempfile::TempDir::new().unwrap();
630            std::fs::rename(&temp_dir1, &temp_dir2).unwrap();
631
632            // Moving the directory triggers an event...
633            assert_file_changed(&mut rx).await;
634            // ...and so does moving it back to its original location
635            std::fs::rename(&temp_dir2, &temp_dir1).unwrap();
636            assert_file_changed(&mut rx).await;
637        });
638    }
639}
tor_config/file_watcher.rs

tor_config/
file_watcher.rs
