diff --git a/Cargo.lock b/Cargo.lock index e59494d..fa4621e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -47,7 +47,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -58,7 +58,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -67,6 +67,20 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "bigfred-shared-daemon" +version = "0.1.0" +source = "git+https://github.com/dcc-bigfred/rust-commons.git?branch=main#5f5053ef9316139c7cde500e6e9e85a70dbbbd09" +dependencies = [ + "log", + "nix", + "notify", + "serde", + "serde_json", + "signal-hook", + "thiserror", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -191,6 +205,16 @@ dependencies = [ "log", ] +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "fastrand" version = "2.5.0" @@ -388,13 +412,13 @@ dependencies = [ name = "microdns" version = "0.1.0" dependencies = [ + "bigfred-shared-daemon", "clap", "env_logger", "libc", "log", "mdns-sd", "nix", - "notify", "serde", "serde_json", "socket2", @@ -579,6 +603,26 @@ dependencies = [ "zmij", ] +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "socket-pktinfo" version = "0.4.1" @@ -597,7 +641,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -691,7 +735,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index ba23bba..7f9b14c 100644 --- a/Cargo.toml +++ b/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" diff --git a/Makefile b/Makefile index 0bdbaae..c93a92b 100644 --- a/Makefile +++ b/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 diff --git a/src/bigfred_watch.rs b/src/bigfred_watch.rs index d9d9d9c..90ff8ee 100644 --- a/src/bigfred_watch.rs +++ b/src/bigfred_watch.rs @@ -4,11 +4,11 @@ //! connection (poll). Success body for `dcc_bus_list` is `{ "programs": [...] }` //! matching REST; errors are `{ "error": "" }`. -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> { } 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> { - 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())) } diff --git a/src/config.rs b/src/config.rs index 397e449..dc296e9 100644 --- a/src/config.rs +++ b/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 { - 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::::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() { diff --git a/src/config_watch.rs b/src/config_watch.rs index d3ff81e..32f1d46 100644 --- a/src/config_watch.rs +++ b/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, Arc)> { - 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, - stop: Arc, -) -> 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| { - 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 = 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())) } diff --git a/src/ctl.rs b/src/ctl.rs index 4255f03..0ec10ff 100644 --- a/src/ctl.rs +++ b/src/ctl.rs @@ -5,15 +5,18 @@ //! (static `services[]` plus dynamic dcc-bus / microinit DNS-SD; not Z21 LAN beacons). use std::collections::HashMap; -use std::io::{Read, Write}; -use std::os::unix::fs::PermissionsExt; -use std::os::unix::net::{UnixListener, UnixStream}; +use std::io::Write; +use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, RwLock}; -use std::thread; use std::time::Duration; +use bigfred_shared_daemon::ipc::{ + read_frame_bytes, write_frame_with_limit, AcceptPolicy, Auth, BindError, BindOptions, Command, + Connection, ErrorHandler, IpcError, RejectReason, Router, SessionMode, +}; use serde::{Deserialize, Serialize}; +use serde_json::Value; use crate::config::ServiceEntry; use crate::datadir; @@ -93,12 +96,6 @@ pub struct ServicesListBody { pub services: Vec, } -#[derive(Debug, Deserialize)] -struct Request { - #[serde(rename = "type")] - type_: String, -} - #[derive(Debug, Serialize, Deserialize)] struct ErrorBody { error: String, @@ -135,32 +132,11 @@ pub fn listed_services(ads: &DesiredAds) -> Vec { } pub fn write_frame_to(writer: &mut impl Write, 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(); - writer.write_all(&len)?; - writer.write_all(&payload)?; - writer.flush()?; - Ok(()) + write_frame_with_limit(writer, msg, MAX_FRAME).map_err(|e| Error::Ipc(e.to_string())) } -pub fn read_frame_from(reader: &mut impl Read) -> Result> { - let mut len_buf = [0u8; 4]; - reader.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]; - reader.read_exact(&mut buf)?; - Ok(buf) +pub fn read_frame_from(reader: &mut impl std::io::Read) -> Result> { + read_frame_bytes(reader, MAX_FRAME).map_err(|e| Error::Ipc(e.to_string())) } pub fn write_frame(stream: &mut UnixStream, msg: &impl Serialize) -> Result<()> { @@ -171,54 +147,98 @@ pub fn read_frame(stream: &mut UnixStream) -> Result> { read_frame_from(stream) } -/// Bind the control socket without stealing a live daemon's inode. -/// -/// 1. `connect` — if a peer answers, refuse (`already running`); do not unlink. -/// 2. `NotFound` / `ConnectionRefused` — leftover inode (or nothing) → unlink, then bind. -/// 3. Any other connect error is returned as-is (do not unlink a mystery path). -fn bind_singleton(socket_path: &Path) -> Result { - match UnixStream::connect(socket_path) { - Ok(stream) => { - let pid = peer_pid(&stream); - let where_ = if pid != 0 { - format!("{} (pid {pid})", socket_path.display()) - } else { - socket_path.display().to_string() - }; - return Err(Error::Ipc(format!("microdns already running at {where_}"))); - } - Err(e) if is_stale_socket_connect_error(&e) => {} - Err(e) => return Err(Error::io_at(socket_path, e)), +fn map_bind(e: BindError) -> Error { + match e { + BindError::AlreadyRunning { + process_name, + location, + .. + } => Error::Ipc(format!("{process_name} already running at {location}")), + BindError::Io { path, source } => Error::io_at(path, source), + } +} + +struct CtlState { + snapshot: Arc>, + runtime: Option, +} + +struct ServicesListCmd; +struct DoctorCmd; + +impl Command for ServicesListCmd { + fn name(&self) -> &'static str { + "services_list" } - match std::fs::remove_file(socket_path) { - Ok(()) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(Error::io_at(socket_path, e)), + fn execute( + &self, + state: &CtlState, + _body: Value, + conn: &mut Connection, + ) -> std::result::Result<(), IpcError> { + let ads = match state.snapshot.read() { + Ok(g) => g.clone(), + Err(_) => { + let _ = conn.reply(&ErrorBody { + error: "internal_error".into(), + }); + return Ok(()); + } + }; + conn.reply(&ServicesListBody { + services: listed_services(&ads), + }) + .map_err(IpcError::from) } - UnixListener::bind(socket_path).map_err(|e| Error::io_at(socket_path, e)) } -fn is_stale_socket_connect_error(err: &std::io::Error) -> bool { - matches!( - err.kind(), - std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused - ) +impl Command for DoctorCmd { + fn name(&self) -> &'static str { + "doctor" + } + fn execute( + &self, + state: &CtlState, + _body: Value, + conn: &mut Connection, + ) -> std::result::Result<(), IpcError> { + match build_daemon_doctor(&state.snapshot, state.runtime.as_ref()) { + Ok(body) => conn.reply(&body).map_err(IpcError::from), + Err(e) => { + let _ = conn.reply(&ErrorBody { + error: e.to_string(), + }); + Ok(()) + } + } + } } -fn peer_pid(stream: &UnixStream) -> u32 { - use nix::sys::socket::{getsockopt, sockopt::PeerCredentials}; - getsockopt(stream, PeerCredentials) - .map(|c| c.pid() as u32) - .unwrap_or(0) +struct CtlHooks; + +impl ErrorHandler for CtlHooks { + fn unknown(&self, _state: &CtlState, _type_name: &str, _body: &Value, conn: &mut Connection) { + let _ = conn.reply(&ErrorBody { + error: "invalid_request".into(), + }); + } + fn error(&self, _state: &CtlState, _err: &IpcError, conn: &mut Connection) { + let _ = conn.reply(&ErrorBody { + error: "invalid_request".into(), + }); + } + fn reject(&self, _state: &CtlState, _reason: RejectReason, _conn: &mut Connection) {} } -fn apply_socket_perms(socket_path: &Path) -> Result<()> { - let mut perms = std::fs::metadata(socket_path) - .map_err(|e| Error::io_at(socket_path, e))? - .permissions(); - perms.set_mode(0o600); - std::fs::set_permissions(socket_path, perms).map_err(|e| Error::io_at(socket_path, e))?; - Ok(()) +fn ctl_router() -> std::result::Result, Error> { + let mut router = Router::new(); + router + .add(ServicesListCmd) + .map_err(|e| Error::Other(e.to_string()))?; + router + .add(DoctorCmd) + .map_err(|e| Error::Other(e.to_string()))?; + Ok(router) } /// Bind `$DATA_DIR/run/microdns.sock` (or `path`) and serve `services_list` in a @@ -233,104 +253,29 @@ pub fn serve_with_runtime( snapshot: Arc>, runtime: Option, ) -> Result<()> { - if let Some(parent) = path.parent() { - if !parent.as_os_str().is_empty() { - std::fs::create_dir_all(parent).map_err(|e| Error::io_at(parent, e))?; - } - } - let listener = bind_singleton(path)?; - apply_socket_perms(path)?; + let state = Arc::new(CtlState { snapshot, runtime }); + bigfred_shared_daemon::ipc::serve_background( + BindOptions { + path: path.to_path_buf(), + mode: 0o600, + chown: None, + process_name: "microdns", + }, + AcceptPolicy { + auth: Auth::None, + session: SessionMode::OneShot, + max_clients: None, + max_frame: MAX_FRAME, + }, + ctl_router()?, + CtlHooks, + state, + ) + .map_err(map_bind)?; log::info!("ctl listening on {}", path.display()); - - let path = path.to_path_buf(); - thread::Builder::new() - .name("ctl".into()) - .spawn(move || { - for conn in listener.incoming() { - match conn { - Ok(stream) => { - let snap = Arc::clone(&snapshot); - let rt = runtime.clone(); - thread::spawn(move || handle_conn(stream, snap, rt)); - } - Err(_) => { - if !path.exists() { - break; - } - } - } - } - }) - .map_err(|e| Error::Ipc(format!("ctl thread: {e}")))?; Ok(()) } -fn handle_conn( - mut stream: UnixStream, - snapshot: Arc>, - runtime: Option, -) { - let raw = match read_frame(&mut stream) { - Ok(b) => b, - Err(_) => return, - }; - let req: Request = match serde_json::from_slice(&raw) { - Ok(r) => r, - Err(_) => { - let _ = write_frame( - &mut stream, - &ErrorBody { - error: "invalid_request".into(), - }, - ); - return; - } - }; - match req.type_.as_str() { - "services_list" => { - let ads = match snapshot.read() { - Ok(g) => g.clone(), - Err(_) => { - let _ = write_frame( - &mut stream, - &ErrorBody { - error: "internal_error".into(), - }, - ); - return; - } - }; - let body = ServicesListBody { - services: listed_services(&ads), - }; - let _ = write_frame(&mut stream, &body); - } - "doctor" => { - let body = match build_daemon_doctor(&snapshot, runtime.as_ref()) { - Ok(b) => b, - Err(e) => { - let _ = write_frame( - &mut stream, - &ErrorBody { - error: e.to_string(), - }, - ); - return; - } - }; - let _ = write_frame(&mut stream, &body); - } - _ => { - let _ = write_frame( - &mut stream, - &ErrorBody { - error: "invalid_request".into(), - }, - ); - } - } -} - fn build_daemon_doctor( snapshot: &RwLock, runtime: Option<&CtlRuntime>, diff --git a/src/datadir.rs b/src/datadir.rs index 3abbae3..0029568 100644 --- a/src/datadir.rs +++ b/src/datadir.rs @@ -1,56 +1,5 @@ //! Persistent data root resolution. //! //! Priority: `DATA_DIR` (absolute only), then `/data` (hub default). -//! Relative values are ignored so misconfiguration cannot silently redirect -//! data under the process working directory. -use std::path::{Path, PathBuf}; - -/// Env var for the persistent data root. -pub const ENV_DATA_DIR: &str = "DATA_DIR"; -/// Hub image default. -pub const DEFAULT_ROOT: &str = "/data"; - -/// Returns the persistent data directory. -#[must_use] -pub fn root() -> PathBuf { - if let Some(v) = root_from_env(ENV_DATA_DIR) { - return v; - } - PathBuf::from(DEFAULT_ROOT) -} - -fn root_from_env(name: &str) -> Option { - let v = std::env::var_os(name)?; - if v.is_empty() { - return None; - } - let p = PathBuf::from(v); - if p.is_absolute() { - Some(p) - } else { - None - } -} - -/// Override `DATA_DIR` for this process (absolute paths only). -pub fn set_root(path: impl AsRef) { - let p = path.as_ref(); - if p.is_absolute() { - std::env::set_var(ENV_DATA_DIR, p.as_os_str()); - } -} - -/// Join `parts` under [`root`]. -#[must_use] -pub fn path(parts: I) -> PathBuf -where - I: IntoIterator, - P: AsRef, -{ - let mut out = root(); - for part in parts { - out.push(part); - } - out -} +pub use bigfred_shared_daemon::datadir::{path, root, set_root, DEFAULT_ROOT, ENV_DATA_DIR}; diff --git a/src/microinit_watch.rs b/src/microinit_watch.rs index d66671a..73f56cf 100644 --- a/src/microinit_watch.rs +++ b/src/microinit_watch.rs @@ -5,7 +5,6 @@ //! advertised set. Reconnects with backoff; the caller keeps last-good ads. use std::collections::{BTreeMap, HashMap, HashSet}; -use std::io::{Read, Write}; use std::os::unix::net::UnixStream; use std::path::Path; use std::sync::atomic::{AtomicBool, Ordering}; @@ -14,6 +13,7 @@ use std::sync::{Arc, Mutex, RwLock}; use std::thread; use std::time::Duration; +use bigfred_shared_daemon::ipc::{read_frame_bytes, write_frame_with_limit}; use serde::{Deserialize, Serialize}; use crate::config::{self, Config, ServiceEntry}; @@ -330,30 +330,9 @@ fn txt_from_labels(labels: &BTreeMap) -> Option 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> { - 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())) } diff --git a/src/run.rs b/src/run.rs index 68ae8c6..0dfabb9 100644 --- a/src/run.rs +++ b/src/run.rs @@ -624,7 +624,7 @@ fn reload_loop( ) { while !stop.load(Ordering::SeqCst) && !signals::shutdown_requested() { match rx.recv_timeout(Duration::from_secs(1)) { - Ok(ReloadSignal) => match config::load_or_create(&shared.config_path) { + Ok(_) => match config::load_or_create(&shared.config_path) { Ok(cfg) => { // load_or_create already validates; keep last-known-good on failure. if let Ok(mut w) = shared.config.write() {