piw/bundle/
watch.rs

1//! Filesystem watching for the runs directory. Events are debounced into a
2//! single "something changed" tick on a tokio channel; consumers rescan the
3//! directory and poll their tailers, which is cheap and race-free because
4//! all bundle files are either append-only or atomically replaced.
5
6use notify::{RecommendedWatcher, RecursiveMode, Watcher};
7use std::path::Path;
8use std::time::Duration;
9use tokio::sync::mpsc;
10
11pub struct RunsWatcher {
12    // Held for its Drop side effect: dropping stops the watcher thread.
13    _watcher: RecommendedWatcher,
14    rx: mpsc::Receiver<()>,
15}
16
17impl RunsWatcher {
18    pub fn new(runs_dir: &Path) -> notify::Result<Self> {
19        let (tx, rx) = mpsc::channel(1);
20        let mut watcher =
21            notify::recommended_watcher(move |result: notify::Result<notify::Event>| {
22                if result.is_ok() {
23                    // try_send: a pending tick already guarantees a rescan.
24                    let _ = tx.try_send(());
25                }
26            })?;
27        watcher.watch(runs_dir, RecursiveMode::Recursive)?;
28        Ok(Self {
29            _watcher: watcher,
30            rx,
31        })
32    }
33
34    /// Wait until something under the runs directory changed, coalescing
35    /// bursts of events with a short quiet period.
36    pub async fn changed(&mut self) {
37        if self.rx.recv().await.is_none() {
38            // Watcher gone; fall back to slow polling so the UI stays live.
39            tokio::time::sleep(Duration::from_secs(2)).await;
40            return;
41        }
42        // Absorb the burst that a single logical write produces.
43        while let Ok(Some(())) =
44            tokio::time::timeout(Duration::from_millis(40), self.rx.recv()).await
45        {}
46    }
47}
piw/bundle/watch.rs

piw/bundle/
watch.rs
