atelier_sdk/
watch.rs

1use std::path::{Component, Path};
2use std::sync::Arc;
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::mpsc::{Receiver, RecvTimeoutError};
5use std::time::Duration;
6
7use notify::{Event, EventKind};
8
9use crate::error::Error;
10use crate::workspace::SKIP_NAMES;
11
12/// How often a blocked watch loop wakes to check its stop handle. Waking
13/// is a condition check, never a filesystem scan — edits arrive as events.
14pub(crate) const STOP_TICK: Duration = Duration::from_millis(100);
15
16/// What a running watch loop reports as it works.
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub enum WatchEvent {
19    /// The watcher is armed: edits from now on raise events. The catch-up
20    /// scan for edits made while no watcher ran follows immediately.
21    Started,
22    /// Outstanding edits became this snapshot, journaled like any snapshot.
23    Snapshotted {
24        /// The snapshot's id.
25        snapshot: String,
26    },
27}
28
29/// Stops a running watch loop from another thread; the loop returns within
30/// its tick. Outstanding edits stay for the next watcher's catch-up scan.
31#[derive(Debug, Clone, Default)]
32pub struct WatchStop(Arc<AtomicBool>);
33
34impl WatchStop {
35    /// A fresh, un-triggered stop handle.
36    #[must_use]
37    pub fn new() -> Self {
38        Self::default()
39    }
40
41    /// Ask the loop to return.
42    pub fn stop(&self) {
43        self.0.store(true, Ordering::Release);
44    }
45
46    pub(crate) fn stopped(&self) -> bool {
47        self.0.load(Ordering::Acquire)
48    }
49}
50
51/// What the notify callback forwards to the watch loop: a content pulse,
52/// or the watcher's own failure — loud, never swallowed.
53pub(crate) type Pulse = Result<(), notify::Error>;
54
55/// Drain the event storm until it stays quiet for `debounce`, then let the
56/// caller snapshot once. A stop request wins over the storm.
57pub(crate) fn settle(
58    pulses: &Receiver<Pulse>,
59    debounce: Duration,
60    stop: &WatchStop,
61) -> Result<(), Error> {
62    loop {
63        if stop.stopped() {
64            return Ok(());
65        }
66        match pulses.recv_timeout(debounce) {
67            Ok(Ok(())) => {}
68            Ok(Err(error)) => return Err(watcher_failed(&error)),
69            Err(RecvTimeoutError::Timeout) => return Ok(()),
70            Err(RecvTimeoutError::Disconnected) => return Err(watcher_gone()),
71        }
72    }
73}
74
75pub(crate) fn watcher_failed(error: &notify::Error) -> Error {
76    Error::Engine(format!("the file watcher failed: {error}"))
77}
78
79pub(crate) fn watcher_gone() -> Error {
80    Error::Engine("the file watcher stopped delivering events".to_owned())
81}
82
83/// An fs event is content when any of its paths lands outside the engine
84/// internals (`.atelier`, `.jj`, `.git`). Events wholly inside them are the
85/// engine's and journal's own writes — reacting to those would loop.
86pub(crate) fn event_is_content(root: &Path, event: &Event) -> bool {
87    if matches!(event.kind, EventKind::Access(_)) {
88        return false;
89    }
90    // A pathless event (a rescan hint) may cover content; a snapshot of an
91    // unchanged tree is a no-op, so erring toward content is safe.
92    if event.paths.is_empty() {
93        return true;
94    }
95    event.paths.iter().any(|path| is_content_path(root, path))
96}
97
98fn is_content_path(root: &Path, path: &Path) -> bool {
99    let Ok(relative) = path.strip_prefix(root) else {
100        return true;
101    };
102    match relative.components().next() {
103        Some(Component::Normal(name)) => !SKIP_NAMES.iter().any(|skip| name == *skip),
104        _ => true,
105    }
106}
atelier_sdk/watch.rs

atelier_sdk/
watch.rs
