aft/watcher/
mod.rs

1//! Minimal watcher-side collection primitives.
2//!
3//! Watcher threads may invalidate paths and collect a stable byte snapshot, but
4//! never hash, store, or publish artifacts. Those operations belong to plane
5//! workers so a burst of watcher events cannot mutate a view directly.
6
7use std::collections::BTreeSet;
8use std::fmt;
9use std::fs;
10use std::io;
11use std::path::{Path, PathBuf};
12use std::time::{SystemTime, UNIX_EPOCH};
13
14/// A changed stat may trigger this many additional reads after the first read.
15pub const MAX_STABLE_READ_RETRIES: usize = 3;
16/// Includes the first read plus every permitted retry.
17pub const MAX_STABLE_READ_ATTEMPTS: usize = MAX_STABLE_READ_RETRIES + 1;
18
19/// The metadata fields that make a read safe to hand to a plane worker.
20#[derive(Clone, Debug, Eq, PartialEq)]
21pub struct FileStamp {
22    pub size: u64,
23    pub modified_ns: Option<u128>,
24    #[cfg(unix)]
25    pub inode: u64,
26    #[cfg(unix)]
27    pub ctime_ns: i128,
28}
29
30impl FileStamp {
31    pub fn from_metadata(metadata: &fs::Metadata) -> Self {
32        Self {
33            size: metadata.len(),
34            modified_ns: system_time_ns(metadata.modified().ok()),
35            #[cfg(unix)]
36            inode: {
37                use std::os::unix::fs::MetadataExt;
38                metadata.ino()
39            },
40            #[cfg(unix)]
41            ctime_ns: {
42                use std::os::unix::fs::MetadataExt;
43                i128::from(metadata.ctime()) * 1_000_000_000 + i128::from(metadata.ctime_nsec())
44            },
45        }
46    }
47
48    #[cfg(test)]
49    fn synthetic(size: u64, modified_ns: u128) -> Self {
50        Self {
51            size,
52            modified_ns: Some(modified_ns),
53            #[cfg(unix)]
54            inode: 1,
55            #[cfg(unix)]
56            ctime_ns: 1,
57        }
58    }
59}
60
61fn system_time_ns(time: Option<SystemTime>) -> Option<u128> {
62    time.and_then(|time| time.duration_since(UNIX_EPOCH).ok())
63        .map(|duration| duration.as_nanos())
64}
65
66/// A value whose bytes were bracketed by equal filesystem stamps.
67#[derive(Clone, Debug, Eq, PartialEq)]
68pub struct StableRead<T> {
69    pub value: T,
70    pub stamp: FileStamp,
71    pub attempts: usize,
72}
73
74#[derive(Debug)]
75pub enum StableReadError {
76    Io(io::Error),
77    /// The file changed around every allowed read. The caller must leave the
78    /// path pending and wait for another watcher event rather than publishing it.
79    Unstable {
80        attempts: usize,
81    },
82}
83
84impl fmt::Display for StableReadError {
85    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
86        match self {
87            Self::Io(error) => write!(f, "stable read I/O error: {error}"),
88            Self::Unstable { attempts } => {
89                write!(
90                    f,
91                    "file changed around all {attempts} stable-read attempt(s)"
92                )
93            }
94        }
95    }
96}
97
98impl std::error::Error for StableReadError {
99    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
100        match self {
101            Self::Io(error) => Some(error),
102            Self::Unstable { .. } => None,
103        }
104    }
105}
106
107impl From<io::Error> for StableReadError {
108    fn from(error: io::Error) -> Self {
109        Self::Io(error)
110    }
111}
112
113/// Reads `path` only when an identical stat surrounds the read.
114pub fn read_stable_file(path: &Path) -> Result<StableRead<Vec<u8>>, StableReadError> {
115    read_stable(
116        || Ok(FileStamp::from_metadata(&fs::metadata(path)?)),
117        || fs::read(path),
118    )
119}
120
121/// Performs the stable-read algorithm with injectable stat and read operations.
122///
123/// The plane worker owns this operation because it is the component that will
124/// hash the collected bytes next. Watcher dispatch only queues the path.
125pub fn read_stable<T>(
126    mut stat: impl FnMut() -> io::Result<FileStamp>,
127    mut read: impl FnMut() -> io::Result<T>,
128) -> Result<StableRead<T>, StableReadError> {
129    for retry in 0..=MAX_STABLE_READ_RETRIES {
130        let before = stat()?;
131        let value = read()?;
132        let after = stat()?;
133        if before == after {
134            return Ok(StableRead {
135                value,
136                stamp: after,
137                attempts: retry + 1,
138            });
139        }
140    }
141    Err(StableReadError::Unstable {
142        attempts: MAX_STABLE_READ_ATTEMPTS,
143    })
144}
145
146/// The watcher-owned queue of invalidations. It deliberately has no hashing,
147/// blob-store, or publication API; plane workers consume its collected paths.
148#[derive(Debug, Default)]
149pub struct WatcherCollector {
150    invalidated_paths: BTreeSet<PathBuf>,
151}
152
153impl WatcherCollector {
154    pub fn invalidate(&mut self, path: impl Into<PathBuf>) {
155        self.invalidated_paths.insert(path.into());
156    }
157
158    /// Drains one deduplicated collection batch. A later watcher event can add a
159    /// persistently unstable path again without special recovery state.
160    pub fn take_collected(&mut self) -> Vec<PathBuf> {
161        std::mem::take(&mut self.invalidated_paths)
162            .into_iter()
163            .collect()
164    }
165}
166
167#[cfg(test)]
168mod tests {
169    use super::*;
170
171    #[test]
172    fn stat_mismatch_gets_three_retries_then_is_left_unstable() {
173        let mut stat_calls = 0;
174        let mut read_calls = 0;
175        let result = read_stable(
176            || {
177                stat_calls += 1;
178                let stamp = if stat_calls % 2 == 1 {
179                    FileStamp::synthetic(1, stat_calls as u128)
180                } else {
181                    FileStamp::synthetic(2, stat_calls as u128)
182                };
183                Ok(stamp)
184            },
185            || {
186                read_calls += 1;
187                Ok::<_, io::Error>(b"read".to_vec())
188            },
189        );
190
191        assert!(matches!(
192            result,
193            Err(StableReadError::Unstable {
194                attempts: MAX_STABLE_READ_ATTEMPTS
195            })
196        ));
197        assert_eq!(read_calls, MAX_STABLE_READ_ATTEMPTS);
198        assert_eq!(stat_calls, MAX_STABLE_READ_ATTEMPTS * 2);
199    }
200
201    #[test]
202    fn third_retry_can_produce_a_stable_snapshot() {
203        let stable = FileStamp::synthetic(9, 12);
204        let changing = FileStamp::synthetic(9, 13);
205        let mut stamps = vec![
206            stable.clone(),
207            changing.clone(),
208            stable.clone(),
209            changing.clone(),
210            stable.clone(),
211            changing,
212            stable.clone(),
213            stable.clone(),
214        ]
215        .into_iter();
216        let result = read_stable(
217            || Ok(stamps.next().expect("one stamp per stat")),
218            || Ok::<_, io::Error>(b"stable bytes".to_vec()),
219        )
220        .expect("third retry succeeds");
221
222        assert_eq!(result.value, b"stable bytes");
223        assert_eq!(result.attempts, MAX_STABLE_READ_ATTEMPTS);
224        assert_eq!(result.stamp, stable);
225    }
226
227    #[test]
228    fn watcher_collector_only_deduplicates_and_drains_paths() {
229        let mut collector = WatcherCollector::default();
230        collector.invalidate("src/z.rs");
231        collector.invalidate("src/a.rs");
232        collector.invalidate("src/z.rs");
233        assert_eq!(
234            collector.take_collected(),
235            vec![PathBuf::from("src/a.rs"), PathBuf::from("src/z.rs")]
236        );
237        assert!(collector.take_collected().is_empty());
238    }
239}
aft/watcher/mod.rs

aft/watcher/
mod.rs
