meerkat_sqlite/
watch.rs

1//! Cross-process change watch for one SQLite database file.
2//!
3//! Another process's commit to a database is invisible to this process's
4//! in-memory signals. This watch turns file-system notifications on the
5//! database, its `-wal` and `-shm` sidecars and their directory into
6//! coalesced ticks for a consumer that then reads durable state.
7//!
8//! A notification is a HINT, never a guarantee: SQLite writes WAL frames
9//! before the WAL index makes the commit visible, so a reader woken by an
10//! event can still read the old state, and no later event is promised. The
11//! watch therefore also ticks on a bounded sweep interval; the consumer's
12//! sweep read is the eventual-wake guarantee. Neither the watch nor the sweep
13//! ever defines permission, custody or expiry: correctness rests with the
14//! consumer's durable reads and its own fences.
15//!
16//! This is the contract the mob event bus has used since its external watch
17//! landed, extracted so every store shares one implementation.
18
19use std::path::{Path, PathBuf};
20use std::sync::mpsc;
21use std::thread;
22use std::time::{Duration, Instant};
23
24use notify::{RecursiveMode, Watcher as _};
25
26/// Default sweep interval: the bounded eventual-wake guarantee.
27pub const SQLITE_WATCH_SWEEP: Duration = Duration::from_millis(5_000);
28/// Settle delay after a notification, so a burst of events (a commit touches
29/// the WAL and its index) coalesces into one tick.
30const SQLITE_WATCH_COALESCE: Duration = Duration::from_millis(10);
31const SQLITE_WATCH_RECOVERY_BACKOFF_MIN: Duration = Duration::from_millis(100);
32const SQLITE_WATCH_RECOVERY_BACKOFF_MAX: Duration = Duration::from_millis(30_000);
33
34/// Why the consumer is being asked to look.
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub enum SqliteWatchTick {
37    /// A relevant file-system notification arrived (coalesced).
38    Changed,
39    /// The sweep interval elapsed without a notification.
40    Sweep,
41}
42
43/// What the consumer's tick achieved.
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45pub enum SqliteWatchControl {
46    /// The tick was handled.
47    Handled,
48    /// The consumer's durable read failed: ticks are suppressed until a
49    /// bounded exponential backoff elapses, so a notification storm cannot
50    /// hammer a failing store.
51    Failed,
52    /// The consumer is gone: stop the watch.
53    Stop,
54}
55
56/// A running watch. Dropping it stops the notifications, and the worker
57/// thread exits at its next wait.
58pub struct SqliteChangeWatch {
59    _watcher: notify::RecommendedWatcher,
60}
61
62impl std::fmt::Debug for SqliteChangeWatch {
63    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
64        f.debug_struct("SqliteChangeWatch").finish_non_exhaustive()
65    }
66}
67
68fn sidecar_path(path: &Path, suffix: &str) -> PathBuf {
69    let mut value = path.as_os_str().to_os_string();
70    value.push(suffix);
71    PathBuf::from(value)
72}
73
74/// The database file and its sidecars.
75pub fn sqlite_watch_paths(path: &Path) -> Vec<PathBuf> {
76    vec![
77        path.to_path_buf(),
78        sidecar_path(path, "-wal"),
79        sidecar_path(path, "-shm"),
80    ]
81}
82
83/// Whether a notification concerns the database (its files or directory).
84pub fn sqlite_watch_event_relevant(
85    event: &notify::Event,
86    parent: &Path,
87    watched_paths: &[PathBuf],
88) -> bool {
89    if matches!(event.kind, notify::EventKind::Access(_)) {
90        return false;
91    }
92    if event.paths.is_empty() {
93        return true;
94    }
95    event.paths.iter().any(|path| {
96        path == parent
97            || watched_paths.iter().any(|watched| {
98                path == watched
99                    || (path.parent() == watched.parent()
100                        && path.file_name() == watched.file_name())
101            })
102    })
103}
104
105/// Start watching `db_path`. `on_tick` runs on a dedicated thread named
106/// `thread_name`, once per coalesced notification and once per `sweep`
107/// without one.
108pub fn start_sqlite_change_watch(
109    db_path: &Path,
110    thread_name: &str,
111    sweep: Duration,
112    mut on_tick: impl FnMut(SqliteWatchTick) -> SqliteWatchControl + Send + 'static,
113) -> Result<SqliteChangeWatch, String> {
114    let parent = db_path
115        .parent()
116        .map(Path::to_path_buf)
117        .ok_or_else(|| format!("{} has no parent directory", db_path.display()))?;
118    let watched_paths = sqlite_watch_paths(db_path);
119    let (wake_tx, wake_rx) = mpsc::channel::<()>();
120
121    let callback_parent = parent.clone();
122    let mut watcher = notify::recommended_watcher(move |result: notify::Result<notify::Event>| {
123        match result {
124            Ok(event) if sqlite_watch_event_relevant(&event, &callback_parent, &watched_paths) => {
125                let _ = wake_tx.send(());
126            }
127            Ok(_) => {}
128            Err(error) => {
129                // The sweep bounds anything a watcher error lost.
130                tracing::warn!(error = %error, "sqlite change watch reported an error");
131            }
132        }
133    })
134    .map_err(|error| format!("cannot create the sqlite change watcher: {error}"))?;
135    watcher
136        .watch(&parent, RecursiveMode::NonRecursive)
137        .map_err(|error| format!("cannot watch {}: {error}", parent.display()))?;
138
139    thread::Builder::new()
140        .name(thread_name.to_string())
141        .spawn(move || {
142            let mut recovery_backoff = SQLITE_WATCH_RECOVERY_BACKOFF_MIN;
143            let mut recovery_deadline: Option<Instant> = None;
144            loop {
145                let wait = recovery_deadline
146                    .map(|deadline| deadline.saturating_duration_since(Instant::now()))
147                    .unwrap_or(sweep);
148                let tick = match wake_rx.recv_timeout(wait) {
149                    Ok(()) => SqliteWatchTick::Changed,
150                    Err(mpsc::RecvTimeoutError::Timeout) => SqliteWatchTick::Sweep,
151                    // The watcher (and every sender) is gone: stop.
152                    Err(mpsc::RecvTimeoutError::Disconnected) => break,
153                };
154                if tick == SqliteWatchTick::Changed {
155                    thread::sleep(SQLITE_WATCH_COALESCE);
156                }
157                while wake_rx.try_recv().is_ok() {}
158                if recovery_deadline.is_some_and(|deadline| Instant::now() < deadline) {
159                    // A notification storm must not bypass a failed read's
160                    // recovery deadline.
161                    continue;
162                }
163                match on_tick(tick) {
164                    SqliteWatchControl::Handled => {
165                        recovery_deadline = None;
166                        recovery_backoff = SQLITE_WATCH_RECOVERY_BACKOFF_MIN;
167                    }
168                    SqliteWatchControl::Failed => {
169                        recovery_deadline = Some(Instant::now() + recovery_backoff);
170                        recovery_backoff = recovery_backoff
171                            .saturating_mul(2)
172                            .min(SQLITE_WATCH_RECOVERY_BACKOFF_MAX);
173                    }
174                    SqliteWatchControl::Stop => break,
175                }
176            }
177        })
178        .map_err(|error| format!("cannot start the sqlite change watch thread: {error}"))?;
179
180    Ok(SqliteChangeWatch { _watcher: watcher })
181}
182
183#[cfg(test)]
184#[allow(clippy::expect_used, clippy::unwrap_used)]
185mod tests {
186    use super::*;
187
188    #[test]
189    fn a_write_to_the_database_ticks_changed() {
190        let dir = tempfile::tempdir().unwrap();
191        let db = dir.path().join("store.sqlite3");
192        std::fs::write(&db, b"").unwrap();
193        let (tick_tx, tick_rx) = mpsc::channel();
194        let _watch =
195            start_sqlite_change_watch(&db, "test-watch", SQLITE_WATCH_SWEEP, move |tick| {
196                let _ = tick_tx.send(tick);
197                SqliteWatchControl::Handled
198            })
199            .expect("watch starts");
200        std::fs::write(sidecar_path(&db, "-wal"), b"frame").unwrap();
201        let tick = tick_rx
202            .recv_timeout(Duration::from_secs(10))
203            .expect("the write is observed within the hang guard");
204        assert_eq!(tick, SqliteWatchTick::Changed);
205    }
206
207    #[test]
208    fn the_sweep_ticks_without_a_notification() {
209        let dir = tempfile::tempdir().unwrap();
210        let db = dir.path().join("store.sqlite3");
211        let (tick_tx, tick_rx) = mpsc::channel();
212        let _watch =
213            start_sqlite_change_watch(&db, "test-sweep", Duration::from_millis(50), move |tick| {
214                let _ = tick_tx.send(tick);
215                SqliteWatchControl::Handled
216            })
217            .expect("watch starts");
218        let tick = tick_rx
219            .recv_timeout(Duration::from_secs(10))
220            .expect("the sweep ticks within the hang guard");
221        assert_eq!(tick, SqliteWatchTick::Sweep);
222    }
223
224    #[test]
225    fn stop_ends_the_worker() {
226        let dir = tempfile::tempdir().unwrap();
227        let db = dir.path().join("store.sqlite3");
228        let (tick_tx, tick_rx) = mpsc::channel();
229        let _watch =
230            start_sqlite_change_watch(&db, "test-stop", Duration::from_millis(20), move |tick| {
231                let _ = tick_tx.send(tick);
232                SqliteWatchControl::Stop
233            })
234            .expect("watch starts");
235        tick_rx
236            .recv_timeout(Duration::from_secs(10))
237            .expect("one tick");
238        assert!(
239            tick_rx.recv_timeout(Duration::from_millis(500)).is_err(),
240            "no tick after Stop: the worker ended (its sender is gone)"
241        );
242    }
243}
meerkat_sqlite/watch.rs

meerkat_sqlite/
watch.rs
