rpi_cli/
fs_watch.rs

1//! Lightweight file watching for live reload.
2//!
3//! Port of native Pi's `packages/coding-agent/src/utils/fs-watch.ts`. Native pi
4//! uses Node's `fs.watch` with a retry-on-error wrapper to (a) keep the git
5//! footer's branch/status fresh and (b) hot-reload the active theme file.
6//!
7//! rpi watches by polling modification times. That avoids a native watcher
8//! dependency and behaves predictably on every platform (network drives, WSL
9//! mounts, editors that replace files by rename). The retry delay matches
10//! native [`FS_WATCH_RETRY_DELAY_MS`].
11
12use std::collections::HashMap;
13use std::path::{Path, PathBuf};
14use std::sync::atomic::{AtomicBool, Ordering};
15use std::sync::Arc;
16use std::time::{Duration, SystemTime};
17
18/// Native `FS_WATCH_RETRY_DELAY_MS`: how long to wait after a watch error
19/// before retrying.
20pub const FS_WATCH_RETRY_DELAY_MS: u64 = 5_000;
21
22/// Default poll interval. Short enough to feel immediate, long enough that a
23/// few watched files cost nothing on the main thread.
24pub const DEFAULT_POLL_INTERVAL_MS: u64 = 1_000;
25
26fn mtime(path: &Path) -> Option<SystemTime> {
27    std::fs::metadata(path).ok().and_then(|m| m.modified().ok())
28}
29
30fn snapshot(paths: &[PathBuf]) -> HashMap<PathBuf, Option<SystemTime>> {
31    paths.iter().map(|p| (p.clone(), mtime(p))).collect()
32}
33
34/// A running watch. Dropping it does **not** stop the watch; call
35/// [`WatchHandle::close`] to stop it deterministically (native `closeWatcher`).
36pub struct WatchHandle {
37    stop: Arc<AtomicBool>,
38    join: Option<tokio::task::JoinHandle<()>>,
39}
40
41impl WatchHandle {
42    /// Signal the watch to stop and wait for its task to finish.
43    pub fn close(mut self) {
44        self.stop.store(true, Ordering::SeqCst);
45        if let Some(join) = self.join.take() {
46            join.abort();
47        }
48    }
49
50    /// Request stop without joining (useful from a drop path).
51    pub fn request_stop(&self) {
52        self.stop.store(true, Ordering::SeqCst);
53    }
54}
55
56impl Drop for WatchHandle {
57    fn drop(&mut self) {
58        self.stop.store(true, Ordering::SeqCst);
59    }
60}
61
62/// Watch `paths` for mtime changes, invoking `on_change` whenever any watched
63/// path changes (content or existence). Missing paths are tracked too: creating
64/// or deleting a watched path fires `on_change`.
65///
66/// Poll failures are tolerated: the watcher waits [`FS_WATCH_RETRY_DELAY_MS`] and
67/// retries, never panicking (native `watchWithErrorHandler`).
68pub fn watch_paths<F>(paths: Vec<PathBuf>, poll_interval: Duration, on_change: F) -> WatchHandle
69where
70    F: Fn() + Send + Sync + 'static,
71{
72    let stop = Arc::new(AtomicBool::new(false));
73    let stop_task = stop.clone();
74    let join = tokio::spawn(async move {
75        let mut prev = snapshot(&paths);
76        loop {
77            tokio::time::sleep(poll_interval).await;
78            if stop_task.load(Ordering::SeqCst) {
79                break;
80            }
81            let current = match std::panic::catch_unwind(|| snapshot(&paths)) {
82                Ok(snap) => snap,
83                Err(_) => {
84                    // A metadata read blew up; wait the retry delay and try
85                    // again rather than tearing down the watch.
86                    tokio::time::sleep(Duration::from_millis(FS_WATCH_RETRY_DELAY_MS)).await;
87                    continue;
88                }
89            };
90            if current != prev {
91                prev = current;
92                on_change();
93            }
94        }
95    });
96    WatchHandle {
97        stop,
98        join: Some(join),
99    }
100}
101
102/// Convenience for a single path.
103pub fn watch_path<F>(path: PathBuf, on_change: F) -> WatchHandle
104where
105    F: Fn() + Send + Sync + 'static,
106{
107    watch_paths(vec![path], Duration::from_millis(DEFAULT_POLL_INTERVAL_MS), on_change)
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113    use std::sync::atomic::AtomicUsize;
114
115    #[test]
116    fn detects_content_change() {
117        let dir = tempfile::tempdir().unwrap();
118        let file = dir.path().join("theme.json");
119        std::fs::write(&file, "a").unwrap();
120
121        let hits = Arc::new(AtomicUsize::new(0));
122        let hits_task = hits.clone();
123        let rt = tokio::runtime::Builder::new_current_thread()
124            .enable_time()
125            .build()
126            .unwrap();
127        rt.block_on(async {
128            let handle = watch_paths(
129                vec![file.clone()],
130                Duration::from_millis(20),
131                move || {
132                    hits_task.fetch_add(1, Ordering::SeqCst);
133                },
134            );
135            tokio::time::sleep(Duration::from_millis(60)).await;
136            std::fs::write(&file, "bb").unwrap();
137            // mtime resolution: ensure a visible change.
138            tokio::time::sleep(Duration::from_millis(120)).await;
139            handle.close();
140        });
141        assert!(hits.load(Ordering::SeqCst) >= 1, "expected at least one change");
142    }
143
144    #[test]
145    fn close_stops_watch() {
146        let dir = tempfile::tempdir().unwrap();
147        let file = dir.path().join("f");
148        std::fs::write(&file, "x").unwrap();
149        let rt = tokio::runtime::Builder::new_current_thread()
150            .enable_time()
151            .build()
152            .unwrap();
153        rt.block_on(async {
154            let handle = watch_path(file, || {});
155            handle.close();
156        });
157    }
158}
rpi_cli/fs_watch.rs

rpi_cli/
fs_watch.rs
