Migrate datadir, config watch, and IPC onto dcc-daemon

Cargo.toml

@@ -19,12 +19,13 @@ path = "src/main.rs"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 clap = { version = "4", features = ["derive"] }
+# Git HTTPS from rust-commons (`main`; bump with `make deps-update`).
+bigfred-shared-daemon = { git = "https://github.com/dcc-bigfred/rust-commons.git", branch = "main" }
 mdns-sd = "0.20"
 nix = { version = "0.29", features = ["signal", "fs", "socket"] }
 thiserror = "2"
 libc = "0.2"
 socket2 = "0.6"
-notify = "8.2.0"
 log = "0.4"
 env_logger = "0.11"
 

Makefile

@@ -6,7 +6,7 @@ RUSTUP_TOOLCHAIN ?= stable
 export RUSTUP_TOOLCHAIN
 
 .PHONY: all build release release-musl check test test-release-assertions \
-	clean fmt clippy hub-upload deploy
+	clean fmt clippy hub-upload deploy deps-update
 
 all: build
 
@@ -38,6 +38,10 @@ fmt:
 clippy:
 	$(CARGO) clippy --all-targets -- -D warnings
 
+# Refresh git crates (bigfred-shared-daemon) and rewrite Cargo.lock. Commit the lockfile afterwards.
+deps-update:
+	$(CARGO) update -p bigfred-shared-daemon
+
 clean:
 	$(CARGO) clean
 	rm -rf dist

src/bigfred_watch.rs

@@ -4,11 +4,11 @@
 //! connection (poll). Success body for `dcc_bus_list` is `{ "programs": [...] }`
 //! matching REST; errors are `{ "error": "<code>" }`.
 
-use std::io::{Read, Write};
 use std::os::unix::net::UnixStream;
 use std::path::Path;
 use std::time::Duration;
 
+use bigfred_shared_daemon::ipc::{read_frame_bytes, write_frame_with_limit};
 use serde::{Deserialize, Serialize};
 
 use crate::error::{Error, Result};
@@ -94,30 +94,9 @@ fn request_raw(socket_path: &Path, req: &Request) -> Result<Vec<u8>> {
 }
 
 fn write_frame(stream: &mut UnixStream, msg: &impl Serialize) -> Result<()> {
-    let payload = serde_json::to_vec(msg)?;
-    if payload.len() > MAX_FRAME {
-        return Err(Error::Ipc(format!(
-            "frame length {} exceeds max {MAX_FRAME}",
-            payload.len()
-        )));
-    }
-    let len = u32::try_from(payload.len())
-        .map_err(|_| Error::Ipc("frame too large for u32 length prefix".into()))?
-        .to_le_bytes();
-    stream.write_all(&len)?;
-    stream.write_all(&payload)?;
-    stream.flush()?;
-    Ok(())
+    write_frame_with_limit(stream, msg, MAX_FRAME).map_err(|e| Error::Ipc(e.to_string()))
 }
 
 fn read_frame(stream: &mut UnixStream) -> Result<Vec<u8>> {
-    let mut len_buf = [0u8; 4];
-    stream.read_exact(&mut len_buf)?;
-    let len = u32::from_le_bytes(len_buf) as usize;
-    if len > MAX_FRAME {
-        return Err(Error::Ipc(format!("frame length {len} too large")));
-    }
-    let mut buf = vec![0u8; len];
-    stream.read_exact(&mut buf)?;
-    Ok(buf)
+    read_frame_bytes(stream, MAX_FRAME).map_err(|e| Error::Ipc(e.to_string()))
 }

src/config.rs

@@ -448,22 +448,25 @@ pub(crate) fn protocol_from_type(type_: &str) -> Option<&'static str> {
 
 /// Load config from `path`, creating a default file if missing.
 pub fn load_or_create(path: &Path) -> Result<Config> {
-    if let Some(parent) = path.parent() {
-        fs::create_dir_all(parent).map_err(|e| Error::io_at(parent, e))?;
-    }
-
-    if !path.exists() {
-        let cfg = Config::default();
-        save(path, &cfg)?;
-        return Ok(cfg);
-    }
-
-    let data = fs::read_to_string(path).map_err(|e| Error::io_at(path, e))?;
-    let cfg: Config = serde_json::from_str(&data)?;
+    use bigfred_shared_daemon::config::Load;
+    let cfg = bigfred_shared_daemon::config::JsonFile::<Config>::new(path)
+        .create_default()
+        .load()
+        .map_err(map_config)?;
     cfg.validate()?;
     Ok(cfg)
 }
 
