From 457a966068b6d86f3e1b27c0a9910bf4e00c7a68 Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 6 Nov 2026 20:00:29 +0000 Subject: [PATCH] Isolate watcher queues and selectively reload resolved service binaries --- host/src/process/watcher.rs | 21 +++--- host/tests/init.rs | 72 ++++++++++++++++++- userland/Cargo.lock | 1 + userland/init/README.md | 12 +++- userland/init/src/lib.rs | 7 +- userland/init/src/main.rs | 114 +++++++++++++++++++++++++------ userland/init/src/systemd-run.rs | 7 +- userland/spawn/Cargo.toml | 1 + userland/spawn/src/lib.rs | 4 +- vfs/src/observe.rs | 52 ++++++++++++-- 10 files changed, 247 insertions(+), 44 deletions(-) diff --git a/host/src/process/watcher.rs b/host/src/process/watcher.rs index 5ec2408..88c8963 100644 --- a/host/src/process/watcher.rs +++ b/host/src/process/watcher.rs @@ -78,17 +78,18 @@ impl HostWatcher for State { .root .observer() .ok_or_else(|| wasmtime::format_err!("filesystem has no observer"))?; + let paths: Vec = paths + .iter() + .map(|p| { + sausage_vfs::resolve("/", p) + .trim_end_matches('/') + .to_owned() + }) + .map(|p| if p.is_empty() { "/".into() } else { p }) + .collect(); Ok(self.table.push(Watcher { - paths: paths - .iter() - .map(|p| { - sausage_vfs::resolve("/", p) - .trim_end_matches('/') - .to_owned() - }) - .map(|p| if p.is_empty() { "/".into() } else { p }) - .collect(), - events: observer.subscribe(), + events: observer.subscribe(paths.clone()), + paths, pending: Vec::new(), })?) } diff --git a/host/tests/init.rs b/host/tests/init.rs index db9b111..844320d 100644 --- a/host/tests/init.rs +++ b/host/tests/init.rs @@ -154,6 +154,74 @@ fn lines(path: &Path) -> Vec { .map(str::to_owned) .collect() } +#[tokio::test] +async fn unrelated_bursts_and_watched_overflow_only_reload_changed_units() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + wait(|| path.join("var/log/journal").exists()).await; + for name in ["changed", "unchanged"] { + write( + &root, + &format!("/run/systemd/transient/{name}.service"), + format!("[Service]\nExecStart=probe server /{name}.log\n").as_bytes(), + ) + .await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", name]) + .await + .0, + 0 + ); + wait(|| lines(&path.join(format!("{name}.log"))) == ["start"]).await; + } + tokio::task::unconstrained(async { + for _ in 0..4000 { + write(&root, "/run/unrelated", b"noise").await; + } + }) + .await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", "unchanged"]) + .await + .0, + 0 + ); + assert_eq!(lines(&path.join("changed.log")), ["start"]); + assert_eq!(lines(&path.join("unchanged.log")), ["start"]); + // This single-thread executor cannot schedule init during ready tmpfs writes. + // The watched queue therefore overflows before init can consume the edits. + tokio::task::unconstrained(async { + write( + &root, + "/run/systemd/transient/changed.service", + b"[Service]\nExecStart=probe server /changed.log\nEnvironment=REV=2\n", + ) + .await; + for _ in 0..1000 { + write(&root, "/run/systemd/transient/noise", b"noise").await; + } + }) + .await; + wait(|| lines(&path.join("changed.log")) == ["start", "stop", "start"]).await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", "unchanged"]) + .await + .0, + 0 + ); + assert_eq!(lines(&path.join("unchanged.log")), ["start"]); + // Bare names also watch the resolved executable, not the literal name. + let mut bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap(); + bytes.extend_from_slice(&[0, 2, 1, b'x']); + write(&root, "/bin/replacement.wasm", &bytes).await; + root.rename("/bin/replacement.wasm", "/bin/probe.wasm") + .await + .unwrap(); + wait(|| lines(&path.join("unchanged.log")) == ["start", "stop", "start"]).await; + supervisor.shutdown().await; +} #[tokio::test(flavor = "multi_thread")] async fn watcher_and_child_exit_pollable() { let _serial = SERIAL.lock().await; @@ -276,7 +344,9 @@ async fn live_unit_changes_and_binary_replacement() { ) .await; wait(|| lines(&path.join("live.log")) == ["start", "stop", "start"]).await; - let bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap(); + let mut bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap(); + // Add a valid custom section, so the replacement differs in content. + bytes.extend_from_slice(&[0, 2, 1, b'x']); write(&root, "/bin/replacement.wasm", &bytes).await; root.rename("/bin/replacement.wasm", "/bin/probe.wasm") .await diff --git a/userland/Cargo.lock b/userland/Cargo.lock index f06325f..1ec69bc 100644 --- a/userland/Cargo.lock +++ b/userland/Cargo.lock @@ -4074,6 +4074,7 @@ dependencies = [ name = "sausage-spawn" version = "0.1.0" dependencies = [ + "wasi 0.14.7+wasi-0.2.4", "wit-bindgen 0.46.0", ] diff --git a/userland/init/README.md b/userland/init/README.md index be2a090..6cabd5e 100644 --- a/userland/init/README.md +++ b/userland/init/README.md @@ -16,7 +16,14 @@ When init receives EOF, it stops dependents before prerequisites. Enable records are files under `default.target.wants` or `multi-user.target.wants` in the persistent unit directory. The `[Install]` section supplies `WantedBy` for `systemctl enable`. Init observes VFS changes and applies unit edits without `daemon-reload`. -A replacement of a running unit's executable restarts that unit. +A change to a running unit's executable content restarts that unit. +Init resolves bare executable names through `PATH`, which defaults to `/usr/local/bin:/bin`. +Each directory tries the bare name before the `.wasm` suffix. +Init watches those candidate paths and compares the resolved executable's content hash. +After an overflow, init rescans definitions and executable hashes. +It restarts only units with a changed definition or executable. +Unrelated world writes do not enter these watcher queues. +Init and request clients block on pollables, with timers for lifecycle deadlines. ## Control protocol @@ -71,5 +78,6 @@ The cap is 1 MiB. Init keeps one rotated file with the `.1` suffix. A watcher matches each supplied path and all its descendants. Its pollable remains ready while changes remain pending. `changes()` returns and clears the pending changes. -The queue holds at most 256 changes. An `overflow` event requires a rescan. +Each watcher filters paths before its independent queue holds at most 256 changes. +An `overflow` event requires a rescan. Observers report successful VFS mutations, not direct host filesystem writes outside the VFS. diff --git a/userland/init/src/lib.rs b/userland/init/src/lib.rs index 8c53313..2478668 100644 --- a/userland/init/src/lib.rs +++ b/userland/init/src/lib.rs @@ -131,6 +131,10 @@ pub fn request(action: &str, unit: &str) -> io::Result { })?, )?; let deadline = std::time::Instant::now() + Duration::from_secs(60); + let timeout = wasi::clocks::monotonic_clock::subscribe_duration( + Duration::from_secs(60).as_nanos() as u64, + ); + let changed = watcher.subscribe(); let mut acknowledged = false; loop { if !acknowledged && let Ok(bytes) = fs::read(&reply) { @@ -167,9 +171,8 @@ pub fn request(action: &str, unit: &str) -> io::Result { "init did not reply", )); } - // The short timer also handles a missing or failed init. + sausage_spawn::poll(&[&changed, &timeout]); let _ = watcher.changes(); - std::thread::sleep(Duration::from_millis(20)); } } #[cfg(test)] diff --git a/userland/init/src/main.rs b/userland/init/src/main.rs index 7a49b80..4a63394 100644 --- a/userland/init/src/main.rs +++ b/userland/init/src/main.rs @@ -5,9 +5,67 @@ use std::{ collections::{BTreeMap, BTreeSet, VecDeque}, fs, io::{self, Write}, - time::{Duration, Instant}, + time::Instant, }; +fn program_paths(unit: &Unit) -> Vec { + let program = shlex::split(unit.get("Service.ExecStart")) + .and_then(|args| args.first().cloned()) + .unwrap_or_default(); + if program.contains('/') { + return vec![program]; + } + let path = shlex::split(unit.get("Service.Environment")) + .unwrap_or_default() + .into_iter() + .find_map(|v| v.strip_prefix("PATH=").map(str::to_owned)) + .unwrap_or_else(|| "/usr/local/bin:/bin".into()); + path.split(':') + .flat_map(|dir| [format!("{dir}/{program}"), format!("{dir}/{program}.wasm")]) + .collect() +} +fn binary(unit: &Unit) -> Option<(String, u64)> { + use std::hash::{Hash, Hasher}; + program_paths(unit).into_iter().find_map(|path| { + let bytes = fs::read(&path).ok()?; + let mut hash = std::collections::hash_map::DefaultHasher::new(); + bytes.hash(&mut hash); + Some((path, hash.finish())) + }) +} +fn wait_events( + services: &BTreeMap, + mut pollables: Vec, + tick_at: Instant, +) { + let now = Instant::now(); + let mut deadline = None; + for service in services.values() { + if let Some(child) = &service.child { + if child.try_wait().is_none() { + pollables.push(child.subscribe()); + } + for stream in [&child.stdout, &child.stderr].into_iter().flatten() { + pollables.push(stream.subscribe()); + } + } + let time = service + .stopping + .or_else(|| (service.desired && service.child.is_none()).then_some(service.next)); + if let Some(time) = time.filter(|time| *time > tick_at) { + deadline = Some(deadline.map_or(time, |old: Instant| old.min(time))); + } + } + if let Some(deadline) = deadline { + pollables.push(wasi::clocks::monotonic_clock::subscribe_duration( + deadline.saturating_duration_since(now).as_nanos() as u64, + )); + } + if !pollables.is_empty() { + sausage_spawn::poll(&pollables.iter().collect::>()); + } +} struct Service { + binary: Option<(String, u64)>, unit: Unit, child: Option, state: String, @@ -20,6 +78,7 @@ struct Service { impl Service { fn new(unit: Unit) -> Self { Self { + binary: binary(&unit), unit, child: None, state: "inactive".into(), @@ -68,7 +127,12 @@ impl Service { self.desired = false; return; }; - let mut command = Command::new(&argv[0]); + let program = self + .binary + .as_ref() + .map(|(path, _)| path.as_str()) + .unwrap_or(&argv[0]); + let mut command = Command::new(program); command .args(&argv[1..]) .new_group() @@ -244,7 +308,9 @@ fn scan(services: &mut BTreeMap) { let service = services .entry(name.clone()) .or_insert_with(|| Service::new(unit.clone())); - if service.unit != unit { + let binary = binary(&unit); + if service.unit != unit || service.binary != binary { + service.binary = binary; let desired = service.desired || service.state == "active"; service.stop(); service.unit = unit; @@ -326,7 +392,8 @@ fn main() -> io::Result<()> { ] { fs::create_dir_all(dir)?; } - let watcher = sausage_spawn::Watcher::new(&["/".into()]); + let mut paths = vec![UNITS.into(), TRANSIENT.into(), format!("{RUN}/requests")]; + let mut watcher = sausage_spawn::Watcher::new(&paths); let stdin = wasi::cli::stdin::get_stdin(); let mut services = BTreeMap::new(); scan(&mut services); @@ -336,23 +403,10 @@ fn main() -> io::Result<()> { } let changes = watcher.changes(); if !changes.is_empty() { - for service in services.values_mut() { - let program = shlex::split(service.unit.get("Service.ExecStart")) - .and_then(|a| a.first().cloned()) - .unwrap_or_default(); - if service.child.is_some() - && changes.iter().any(|c| { - c.path == program || matches!(c.kind, sausage_spawn::ChangeKind::Overflow) - }) - { - service.stop(); - service.desired = true; - service.next = Instant::now(); - } - } scan(&mut services); } controls(&mut services); + let tick_at = Instant::now(); let names: Vec<_> = services.keys().cloned().collect(); for name in &names { services.get_mut(name).unwrap().tick(name); @@ -388,7 +442,24 @@ fn main() -> io::Result<()> { for (name, service) in &services { service.publish(name); } - std::thread::sleep(Duration::from_millis(20)); + let mut next_paths = vec![UNITS.into(), TRANSIENT.into(), format!("{RUN}/requests")]; + for service in services.values() { + next_paths.extend(program_paths(&service.unit)); + } + next_paths.sort(); + next_paths.dedup(); + if paths != next_paths { + paths = next_paths; + watcher = sausage_spawn::Watcher::new(&paths); + // Close the snapshot/subscription race before blocking. + scan(&mut services); + continue; + } + wait_events( + &services, + vec![watcher.subscribe(), stdin.subscribe()], + tick_at, + ); } // Stop dependents before their prerequisites. A cycle stops together. while services.values().any(|s| s.child.is_some()) { @@ -409,13 +480,16 @@ fn main() -> io::Result<()> { if leaves.is_empty() { leaves = live; } + let tick_at = Instant::now(); for name in leaves { let service = services.get_mut(&name).unwrap(); service.stop(); service.tick(&name); service.publish(&name); } - std::thread::sleep(Duration::from_millis(20)); + if services.values().any(|s| s.child.is_some()) { + wait_events(&services, Vec::new(), tick_at); + } } Ok(()) } diff --git a/userland/init/src/systemd-run.rs b/userland/init/src/systemd-run.rs index 968face..172d8a8 100644 --- a/userland/init/src/systemd-run.rs +++ b/userland/init/src/systemd-run.rs @@ -1,6 +1,6 @@ //! Creates transient unit files and optionally waits for their final state. use sausage_init::*; -use std::{fs, io, time::Duration}; +use std::{fs, io}; fn run() -> io::Result { let mut args = std::env::args().skip(1).peekable(); let mut unit = format!( @@ -65,6 +65,8 @@ fn run() -> io::Result { return Ok(reply.code); } if wait { + let watcher = sausage_spawn::Watcher::new(&[format!("{RUN}/state/{unit}")]); + let changed = watcher.subscribe(); loop { if let Ok(bytes) = fs::read(format!("{RUN}/state/{unit}")) { let status: Status = serde_json::from_slice(&bytes)?; @@ -72,7 +74,8 @@ fn run() -> io::Result { return Ok(code); } } - std::thread::sleep(Duration::from_millis(20)); + changed.block(); + let _ = watcher.changes(); } } Ok(0) diff --git a/userland/spawn/Cargo.toml b/userland/spawn/Cargo.toml index 794a6a2..92e83db 100644 --- a/userland/spawn/Cargo.toml +++ b/userland/spawn/Cargo.toml @@ -5,3 +5,4 @@ edition = "2024" [dependencies] wit-bindgen = "=0.46.0" +wasi = "0.14" diff --git a/userland/spawn/src/lib.rs b/userland/spawn/src/lib.rs index a8b4020..772cb72 100644 --- a/userland/spawn/src/lib.rs +++ b/userland/spawn/src/lib.rs @@ -3,12 +3,12 @@ //! Pipe callers must drain stdout and stderr before wait to avoid bounded-pipe deadlocks. use std::{collections::BTreeMap, io, path::{Path, PathBuf}}; mod bindings { - wit_bindgen::generate!({ path: "../../host/wit", world: "program", generate_all }); + wit_bindgen::generate!({ path: "../../host/wit", world: "program", generate_all, with: { "wasi:io/poll@0.2.12": ::wasi::io::poll } }); } use bindings::sausage::process::host; pub use bindings::wasi::io::streams::{InputStream, OutputStream}; pub use host::{Stdio, Watcher, Change, ChangeKind}; -pub use bindings::wasi::io::poll::{Pollable, poll}; +pub use wasi::io::poll::{Pollable, poll}; /// Hard-stop a process group by a live process ID. pub fn kill(pid: u32) -> io::Result<()> { host::kill(pid).map_err(io::Error::other) } diff --git a/vfs/src/observe.rs b/vfs/src/observe.rs index 1738164..4a5008b 100644 --- a/vfs/src/observe.rs +++ b/vfs/src/observe.rs @@ -23,12 +23,16 @@ enum Held { type TrackedPath = Mutex>; type Path = Arc; type Callback = dyn Fn(&str) + Send + Sync; +type Subscription = ( + Vec, + tokio::sync::broadcast::Sender<(String, ChangeKind)>, +); /// Shared observation for mount namespaces. Paths supplied to the callback are virtual paths. /// Callbacks run after filesystem locks have been released; they must not block. #[derive(Clone)] pub struct Observer { changed: Arc, - events: tokio::sync::broadcast::Sender<(String, ChangeKind)>, + events: Arc>>, paths: Arc>>>, // Only open-registration versus rename/remove needs ordering. Lock the affected // subtrees, not all IO: a blocked open must not stall unrelated namespaces. @@ -38,18 +42,31 @@ impl Observer { pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self { Self { changed: Arc::new(changed), - events: tokio::sync::broadcast::channel(256).0, + events: Arc::default(), paths: Arc::default(), namespace: Arc::default(), } } /// Subscribe before the first mutation of interest. Lag means callers must rescan. - pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<(String, ChangeKind)> { - self.events.subscribe() + pub fn subscribe( + &self, + paths: Vec, + ) -> tokio::sync::broadcast::Receiver<(String, ChangeKind)> { + let (sender, receiver) = tokio::sync::broadcast::channel(256); + self.events.lock().unwrap().push((paths, sender)); + receiver } fn notify(&self, path: &str, kind: ChangeKind) { (self.changed)(path); - let _ = self.events.send((path.to_owned(), kind)); + self.events.lock().unwrap().retain(|(paths, sender)| { + if sender.receiver_count() == 0 { + return false; + } + if paths.iter().any(|base| under(path, base)) { + let _ = sender.send((path.to_owned(), kind)); + } + true + }); } /// Wrap a layer at its canonical namespace prefix (e.g. `/run`). pub fn layer(&self, layer: Arc, prefix: &str, process: Process) -> Arc { @@ -334,6 +351,31 @@ impl File for Handle { #[cfg(test)] mod tests { + #[test] + fn queues_filter_before_the_bound_and_overflow_independently() { + let observer = super::Observer::new(|_| {}); + let mut a = observer.subscribe(vec!["/watched/a".into()]); + let mut b = observer.subscribe(vec!["/watched/b".into()]); + for _ in 0..4000 { + observer.notify("/elsewhere", super::ChangeKind::Modified); + } + assert!(matches!( + a.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + for _ in 0..300 { + observer.notify("/watched/a", super::ChangeKind::Modified); + } + assert!(matches!( + a.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) + )); + assert!(matches!( + b.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + } + use super::*; use tokio::sync::mpsc; #[test]