use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::rc::{Rc, Weak};
use std::sync::Arc;
use notify::{RecommendedWatcher, RecursiveMode, Watcher};
use crate::file::SettingsFileError;
use crate::reload::Reloadable;
pub type SettingsReloadSink = Arc<dyn Fn(PathBuf) + Send + Sync + 'static>;
pub struct SettingsWatcher {
_inner: RecommendedWatcher,
}
impl SettingsWatcher {
pub fn new(dirs: Vec<PathBuf>, sink: SettingsReloadSink) -> Result<Self, notify::Error> {
let mut seen: std::collections::HashSet<PathBuf> = std::collections::HashSet::new();
let mut targets: Vec<PathBuf> = Vec::new();
for dir in dirs {
let canonical = match dir.canonicalize() {
Ok(p) => p,
Err(e) => {
eprintln!(
"teksilo-settings: cannot watch `{}` ({e}); live settings reload \
disabled for that directory",
dir.display()
);
continue;
}
};
if seen.insert(canonical.clone()) {
targets.push(canonical);
}
}
let sink_handle = sink.clone();
let mut watcher = notify::recommended_watcher(
move |res: Result<notify::Event, notify::Error>| match res {
Ok(event) if should_reload(&event.kind) => {
for path in &event.paths {
(sink_handle)(path.clone());
}
}
Ok(_) => {}
Err(e) => {
eprintln!("teksilo-settings: watcher error: {e}");
}
},
)?;
for target in &targets {
watcher.watch(target, RecursiveMode::NonRecursive)?;
}
Ok(Self { _inner: watcher })
}
}
fn should_reload(kind: ¬ify::EventKind) -> bool {
use notify::EventKind::*;
matches!(kind, Modify(_) | Create(_))
}
fn canonical_settings_path(path: &Path) -> PathBuf {
match (path.parent(), path.file_name()) {
(Some(parent), Some(name)) => match parent.canonicalize() {
Ok(canonical_parent) => canonical_parent.join(name),
Err(_) => path.to_path_buf(),
},
_ => path.to_path_buf(),
}
}
#[derive(Clone, Default)]
pub struct SettingsRegistry {
entries: Rc<std::cell::RefCell<HashMap<PathBuf, Weak<dyn Reloadable>>>>,
}
impl SettingsRegistry {
pub fn new() -> Self {
Self::default()
}
pub fn register(&self, reloadable: Rc<dyn Reloadable>) -> Rc<dyn Reloadable> {
let key = canonical_settings_path(reloadable.path());
self.entries
.borrow_mut()
.insert(key, Rc::downgrade(&reloadable));
reloadable
}
pub fn dispatch(&self, changed_path: &Path) -> Result<bool, SettingsFileError> {
let key = canonical_settings_path(changed_path);
let weak = { self.entries.borrow().get(&key).cloned() };
let Some(weak) = weak else {
return Ok(false);
};
match weak.upgrade() {
Some(reloadable) => reloadable.reload_from_disk(),
None => {
self.entries.borrow_mut().remove(&key);
Ok(false)
}
}
}
pub fn registered_paths(&self) -> Vec<PathBuf> {
self.entries.borrow().keys().cloned().collect()
}
pub fn live_count(&self) -> usize {
self.entries
.borrow()
.values()
.filter(|w| w.upgrade().is_some())
.count()
}
}
impl std::fmt::Debug for SettingsRegistry {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let entries = self.entries.borrow();
f.debug_struct("SettingsRegistry")
.field("registered_paths", &entries.len())
.field(
"live",
&entries.values().filter(|w| w.upgrade().is_some()).count(),
)
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::file::SettingsFile;
use crate::migration::{Migrator, Versioned};
use serde::{Deserialize, Serialize};
use std::time::{Duration, Instant};
use tempfile::tempdir;
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
struct Prefs {
version: u32,
value: String,
}
impl Versioned for Prefs {
const CURRENT_VERSION: u32 = 1;
fn version(&self) -> u32 {
self.version
}
fn set_version(&mut self, v: u32) {
self.version = v;
}
}
fn poll_until(timeout: Duration, mut condition: impl FnMut() -> bool) -> bool {
let deadline = Instant::now() + timeout;
loop {
if condition() {
return true;
}
if Instant::now() >= deadline {
return false;
}
std::thread::sleep(Duration::from_millis(20));
}
}
const GENEROUS_TIMEOUT: Duration = Duration::from_secs(5);
#[test]
fn canonical_settings_path_resolves_even_when_file_is_missing() {
let dir = tempdir().unwrap();
let missing = dir.path().join("not-yet-written.toml");
let resolved = canonical_settings_path(&missing);
assert_eq!(
resolved.parent().unwrap(),
dir.path().canonicalize().unwrap()
);
assert_eq!(resolved.file_name().unwrap(), "not-yet-written.toml");
}
#[test]
fn canonical_settings_path_is_stable_across_existence() {
let dir = tempdir().unwrap();
let path = dir.path().join("general.toml");
let before = canonical_settings_path(&path);
std::fs::write(&path, "version = 1\n").unwrap();
let after = canonical_settings_path(&path);
assert_eq!(
before, after,
"registering before vs. after creation must key identically"
);
}
#[test]
fn dispatch_on_unregistered_path_is_a_harmless_no_op() {
let registry = SettingsRegistry::new();
let dir = tempdir().unwrap();
assert!(!registry.dispatch(&dir.path().join("unknown.toml")).unwrap());
}
#[test]
fn register_then_dispatch_reloads_a_peers_write() {
let dir = tempdir().unwrap();
let path = dir.path().join("prefs.toml");
let a: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let b: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let registry = SettingsRegistry::new();
let _handle = registry.register(Rc::new(b.clone()) as Rc<dyn Reloadable>);
a.mutate(|p| p.value = "peer-write".into()).unwrap();
assert!(registry.dispatch(&path).unwrap());
assert_eq!(b.snapshot().value, "peer-write");
}
#[test]
fn dropped_service_is_pruned_and_never_dispatched_to() {
let dir = tempdir().unwrap();
let path = dir.path().join("prefs.toml");
let a: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let b: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let registry = SettingsRegistry::new();
let handle = registry.register(Rc::new(b.clone()) as Rc<dyn Reloadable>);
assert_eq!(registry.live_count(), 1);
drop(handle);
assert_eq!(registry.live_count(), 0);
a.mutate(|p| p.value = "peer-write-after-drop".into())
.unwrap();
assert!(!registry.dispatch(&path).unwrap());
assert!(registry.registered_paths().is_empty());
assert_eq!(b.snapshot().value, "");
}
#[test]
fn registering_under_the_same_path_replaces_the_previous_owner() {
let dir = tempdir().unwrap();
let path = dir.path().join("prefs.toml");
let first: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let second: SettingsFile<Prefs> =
SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let registry = SettingsRegistry::new();
let _first_handle = registry.register(Rc::new(first.clone()) as Rc<dyn Reloadable>);
let _second_handle = registry.register(Rc::new(second.clone()) as Rc<dyn Reloadable>);
assert_eq!(registry.registered_paths().len(), 1);
let writer: SettingsFile<Prefs> =
SettingsFile::load(path.clone(), Migrator::new()).unwrap();
writer.mutate(|p| p.value = "via-second".into()).unwrap();
assert!(registry.dispatch(&path).unwrap());
assert_eq!(second.snapshot().value, "via-second");
assert_eq!(first.snapshot().value, "");
}
#[test]
fn construction_over_a_missing_directory_does_not_error() {
let sink: SettingsReloadSink = Arc::new(|_path| {});
let watcher = SettingsWatcher::new(
vec![PathBuf::from("/definitely/does/not/exist/anywhere")],
sink,
);
assert!(watcher.is_ok());
}
#[test]
fn construction_dedupes_identical_directories() {
let dir = tempdir().unwrap();
let sink: SettingsReloadSink = Arc::new(|_path| {});
let watcher = SettingsWatcher::new(
vec![dir.path().to_path_buf(), dir.path().to_path_buf()],
sink,
);
assert!(watcher.is_ok());
}
type PathQueue = Arc<std::sync::Mutex<std::collections::VecDeque<PathBuf>>>;
fn queueing_sink(queue: PathQueue) -> SettingsReloadSink {
Arc::new(move |path| {
queue.lock().unwrap().push_back(path);
})
}
fn drain_and_dispatch(registry: &SettingsRegistry, queue: &PathQueue) -> usize {
let paths: Vec<PathBuf> = queue.lock().unwrap().drain(..).collect();
paths
.into_iter()
.filter(|p| matches!(registry.dispatch(p), Ok(true)))
.count()
}
#[test]
fn external_write_triggers_exactly_one_effective_reload() {
let dir = tempdir().unwrap();
let path = dir.path().join("prefs.toml");
let peer: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let mine: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let registry = SettingsRegistry::new();
let _handle = registry.register(Rc::new(mine.clone()) as Rc<dyn Reloadable>);
let queue: PathQueue = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new()));
let _watcher =
SettingsWatcher::new(vec![dir.path().to_path_buf()], queueing_sink(queue.clone()))
.unwrap();
peer.mutate(|p| p.value = "external".into()).unwrap();
let mut effective_reloads = 0usize;
assert!(
poll_until(GENEROUS_TIMEOUT, || {
effective_reloads += drain_and_dispatch(®istry, &queue);
effective_reloads >= 1
}),
"expected the external write to be picked up within the timeout"
);
let deadline = Instant::now() + Duration::from_millis(300);
while Instant::now() < deadline {
effective_reloads += drain_and_dispatch(®istry, &queue);
std::thread::sleep(Duration::from_millis(20));
}
assert_eq!(
effective_reloads, 1,
"exactly one effective reload, however many raw fs events fired"
);
assert_eq!(mine.snapshot().value, "external");
}
#[test]
fn our_own_write_triggers_no_effective_reload() {
let dir = tempdir().unwrap();
let path = dir.path().join("prefs.toml");
let mine: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let registry = SettingsRegistry::new();
let _handle = registry.register(Rc::new(mine.clone()) as Rc<dyn Reloadable>);
mine.mutate(|p| p.value = "mine".into()).unwrap();
assert!(!registry.dispatch(&path).unwrap());
assert_eq!(
mine.snapshot().value,
"mine",
"our own value must be untouched"
);
}
#[test]
fn dropped_service_is_never_called_by_a_live_watcher() {
let dir = tempdir().unwrap();
let dropped_path = dir.path().join("dropped.toml");
let sentinel_path = dir.path().join("sentinel.toml");
let dropped_peer: SettingsFile<Prefs> =
SettingsFile::load(dropped_path.clone(), Migrator::new()).unwrap();
let dropped_mine: SettingsFile<Prefs> =
SettingsFile::load(dropped_path.clone(), Migrator::new()).unwrap();
let sentinel_peer: SettingsFile<Prefs> =
SettingsFile::load(sentinel_path.clone(), Migrator::new()).unwrap();
let sentinel_mine: SettingsFile<Prefs> =
SettingsFile::load(sentinel_path.clone(), Migrator::new()).unwrap();
let registry = SettingsRegistry::new();
let dropped_handle = registry.register(Rc::new(dropped_mine.clone()) as Rc<dyn Reloadable>);
let _sentinel_handle =
registry.register(Rc::new(sentinel_mine.clone()) as Rc<dyn Reloadable>);
drop(dropped_handle);
assert_eq!(registry.live_count(), 1);
let queue: PathQueue = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new()));
let _watcher =
SettingsWatcher::new(vec![dir.path().to_path_buf()], queueing_sink(queue.clone()))
.unwrap();
dropped_peer
.mutate(|p| p.value = "should-never-land".into())
.unwrap();
sentinel_peer
.mutate(|p| p.value = "sentinel-fired".into())
.unwrap();
let dropped_key = canonical_settings_path(&dropped_path);
let sentinel_key = canonical_settings_path(&sentinel_path);
let mut dropped_reloads = 0usize;
let mut sentinel_reloads = 0usize;
let deadline = Instant::now() + GENEROUS_TIMEOUT;
loop {
let paths: Vec<PathBuf> = queue.lock().unwrap().drain(..).collect();
for p in paths {
let key = canonical_settings_path(&p);
if let Ok(true) = registry.dispatch(&p) {
if key == dropped_key {
dropped_reloads += 1;
} else if key == sentinel_key {
sentinel_reloads += 1;
}
}
}
if sentinel_reloads >= 1 || Instant::now() >= deadline {
break;
}
std::thread::sleep(Duration::from_millis(20));
}
assert!(
sentinel_reloads >= 1,
"sentinel write should have been observed within the timeout"
);
assert_eq!(
dropped_reloads, 0,
"a dropped service's registration must never be dispatched to"
);
assert_eq!(
dropped_mine.snapshot().value,
"",
"the dropped service's in-memory value must be untouched"
);
}
#[test]
fn survives_file_deletion_and_recreation() {
let dir = tempdir().unwrap();
let path = dir.path().join("prefs.toml");
std::fs::write(&path, "version = 1\nvalue = \"\"\n").unwrap();
let mine: SettingsFile<Prefs> = SettingsFile::load(path.clone(), Migrator::new()).unwrap();
let registry = SettingsRegistry::new();
let _handle = registry.register(Rc::new(mine.clone()) as Rc<dyn Reloadable>);
let queue: PathQueue = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new()));
let _watcher =
SettingsWatcher::new(vec![dir.path().to_path_buf()], queueing_sink(queue.clone()))
.unwrap();
std::fs::remove_file(&path).unwrap();
let recreated: SettingsFile<Prefs> =
SettingsFile::load(path.clone(), Migrator::new()).unwrap();
recreated
.mutate(|p| p.value = "recreated-after-delete".into())
.unwrap();
let mut effective_reloads = 0usize;
assert!(
poll_until(GENEROUS_TIMEOUT, || {
effective_reloads += drain_and_dispatch(®istry, &queue);
effective_reloads >= 1
}),
"the recreated file's write should still be observed after the watched \
file was deleted and recreated"
);
assert_eq!(mine.snapshot().value, "recreated-after-delete");
}
}