+fn map_config(e: bigfred_shared_daemon::config::ConfigError) -> Error {
+    match e {
+        bigfred_shared_daemon::config::ConfigError::Io { path, source } => {
+            Error::io_at(PathBuf::from(path), source)
+        }
+        bigfred_shared_daemon::config::ConfigError::Json(j) => Error::Json(j),
+        bigfred_shared_daemon::config::ConfigError::Other(s) => Error::Other(s),
+    }
+}
+
 /// Persist config as pretty JSON.
 pub fn save(path: &Path, cfg: &Config) -> Result<()> {
     if let Some(parent) = path.parent() {

src/config_watch.rs

@@ -1,143 +1,27 @@
 //! Linux inotify-based configuration watcher (no polling).
-//!
-//! Watches the parent directory of the config file for changes, debounces
-//! bursts (atomic write+rename), then signals a reload.
 
 use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc::{self, Receiver, Sender};
+use std::sync::atomic::AtomicBool;
+use std::sync::mpsc::Receiver;
 use std::sync::Arc;
-use std::thread;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
-use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
+use bigfred_shared_daemon::config::{spawn_signal, PathFilter, WatchSpec};
 
 use crate::error::{Error, Result};
 
 const DEBOUNCE: Duration = Duration::from_millis(300);
 
 /// Signal that the configuration file may have changed.
-pub struct ReloadSignal;
+pub type ReloadSignal = bigfred_shared_daemon::config::Reload;
 
 /// Filter path events relevant to the watched config basename.
 pub fn is_relevant_path(path: &Path, config_name: &str) -> bool {
-    let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
-        return false;
-    };
-    if name.starts_with('.') {
-        return false;
-    }
-    if name.ends_with('~') || name.ends_with(".swp") || name.ends_with(".tmp") {
-        return false;
-    }
-    name == config_name
+    bigfred_shared_daemon::config::is_relevant_path(path, &PathFilter::Basename(config_name.to_string()))
 }
 
 /// Spawn an inotify watcher thread. Returns a receiver of debounce-coalesced reload signals.
 pub fn spawn(config_path: PathBuf) -> Result<(Receiver<ReloadSignal>, Arc<AtomicBool>)> {
-    let (tx, rx) = mpsc::channel();
-    let stop = Arc::new(AtomicBool::new(false));
-    let stop_thr = Arc::clone(&stop);
-
-    thread::Builder::new()
-        .name("config-watch".into())
-        .spawn(move || {
-            if let Err(e) = watch_loop(config_path, tx, stop_thr) {
-                log::warn!("config watcher stopped: {e}");
-            }
-        })
-        .map_err(|e| Error::Other(e.to_string()))?;
-
-    Ok((rx, stop))
-}
-
-fn watch_loop(
-    config_path: PathBuf,
-    reload_tx: Sender<ReloadSignal>,
-    stop: Arc<AtomicBool>,
-) -> Result<()> {
-    let config_name = config_path
-        .file_name()
-        .and_then(|s| s.to_str())
-        .unwrap_or("microdns.json")
-        .to_string();
-
-    let watch_dir = config_path
-        .parent()
-        .map(Path::to_path_buf)
-        .unwrap_or_else(|| PathBuf::from("."));
-
-    let (raw_tx, raw_rx) = mpsc::channel();
-
-    let mut watcher = RecommendedWatcher::new(
-        move |res: std::result::Result<Event, notify::Error>| {
-            let _ = raw_tx.send(res);
-        },
-        notify::Config::default(),
-    )
-    .map_err(|e| Error::Other(format!("inotify watcher: {e}")))?;
-
-    if !watch_dir.is_dir() {
-        let _ = std::fs::create_dir_all(&watch_dir);
-    }
-    if watch_dir.is_dir() {
-        watcher
-            .watch(&watch_dir, RecursiveMode::NonRecursive)
-            .map_err(|e| Error::Other(format!("watch {}: {e}", watch_dir.display())))?;
-    } else if let Some(parent) = watch_dir.parent() {
-        if parent.is_dir() {
-            let _ = watcher.watch(parent, RecursiveMode::NonRecursive);
-        }
-    }
-
-    log::info!("config watch active on {}", watch_dir.display());
-
-    let mut pending: Option<Instant> = None;
-
-    loop {
-        if stop.load(Ordering::SeqCst) {
-            break;
-        }
-
-        let timeout = pending
-            .map(|t| {
-                let elapsed = t.elapsed();
-                if elapsed >= DEBOUNCE {
-                    Duration::from_millis(0)
-                } else {
-                    DEBOUNCE - elapsed
-                }
-            })
-            .unwrap_or(Duration::from_secs(1));
-
-        match raw_rx.recv_timeout(timeout) {
-            Ok(Ok(event)) => {
-                let relevant = matches!(
-                    event.kind,
-                    EventKind::Create(_)
-                        | EventKind::Modify(_)
-                        | EventKind::Remove(_)
-                        | EventKind::Any
-                ) && event
-                    .paths
-                    .iter()
-                    .any(|p| is_relevant_path(p, &config_name));
-
-                if relevant {
-                    pending = Some(Instant::now());
-                }
-            }
-            Ok(Err(e)) => {
-                log::warn!("config watch error: {e}");
-            }
-            Err(mpsc::RecvTimeoutError::Timeout) => {
-                if pending.is_some_and(|t| t.elapsed() >= DEBOUNCE) {
-                    pending = None;
-                    let _ = reload_tx.send(ReloadSignal);
-                }
-            }
-            Err(mpsc::RecvTimeoutError::Disconnected) => break,
-        }
-    }
-    Ok(())
+    spawn_signal(vec![WatchSpec::file(config_path)], DEBOUNCE)
+        .map_err(|e| Error::Other(e.to_string()))
 }