agentmux_srv/backend/fs_watch/
recovery.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Health tracking + retry policy for `FsWatchPool`. Split out of `pool.rs`
5//! so the recovery bookkeeping (what's degraded, which backend is active)
6//! is testable independent of the notify/tokio wiring.
7
8use std::collections::HashMap;
9use std::path::PathBuf;
10use std::sync::Mutex;
11use std::time::Duration;
12
13/// Which `notify` backend is actually driving watches right now. `Polling`
14/// only happens if the native backend failed to construct at all (rare —
15/// e.g. inotify unavailable) — see `SPEC_SHARED_FS_WATCHER_FRAMEWORK_2026_08_07.md`
16/// §4 point 2. Once picked at pool construction, this doesn't change for the
17/// pool's lifetime; per-path failures are tracked in `degraded_paths`
18/// instead, independent of which backend is in use.
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub enum WatchBackend {
21    Native,
22    Polling,
23}
24
25/// Snapshot of the pool's current health, for diagnostics (`muxlog`,
26/// eventually a UI surface — not built here, see spec §6 non-goals).
27#[derive(Debug, Clone)]
28#[allow(dead_code)] // Diagnostics surface, SPEC_SHARED_FS_WATCHER_FRAMEWORK_2026_08_07.md §4 point 4.
29pub struct FsWatchHealth {
30    pub backend: WatchBackend,
31    pub active_watches: usize,
32    /// Paths that failed to (re-)establish a watch on their last attempt,
33    /// paired with the error. Retried in the background per
34    /// `RETRY_BACKOFF`/the periodic sweep; a path leaves this list the
35    /// moment any attempt succeeds.
36    pub degraded_paths: Vec<(PathBuf, String)>,
37}
38
39/// Backoff schedule for retrying a failed `watch()` call, per spec §4 point
40/// 1 — bounded at 3 attempts over ~4.2s total, not indefinite: a path that's
41/// still failing after this moves into `degraded_paths` and is picked up by
42/// the periodic health sweep (`pool.rs`'s `HEALTH_SWEEP_INTERVAL`) instead of
43/// spinning a dedicated retry loop forever.
44pub const RETRY_BACKOFF: [Duration; 3] = [
45    Duration::from_millis(200),
46    Duration::from_millis(800),
47    Duration::from_millis(3200),
48];
49
50/// How often the background sweep re-arms every currently-*degraded*
51/// path, self-healing a watch that failed to establish (or a prior sweep
52/// found dead) without needing a dedicated retry loop running forever.
53/// **Deliberately scoped to degraded targets only, not every subscribed
54/// path** — an earlier version of this sweep re-issued `watch()`
55/// unconditionally on everything, every tick, to also catch a *silently*
56/// dead healthy-looking watch (inotify instance-limit churn, a watched
57/// directory deleted and recreated at a new inode, a flaky network mount).
58/// That cost more than it bought: `notify` 7.0.0's Windows backend leaks a
59/// File + Semaphore handle pair on every redundant `watch()` call (no
60/// cleanup of the previous entry), and even with an `unwatch()` first to
61/// avoid the leak, the gap between `unwatch()` and `watch()` drops event
62/// coverage for consumers with no reconciliation backstop of their own
63/// (`config_watcher_fs`, `EditorFileWatcher`, `MediaFileWatcher`). See
64/// `pool.rs`'s `sweep()` and
65/// `docs/status/STATUS_FS_WATCH_SWEEP_HANDLE_LEAK_2026_08_22.md` for the
66/// full history and the accepted trade-off (a truly silent death of an
67/// already-healthy watch is no longer self-healed by this sweep).
68pub const HEALTH_SWEEP_INTERVAL: Duration = Duration::from_secs(30);
69
70#[derive(Default)]
71pub(super) struct HealthState {
72    degraded: Mutex<HashMap<PathBuf, String>>,
73}
74
75impl HealthState {
76    pub(super) fn mark_degraded(&self, path: PathBuf, err: String) {
77        self.degraded.lock().unwrap().insert(path, err);
78    }
79
80    pub(super) fn clear_degraded(&self, path: &std::path::Path) {
81        self.degraded.lock().unwrap().remove(path);
82    }
83
84    pub(super) fn is_degraded(&self, path: &std::path::Path) -> bool {
85        self.degraded.lock().unwrap().contains_key(path)
86    }
87
88    pub(super) fn snapshot(&self) -> Vec<(PathBuf, String)> {
89        self.degraded
90            .lock()
91            .unwrap()
92            .iter()
93            .map(|(p, e)| (p.clone(), e.clone()))
94            .collect()
95    }
96}
97
98#[cfg(test)]
99mod tests {
100    use super::*;
101
102    #[test]
103    fn mark_then_clear_round_trips() {
104        let state = HealthState::default();
105        let path = PathBuf::from("/tmp/watched-thing");
106        state.mark_degraded(path.clone(), "permission denied".to_string());
107        assert_eq!(state.snapshot(), vec![(path.clone(), "permission denied".to_string())]);
108
109        state.clear_degraded(&path);
110        assert!(state.snapshot().is_empty());
111    }
112
113    #[test]
114    fn clearing_an_unknown_path_is_a_no_op() {
115        let state = HealthState::default();
116        state.clear_degraded(std::path::Path::new("/never/watched"));
117        assert!(state.snapshot().is_empty());
118    }
119
120    #[test]
121    fn retry_backoff_is_bounded_and_increasing() {
122        assert_eq!(RETRY_BACKOFF.len(), 3);
123        assert!(RETRY_BACKOFF[0] < RETRY_BACKOFF[1]);
124        assert!(RETRY_BACKOFF[1] < RETRY_BACKOFF[2]);
125    }
126}
agentmux_srv/backend/fs_watch/recovery.rs

agentmux_srv/backend/fs_watch/
recovery.rs
