diff --git a/host/src/harness/mod.rs b/host/src/harness/mod.rs index b95d1cf..357b9e4 100644 --- a/host/src/harness/mod.rs +++ b/host/src/harness/mod.rs @@ -208,7 +208,6 @@ pub async fn run_with_userland(path: &Path, userland: Option<&Path>) -> anyhow:: })?; let services = crate::process::supervise(runtime.clone()); let result = run_with_exec(world, Arc::new(runtime)).await; - services.abort(); - let _ = services.await; + services.shutdown().await; result } diff --git a/host/src/process/mod.rs b/host/src/process/mod.rs index 187c423..04858c5 100644 --- a/host/src/process/mod.rs +++ b/host/src/process/mod.rs @@ -41,11 +41,14 @@ mod bindings { with: { "wasi:io": wasmtime_wasi::p2::bindings::io, "sausage:process/host.child": super::Child, + "sausage:process/host.watcher": super::Watcher, }, }); } use bindings::sausage::process::host::{Host, HostChild, Request, Stdio}; mod services; +mod watcher; +pub use watcher::Watcher; mod stdio; pub use services::supervise; use stdio::Io; @@ -470,7 +473,23 @@ impl Host for State { Ok(self.runtime.kill_pid(pid).map_err(|e| e.to_string())) } } +#[async_trait::async_trait] +impl wasmtime_wasi::p2::Pollable for Child { + async fn ready(&mut self) { + while self.code.borrow_and_update().is_none() { + if self.code.changed().await.is_err() { + break; + } + } + } +} impl HostChild for State { + async fn subscribe( + &mut self, + child: Resource, + ) -> wasmtime::Result> { + wasmtime_wasi::p2::subscribe(&mut self.table, child) + } async fn pid(&mut self, child: Resource) -> wasmtime::Result { Ok(self.table.get(&child)?.pid) } diff --git a/host/src/process/services.rs b/host/src/process/services.rs index f61e065..147ae8c 100644 --- a/host/src/process/services.rs +++ b/host/src/process/services.rs @@ -1,49 +1,28 @@ -//! Supervises the world's services list, live edits, crash limits, and graceful stops. -use super::{GroupGuard, Request, Runtime, Started, Stdio}; -use serde::Deserialize; -use std::{ - collections::{BTreeMap, VecDeque}, - sync::Arc, - time::Duration, -}; -use tokio::time::Instant; - -#[derive(Clone, Deserialize, PartialEq, Eq)] -struct Spec { - name: String, - argv: Vec, - #[serde(default = "root")] - cwd: String, - #[serde(default)] - env: BTreeMap, +//! Starts world init, reaps its command tree, and applies host restart and shutdown deadlines. +use super::{GroupGuard, Request, Runtime, Stdio}; +use std::{sync::Arc, time::Duration}; +pub struct Supervisor { + shutdown: tokio::sync::oneshot::Sender<()>, + task: tokio::task::JoinHandle<()>, } -fn root() -> String { - "/".into() +impl Supervisor { + pub async fn shutdown(self) { + let _ = self.shutdown.send(()); + let _ = self.task.await; + } } -struct Live { - child: Started, - _guard: GroupGuard, -} -struct Service { - spec: Spec, - live: Option, - crashes: VecDeque, - stopped: bool, -} -impl Service { - async fn start(&mut self, runtime: &Arc) { - match runtime - .start( +/// The host knows only init. Unit policy belongs to the WASI program. +pub fn supervise(runtime: Arc) -> Supervisor { + let (shutdown, mut stop) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(async move { + let mut backoff = Duration::from_secs(1); + loop { + let start = runtime.start( Request { - program: self.spec.argv.first().cloned().unwrap_or_default(), - argv: self.spec.argv.clone(), - cwd: self.spec.cwd.clone(), - env: self - .spec - .env - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect(), + program: "/bin/init.wasm".into(), + argv: vec!["/bin/init.wasm".into()], + cwd: "/".into(), + env: Vec::new(), group: None, stdin: Stdio::Pipe, stdout: Stdio::Null, @@ -54,116 +33,56 @@ impl Service { 0, None, None, - ) - .await - { - Ok(child) => { + ); + let child = tokio::select! { _ = &mut stop => return, child = start => child }; + if let Ok(mut child) = child { let guard = GroupGuard { runtime: runtime.clone(), group: child.pid, }; - self.live = Some(Live { - child, - _guard: guard, - }); + let exiting = async { + while child.code.borrow_and_update().is_none() { + if child.code.changed().await.is_err() { + break; + } + } + }; + let shutdown = tokio::select! { _ = &mut stop => true, _ = exiting => false }; + if shutdown { + drop(child.input.take()); + let _ = tokio::time::timeout(Duration::from_secs(15), async { + while child.code.borrow_and_update().is_none() { + if child.code.changed().await.is_err() { + break; + } + } + }) + .await; + drop(guard); + reaped(&runtime, child.pid).await; + return; + } + // This guard kills all descendants, including separate process groups. + drop(guard); + reaped(&runtime, child.pid).await; } - Err(_) => self.crashed(), + tokio::select! { _ = &mut stop => return, _ = tokio::time::sleep(backoff) => () } + backoff = (backoff * 2).min(Duration::from_secs(30)); } - } - fn crashed(&mut self) { - let now = Instant::now(); - while self - .crashes - .front() - .is_some_and(|t| now.duration_since(*t) >= Duration::from_secs(10)) + }); + Supervisor { shutdown, task } +} +async fn reaped(runtime: &Runtime, command: u32) { + loop { + if !runtime + .entries + .lock() + .unwrap() + .values() + .any(|entry| entry.command == command) { - self.crashes.pop_front(); - } - self.crashes.push_back(now); - self.stopped = self.crashes.len() > 5; - } - async fn stop(&mut self) { - if let Some(mut live) = self.live.take() { - drop(live.child.input.take()); - let _ = tokio::time::timeout(Duration::from_secs(5), async { - loop { - if live.child.code.borrow().is_some() { - break; - } - if live.child.code.changed().await.is_err() { - break; - } - } - }) - .await; - // The guard stops remaining descendants, or the process after its grace period. + return; } + tokio::task::yield_now().await; } } -/// Start the supervisor. Aborting its task stops its service processes. -/// The YAML file contains name, argv, optional cwd, and an optional map of environment variables. -pub fn supervise(runtime: Arc) -> tokio::task::JoinHandle<()> { - tokio::spawn(async move { - let mut services: BTreeMap = BTreeMap::new(); - let mut previous = Vec::new(); - let mut interval = tokio::time::interval(Duration::from_millis(100)); - loop { - interval.tick().await; - let content = match runtime - .world - .root - .open("/config/services.yaml", Default::default()) - .await - { - Ok(file) => match file.stat().await { - Ok(stat) => file.read(0, stat.size as usize).await.ok(), - Err(_) => None, - }, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Some(b"[]".to_vec()), - Err(_) => None, - }; - if let Some(content) = content.filter(|c| *c != previous) - && let Ok(specs) = serde_yaml_ng::from_slice::>(&content) - { - let specs: BTreeMap<_, _> = - specs.into_iter().map(|s| (s.name.clone(), s)).collect(); - let removed: Vec<_> = services - .keys() - .filter(|name| !specs.contains_key(*name)) - .cloned() - .collect(); - for name in removed { - services.remove(&name).unwrap().stop().await; - } - for (name, spec) in specs { - let service = services.entry(name).or_insert_with(|| Service { - spec: spec.clone(), - live: None, - crashes: VecDeque::new(), - stopped: false, - }); - if service.spec != spec { - service.stop().await; - } - service.spec = spec; - service.crashes.clear(); - service.stopped = false; - } - previous = content; - } - for service in services.values_mut() { - if service - .live - .as_ref() - .is_some_and(|live| live.child.code.borrow().is_some()) - { - service.live = None; - service.crashed(); - } - if service.live.is_none() && !service.stopped { - service.start(&runtime).await; - } - } - } - }) -} diff --git a/host/src/process/watcher.rs b/host/src/process/watcher.rs new file mode 100644 index 0000000..5ec2408 --- /dev/null +++ b/host/src/process/watcher.rs @@ -0,0 +1,110 @@ +//! Adapts VFS observation to bounded, pollable world-path change queues. +use super::{ + State, + bindings::sausage::process::host::{Change, ChangeKind, HostWatcher}, +}; +use sausage_vfs::observe::{ChangeKind as Kind, under}; +use tokio::sync::broadcast::{Receiver, error::TryRecvError}; +use wasmtime::component::Resource; +use wasmtime_wasi::p2::{DynPollable, Pollable, subscribe}; + +pub struct Watcher { + paths: Vec, + events: Receiver<(String, Kind)>, + pending: Vec, +} +impl Watcher { + fn push(&mut self, path: String, kind: Kind) { + if self.paths.iter().any(|base| under(&path, base)) { + if self + .pending + .iter() + .any(|c| matches!(c.kind, ChangeKind::Overflow)) + { + return; + } + if self.pending.len() >= 256 { + self.overflow(); + return; + } + self.pending.push(Change { + path, + kind: match kind { + Kind::Created => ChangeKind::Created, + Kind::Modified => ChangeKind::Modified, + Kind::Removed => ChangeKind::Removed, + }, + }); + } + } + fn overflow(&mut self) { + self.pending.clear(); + self.pending.push(Change { + path: "/".into(), + kind: ChangeKind::Overflow, + }); + } + fn drain(&mut self) { + loop { + match self.events.try_recv() { + Ok((path, kind)) => self.push(path, kind), + Err(TryRecvError::Lagged(_)) => self.overflow(), + Err(_) => break, + } + } + } +} +#[async_trait::async_trait] +impl Pollable for Watcher { + async fn ready(&mut self) { + loop { + self.drain(); + if !self.pending.is_empty() { + return; + } + match self.events.recv().await { + Ok((path, kind)) => self.push(path, kind), + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => self.overflow(), + Err(_) => return, + } + } + } +} +impl HostWatcher for State { + async fn new(&mut self, paths: Vec) -> wasmtime::Result> { + let observer = self + .runtime + .world + .root + .observer() + .ok_or_else(|| wasmtime::format_err!("filesystem has no observer"))?; + 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(), + pending: Vec::new(), + })?) + } + async fn subscribe( + &mut self, + watcher: Resource, + ) -> wasmtime::Result> { + subscribe(&mut self.table, watcher) + } + async fn changes(&mut self, watcher: Resource) -> wasmtime::Result> { + let watcher = self.table.get_mut(&watcher)?; + watcher.drain(); + Ok(std::mem::take(&mut watcher.pending)) + } + async fn drop(&mut self, watcher: Resource) -> wasmtime::Result<()> { + self.table.delete(watcher)?; + Ok(()) + } +} diff --git a/host/tests/init.rs b/host/tests/init.rs new file mode 100644 index 0000000..db9b111 --- /dev/null +++ b/host/tests/init.rs @@ -0,0 +1,455 @@ +//! Init acceptance tests use real WASI programs, a persistent guest cache, and temporary worlds. +use sausage::{ + exec::{Command, Exec, Options, Stream}, + process::{Runtime, supervise}, + world::World, +}; +use sausage_vfs::{Layer, Open}; +use std::{ + path::{Path, PathBuf}, + sync::{Arc, OnceLock}, + time::Duration, +}; +static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +fn guests() -> &'static Path { + static BUILT: OnceLock = OnceLock::new(); + BUILT.get_or_init(|| { + let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("init-guests"); + let manifest = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland/Cargo.toml"); + assert!( + std::process::Command::new("cargo") + .args([ + "build", + "--release", + "--target", + "wasm32-wasip2", + "-p", + "sausage-init", + "--manifest-path" + ]) + .arg(manifest) + .env("CARGO_TARGET_DIR", &target) + .status() + .unwrap() + .success() + ); + target.join("wasm32-wasip2/release") + }) +} +fn world() -> (tempfile::TempDir, Arc, Arc) { + let temp = tempfile::tempdir().unwrap(); + let bin = temp.path().join("world/bin"); + std::fs::create_dir_all(&bin).unwrap(); + for (source, name) in [ + ("init", "init"), + ("systemctl", "systemctl"), + ("journalctl", "journalctl"), + ("systemd-run", "systemd-run"), + ("init-probe", "probe"), + ] { + std::fs::copy( + guests().join(format!("{source}.wasm")), + bin.join(format!("{name}.wasm")), + ) + .unwrap(); + } + std::fs::create_dir_all( + temp.path() + .join("world/etc/systemd/system/default.target.wants"), + ) + .unwrap(); + let cache = Path::new(env!("CARGO_TARGET_TMPDIR")).join("init-components"); + std::fs::create_dir_all(&cache).unwrap(); + std::fs::create_dir_all(temp.path().join("host")).unwrap(); + std::os::unix::fs::symlink(cache, temp.path().join("host/compile-cache")).unwrap(); + let world = World::open(temp.path()).unwrap(); + let root = world.root.clone(); + (temp, Runtime::new(world).unwrap(), root) +} +async fn write(root: &Arc, path: &str, text: &[u8]) { + let file = root + .open( + path, + Open { + create: true, + write: true, + truncate: true, + ..Default::default() + }, + ) + .await + .unwrap(); + file.write(0, text.to_vec()).await.unwrap(); +} +async fn unit(root: &Arc, name: &str, body: &str) { + write( + root, + &format!("/etc/systemd/system/default.target.wants/{name}.service"), + b"", + ) + .await; + write( + root, + &format!("/etc/systemd/system/{name}.service"), + body.as_bytes(), + ) + .await; +} +async fn run(runtime: &Arc, argv: &[&str]) -> (i32, String) { + let mut out = Vec::new(); + let code = runtime + .exec( + Command::Argv(argv.iter().map(|a| (*a).into()).collect()), + Options { + timeout: Some(Duration::from_secs(30)), + cwd: "/".into(), + ..Default::default() + }, + &mut |stream, bytes| { + if stream == Stream::Stdout { + out.extend_from_slice(bytes); + } else { + eprint!("{}", String::from_utf8_lossy(bytes)); + } + }, + ) + .await + .unwrap(); + (code, String::from_utf8_lossy(&out).into_owned()) +} +async fn wait(mut check: impl FnMut() -> bool) { + tokio::time::timeout(Duration::from_secs(30), async { + while !check() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} +async fn state(root: &Arc, unit: &str, expected: &str) { + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if let Ok(file) = root + .open( + &format!("/run/systemd/state/{unit}.service"), + Default::default(), + ) + .await + && let Ok(bytes) = file.read(0, 65536).await + && serde_json::from_slice::(&bytes) + .is_ok_and(|s| s["state"] == expected) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} +fn lines(path: &Path) -> Vec { + std::fs::read_to_string(path) + .unwrap_or_default() + .lines() + .map(str::to_owned) + .collect() +} +#[tokio::test(flavor = "multi_thread")] +async fn watcher_and_child_exit_pollable() { + let _serial = SERIAL.lock().await; + let (_temp, runtime, _) = world(); + let result = run(&runtime, &["/bin/probe.wasm", "watch"]).await; + assert_eq!(result.0, 0, "{}", result.1); +} +#[tokio::test(flavor = "multi_thread")] +async fn dependencies_restart_limit_and_journal() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + unit( + &root, + "prepare", + "[Service]\nType=oneshot\nExecStart=/bin/probe.wasm oneshot /order.log\n", + ) + .await; + unit(&root, "after", "[Unit]\nAfter=prepare.service\nRequires=prepare.service\n[Service]\nExecStart=/bin/probe.wasm after /after.log\n").await; + unit( + &root, + "fail", + "[Service]\nType=oneshot\nExecStart=/bin/probe.wasm exit 2\n", + ) + .await; + unit(&root, "blocked", "[Unit]\nAfter=fail.service\nRequires=fail.service\n[Service]\nExecStart=/bin/probe.wasm oneshot /blocked.log\n").await; + unit(&root, "crash", "[Service]\nExecStart=/bin/probe.wasm crash /crash.log\nRestart=on-failure\nRestartSec=0.01\n").await; + unit(&root, "always", "[Service]\nExecStart=/bin/probe.wasm oneshot /always.log\nRestart=always\nRestartSec=0.01\nStartLimitBurst=2\n").await; + unit( + &root, + "success", + "[Service]\nExecStart=/bin/probe.wasm oneshot /success.log\nRestart=on-failure\n", + ) + .await; + unit( + &root, + "no", + "[Service]\nExecStart=/bin/probe.wasm crash /no.log\nRestart=no\n", + ) + .await; + unit( + &root, + "linger", + "[Service]\nExecStart=/bin/probe.wasm linger /linger.log\nTimeoutStopSec=0.02\n", + ) + .await; + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + wait(|| { + lines(&path.join("after.log")) == ["start"] && lines(&path.join("crash.log")).len() == 5 + }) + .await; + state(&root, "always", "failed").await; + assert_eq!(lines(&path.join("always.log")).len(), 2); + assert_eq!(lines(&path.join("success.log")).len(), 1); + assert_eq!(lines(&path.join("no.log")).len(), 1); + state(&root, "crash", "failed").await; + state(&root, "after", "active").await; + assert_eq!(lines(&path.join("order.log")), ["start"]); + assert!(!path.join("blocked.log").exists()); + wait(|| { + std::fs::read_to_string(path.join("var/log/journal/after.service.log")) + .unwrap_or_default() + .contains("line 10") + }) + .await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "is-active", "after"]) + .await + .0, + 0 + ); + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "is-active", "crash"]) + .await + .0, + 3 + ); + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "is-failed", "crash"]) + .await + .0, + 0 + ); + let (code, output) = run( + &runtime, + &["/bin/journalctl.wasm", "-u", "after", "-n", "5"], + ) + .await; + assert_eq!(code, 0); + assert_eq!(output, "line 6\nline 7\nline 8\nline 9\nline 10\n"); + state(&root, "linger", "active").await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "stop", "linger"]) + .await + .0, + 0 + ); + state(&root, "linger", "inactive").await; + assert_eq!(lines(&path.join("linger.log")), ["start"]); + supervisor.shutdown().await; + assert_eq!(lines(&path.join("after.log")), ["start", "stop"]); +} +#[tokio::test(flavor = "multi_thread")] +async fn live_unit_changes_and_binary_replacement() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + unit( + &root, + "live", + "[Service]\nExecStart=/bin/probe.wasm server /live.log\n", + ) + .await; + wait(|| lines(&path.join("live.log")) == ["start"]).await; + write( + &root, + "/etc/systemd/system/live.service", + b"[Service]\nExecStart=/bin/probe.wasm server /live.log\nEnvironment=REV=2\n", + ) + .await; + wait(|| lines(&path.join("live.log")) == ["start", "stop", "start"]).await; + let bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap(); + write(&root, "/bin/replacement.wasm", &bytes).await; + root.rename("/bin/replacement.wasm", "/bin/probe.wasm") + .await + .unwrap(); + wait(|| lines(&path.join("live.log")).len() >= 5).await; + root.remove_file("/etc/systemd/system/live.service") + .await + .unwrap(); + wait(|| { + lines(&path.join("live.log")) + .last() + .is_some_and(|l| l == "stop") + }) + .await; + supervisor.shutdown().await; +} +#[tokio::test(flavor = "multi_thread")] +async fn enable_persists_but_transient_units_do_not() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + write(&root, "/etc/systemd/system/persistent.service", b"[Service]\nExecStart=/bin/probe.wasm server /persistent.log\n[Install]\nWantedBy=default.target\n").await; + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + wait(|| path.join("var/log/journal").exists()).await; + assert_eq!( + run( + &runtime, + &["/bin/systemctl.wasm", "enable", "--now", "persistent"] + ) + .await + .0, + 0 + ); + wait(|| lines(&path.join("persistent.log")) == ["start"]).await; + assert_eq!( + run( + &runtime, + &["/bin/systemctl.wasm", "is-enabled", "persistent"] + ) + .await + .0, + 0 + ); + assert_eq!( + run( + &runtime, + &[ + "/bin/systemd-run.wasm", + "--unit", + "transient", + "--wait", + "/bin/probe.wasm", + "oneshot", + "/transient.log" + ] + ) + .await + .0, + 0 + ); + supervisor.shutdown().await; + drop(root); + drop(runtime); + let world = World::open(temp.path()).unwrap(); + assert!( + world + .root + .stat("/run/systemd/transient/transient.service", true) + .await + .is_err() + ); + let runtime = Runtime::new(world).unwrap(); + let supervisor = supervise(runtime.clone()); + wait(|| lines(&path.join("persistent.log")) == ["start", "stop", "start"]).await; + assert_eq!(lines(&path.join("transient.log")), ["start"]); + supervisor.shutdown().await; +} +#[tokio::test(flavor = "multi_thread")] +async fn init_crash_reaps_children_and_restarts_enabled_units() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + unit( + &root, + "server", + "[Service]\nExecStart=/bin/probe.wasm server /server.log\n", + ) + .await; + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world/server.log"); + wait(|| lines(&path) == ["start"]).await; + assert_eq!(run(&runtime, &["/bin/probe.wasm", "kill-init"]).await.0, 0); + wait(|| { + lines(&path) + .iter() + .filter(|line| line.as_str() == "start") + .count() + == 2 + }) + .await; + let table = root + .open("/proc/processes", Default::default()) + .await + .unwrap() + .read(0, 65536) + .await + .unwrap(); + let table = String::from_utf8(table).unwrap(); + assert_eq!( + table + .lines() + .filter(|l| l.contains("/bin/probe.wasm server")) + .count(), + 1 + ); + supervisor.shutdown().await; +} +#[tokio::test(flavor = "multi_thread")] +async fn journal_follow_observes_new_lines() { + let _serial = SERIAL.lock().await; + let (_temp, runtime, root) = world(); + // Init creates the public journal directory. + let supervisor = supervise(runtime.clone()); + tokio::time::timeout(Duration::from_secs(30), async { + while root.stat("/var/log/journal", true).await.is_err() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + write(&root, "/var/log/journal/x.service.log", b"old\n").await; + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let executor = runtime.clone(); + let task = tokio::spawn(async move { + executor + .exec( + Command::Argv(vec![ + "/bin/journalctl.wasm".into(), + "-u".into(), + "x".into(), + "-f".into(), + ]), + Options { + cwd: "/".into(), + ..Default::default() + }, + &mut |_, bytes| { + let _ = tx.send(bytes.to_vec()); + }, + ) + .await + }); + let first = tokio::time::timeout(Duration::from_secs(30), rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(first, b"old\n"); + let file = root + .open( + "/var/log/journal/x.service.log", + Open { + write: true, + ..Default::default() + }, + ) + .await + .unwrap(); + file.append(b"new\n".to_vec()).await.unwrap(); + let next = tokio::time::timeout(Duration::from_secs(30), rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(next, b"new\n"); + task.abort(); + let _ = task.await; + supervisor.shutdown().await; +} diff --git a/host/tests/process.rs b/host/tests/process.rs index 3332f75..1a9d6d1 100644 --- a/host/tests/process.rs +++ b/host/tests/process.rs @@ -12,7 +12,7 @@ use std::{ mod features { //! Acceptance coverage for preview1, identity, inherited IO, services, HTTP, and bundle aliases. use super::*; - use std::{collections::BTreeMap, time::Instant}; + use std::collections::BTreeMap; async fn poll(mut condition: impl FnMut() -> bool) { tokio::time::timeout(Duration::from_secs(10), async { @@ -23,13 +23,6 @@ mod features { .await .unwrap(); } - fn lines(path: &Path) -> Vec { - std::fs::read_to_string(path) - .unwrap_or_default() - .lines() - .map(str::to_owned) - .collect() - } async fn prepare(runtime: &Arc, argv: Vec) { assert_eq!( runtime @@ -285,61 +278,6 @@ mod features { assert_eq!(runtime.compiled_programs(), 1); } #[tokio::test(flavor = "multi_thread")] - async fn services_restart_edit_limit_and_eof() { - let _serial = SERIAL.lock().await; - let (temp, runtime, processes) = world(); - prepare(&runtime, vec!["/bin/launcher".into()]).await; - let root = temp.path().join("world"); - std::fs::create_dir_all(root.join("config")).unwrap(); - let spec = |name: &str, mode: &str, log: &str| serde_json::json!({"name": name, "argv": ["/bin/launcher", mode, log]}); - let write = |specs: Vec| { - std::fs::write( - root.join("config/services.yaml"), - serde_yaml_ng::to_string(&specs).unwrap(), - ) - .unwrap() - }; - write(vec![ - spec("a", "server", "/a.log"), - spec("crash", "crash", "/crash.log"), - ]); - let supervisor = sausage::process::supervise(runtime.clone()); - poll(|| { - lines(&root.join("a.log")) == ["start"] && lines(&root.join("crash.log")).len() == 6 - }) - .await; - prepare(&runtime, vec!["/bin/launcher".into()]).await; - assert_eq!(lines(&root.join("crash.log")).len(), 6); - assert!( - processes - .lock() - .unwrap() - .values() - .all(|p| p.conversation.is_none()) - ); - write(vec![spec("b", "server", "/b.log")]); - poll(|| { - lines(&root.join("a.log")) == ["start", "stop"] - && lines(&root.join("b.log")) == ["start"] - }) - .await; - let mut changed = spec("b", "server", "/b.log"); - changed["env"] = serde_json::json!({"REV": "new"}); - write(vec![changed]); - poll(|| lines(&root.join("b.log")) == ["start", "stop", "start"]).await; - write(vec![]); - poll(|| lines(&root.join("b.log")) == ["start", "stop", "start", "stop"]).await; - write(vec![spec("linger", "linger", "/linger.log")]); - poll(|| lines(&root.join("linger.log")) == ["start"]).await; - let stop = Instant::now(); - write(vec![]); - poll(|| processes.lock().unwrap().is_empty()).await; - assert!(stop.elapsed() >= Duration::from_secs(5)); - supervisor.abort(); - let _ = supervisor.await; - empty(&processes).await; - } - #[tokio::test(flavor = "multi_thread")] async fn git_local_and_anonymous_https_clone() { let _serial = SERIAL.lock().await; let (temp, runtime, _) = world(); diff --git a/host/wit/process.wit b/host/wit/process.wit index c38a856..f993026 100644 --- a/host/wit/process.wit +++ b/host/wit/process.wit @@ -3,6 +3,14 @@ package sausage:process@0.1.0; /// Programs share one world. These resources provide lifecycle and bounded stdio transport. interface host { use wasi:io/streams@0.2.12.{input-stream, output-stream}; + use wasi:io/poll@0.2.12.{pollable}; + enum change-kind { created, modified, removed, overflow } + record change { path: string, kind: change-kind } + resource watcher { + constructor(paths: list); + subscribe: func() -> pollable; + changes: func() -> list; + } enum stdio { pipe, inherit, null } record request { program: string, @@ -21,6 +29,7 @@ interface host { stdin: func() -> output-stream; stdout: func() -> input-stream; stderr: func() -> input-stream; + subscribe: func() -> pollable; status: func() -> option; wait: func() -> s32; kill: func(); diff --git a/userland/Cargo.lock b/userland/Cargo.lock index 6840a5c..f06325f 100644 --- a/userland/Cargo.lock +++ b/userland/Cargo.lock @@ -4045,6 +4045,17 @@ dependencies = [ "walkdir", ] +[[package]] +name = "sausage-init" +version = "0.1.0" +dependencies = [ + "sausage-spawn", + "serde", + "serde_json", + "shlex 1.3.0", + "wasi 0.14.7+wasi-0.2.4", +] + [[package]] name = "sausage-procps" version = "0.1.0" diff --git a/userland/Cargo.toml b/userland/Cargo.toml index 271c59c..b774fb9 100644 --- a/userland/Cargo.toml +++ b/userland/Cargo.toml @@ -1,7 +1,7 @@ # The default CLIs of a world, built for wasm32-wasip2. Its own workspace, outside the host's. [workspace] resolver = "3" -members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools"] +members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools", "init"] # WASI compatibility patches: git/vendor/PATCHES.md. [patch.crates-io] diff --git a/userland/init/Cargo.toml b/userland/init/Cargo.toml new file mode 100644 index 0000000..971e321 --- /dev/null +++ b/userland/init/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "sausage-init" +version = "0.1.0" +edition = "2024" + +[dependencies] +sausage-spawn = { path = "../spawn" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +wasi = "0.14" +shlex = "1" + +[[bin]] +name = "init-probe" +path = "src/probe.rs" + +[[bin]] +name = "init" +path = "src/main.rs" + +[[bin]] +name = "systemctl" +path = "src/systemctl.rs" + +[[bin]] +name = "journalctl" +path = "src/journalctl.rs" + +[[bin]] +name = "systemd-run" +path = "src/systemd-run.rs" diff --git a/userland/init/README.md b/userland/init/README.md new file mode 100644 index 0000000..be2a090 --- /dev/null +++ b/userland/init/README.md @@ -0,0 +1,75 @@ +# World init + +`/bin/init.wasm` owns unit policy. The host starts init without a conversation. +The host restarts init after an exit and kills its descendants before the next start. + +Init reads persistent units from `/etc/systemd/system/*.service`. +It reads transient units from `/run/systemd/transient/*.service`. +The host mounts `/run` in memory, so transient units disappear after a host restart. + +Unit files use `[Unit]`, `[Service]`, and `[Install]` sections. +Init supports dependency ordering, simple and oneshot services, environment variables, and restart policies. +The default start limit permits five starts within ten seconds. +A stop closes stdin. Init kills the process group after `TimeoutStopSec`, which defaults to ten seconds. +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. + +## Control protocol + +The protocol uses UTF-8 JSON files. Writers create a temporary file, then rename it to the final path. +Each request uses a unique identifier. + +1. The client watches `/run/systemd/replies/` before it submits the request. +2. The client writes `/run/systemd/requests/.json`. +3. Init reads the request and applies the action. +4. Init writes `/run/systemd/replies/` and removes the request. +5. The client reads and removes the reply. + +Request example: + +```json +{"action":"start","unit":"example.service"} +``` + +Actions are `start`, `stop`, and `restart`. +A reply acknowledges the action. It does not imply that the service completed its start or stop. +The control tools then wait for the requested state. A failed start returns code 1. + +Reply example: + +```json +{"code":0,"text":""} +``` + +A missing unit returns code 5. An unknown action returns code 1. +Clients read `/run/systemd/state/` for the current state: + +```json +{"state":"active","pid":42,"code":null} +``` + +States are `inactive`, `activating`, `active`, `deactivating`, and `failed`. +The `code` field records the most recent exit status. +`systemctl is-active` returns 3 for an inactive unit. +`systemctl is-failed` returns 0 for a failed unit and 1 otherwise. +`daemon-reload` returns 0 without an action. + +## Journal + +Init drains each unit's stdout and stderr into `/var/log/journal/.log`. +The cap is 1 MiB. Init keeps one rotated file with the `.1` suffix. +`journalctl -u -n ` prints the last lines. +`journalctl -f` observes the journal file and prints appended bytes. + +## Process observation + +`sausage-spawn` exposes `Watcher`, `ChangeKind`, and child exit pollables. +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. +Observers report successful VFS mutations, not direct host filesystem writes outside the VFS. diff --git a/userland/init/src/journalctl.rs b/userland/init/src/journalctl.rs new file mode 100644 index 0000000..c2c5c64 --- /dev/null +++ b/userland/init/src/journalctl.rs @@ -0,0 +1,66 @@ +//! Reads a unit journal, with optional live output through the world watcher. +use sausage_init::name; +use std::{ + fs, + io::{self, Write}, +}; +fn run() -> io::Result<()> { + let mut args = std::env::args().skip(1); + let (mut unit, mut lines, mut follow) = (String::new(), 10usize, false); + while let Some(arg) = args.next() { + match arg.as_str() { + "-u" => { + unit = name( + &args + .next() + .ok_or_else(|| io::Error::other("missing unit"))?, + ) + } + "-n" => { + lines = args + .next() + .and_then(|n| n.parse().ok()) + .ok_or_else(|| io::Error::other("missing count"))? + } + "-f" => follow = true, + "--no-pager" => (), + _ => return Err(io::Error::other("unknown argument")), + } + } + let path = format!("/var/log/journal/{unit}.log"); + let watcher = sausage_spawn::Watcher::new(&[path.clone()]); + let bytes = fs::read(&path).unwrap_or_default(); + let mut starts = vec![0]; + for (i, b) in bytes.iter().enumerate() { + if *b == b'\n' && i + 1 < bytes.len() { + starts.push(i + 1); + } + } + let start = if lines == 0 { + bytes.len() + } else { + starts[starts.len().saturating_sub(lines)] + }; + io::stdout().write_all(&bytes[start..])?; + io::stdout().flush()?; + let mut offset = bytes.len(); + while follow { + let pollable = watcher.subscribe(); + pollable.block(); + drop(pollable); + watcher.changes(); + let bytes = fs::read(&path).unwrap_or_default(); + if bytes.len() < offset { + offset = 0; + } + io::stdout().write_all(&bytes[offset..])?; + io::stdout().flush()?; + offset = bytes.len(); + } + Ok(()) +} +fn main() { + if run().is_err() { + std::process::exit(1); + } +} diff --git a/userland/init/src/lib.rs b/userland/init/src/lib.rs new file mode 100644 index 0000000..8c53313 --- /dev/null +++ b/userland/init/src/lib.rs @@ -0,0 +1,188 @@ +//! Unit definitions and the file protocol shared by init and its control tools. +use serde::{Deserialize, Serialize}; +use std::{collections::BTreeMap, fs, io, time::Duration}; +/// WASI 0.2's legacy exit call loses numeric error codes. Use the standard extended call. +pub fn exit(code: i32) -> ! { + wasi::cli::exit::exit_with_code(code as u8); + unreachable!("exit returned") +} +pub const RUN: &str = "/run/systemd"; +pub const UNITS: &str = "/etc/systemd/system"; +pub const TRANSIENT: &str = "/run/systemd/transient"; +#[derive(Clone, Debug, PartialEq)] +pub struct Unit { + pub values: BTreeMap, + pub source: String, +} +impl Unit { + pub fn parse(source: String) -> io::Result { + let mut values = BTreeMap::new(); + let mut section = ""; + for line in source.lines().map(str::trim) { + if line.is_empty() || line.starts_with(['#', ';']) { + continue; + } + if line.starts_with('[') && line.ends_with(']') { + section = &line[1..line.len() - 1]; + continue; + } + let (key, value) = line + .split_once('=') + .ok_or_else(|| io::Error::other("expected Key=Value"))?; + let key = format!("{section}.{}", key.trim()); + if matches!( + key.as_str(), + "Unit.After" + | "Unit.Before" + | "Unit.Requires" + | "Unit.Wants" + | "Service.Environment" + ) { + values + .entry(key) + .and_modify(|v: &mut String| { + v.push(' '); + v.push_str(value.trim()); + }) + .or_insert(value.trim().into()); + } else { + values.insert(key, value.trim().into()); + } + } + if !values.contains_key("Service.ExecStart") { + return Err(io::Error::other("missing ExecStart")); + } + Ok(Self { values, source }) + } + pub fn get(&self, key: &str) -> &str { + self.values.get(key).map(String::as_str).unwrap_or("") + } + pub fn words(&self, key: &str) -> Vec { + self.get(key).split_whitespace().map(name).collect() + } + pub fn seconds(&self, key: &str, default: f64) -> Duration { + Duration::from_secs_f64( + self.get(key) + .trim_end_matches('s') + .parse::() + .ok() + .filter(|n| n.is_finite() && *n >= 0.0) + .unwrap_or(default), + ) + } +} +pub fn name(s: &str) -> String { + if s.ends_with(".service") || s.ends_with(".target") { + s.into() + } else { + format!("{s}.service") + } +} +pub fn unit_path(name: &str) -> String { + let persistent = format!("{UNITS}/{name}"); + if std::path::Path::new(&persistent).exists() { + persistent + } else { + format!("{TRANSIENT}/{name}") + } +} +pub fn enabled(name: &str) -> bool { + ["default.target", "multi-user.target"] + .iter() + .any(|target| std::path::Path::new(&format!("{UNITS}/{target}.wants/{name}")).exists()) +} +#[derive(Serialize, Deserialize, Default)] +pub struct Status { + pub state: String, + pub pid: Option, + pub code: Option, +} +#[derive(Serialize, Deserialize)] +pub struct Request { + pub action: String, + pub unit: String, +} +#[derive(Serialize, Deserialize)] +pub struct Reply { + pub code: i32, + pub text: String, +} +pub fn atomic(path: &str, bytes: &[u8]) -> io::Result<()> { + let tmp = format!("{path}.tmp"); + fs::write(&tmp, bytes)?; + fs::rename(tmp, path) +} +pub fn request(action: &str, unit: &str) -> io::Result { + let id = format!( + "{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let reply = format!("{RUN}/replies/{id}"); + let state = format!("{RUN}/state/{}", name(unit)); + let watcher = sausage_spawn::Watcher::new(&[reply.clone(), state.clone()]); + atomic( + &format!("{RUN}/requests/{id}.json"), + &serde_json::to_vec(&Request { + action: action.into(), + unit: name(unit), + })?, + )?; + let deadline = std::time::Instant::now() + Duration::from_secs(60); + let mut acknowledged = false; + loop { + if !acknowledged && let Ok(bytes) = fs::read(&reply) { + fs::remove_file(&reply)?; + let reply: Reply = serde_json::from_slice(&bytes)?; + if reply.code != 0 { + return Ok(reply); + } + acknowledged = true; + } + if acknowledged && let Ok(bytes) = fs::read(&state) { + let status: Status = serde_json::from_slice(&bytes)?; + let done = if action == "stop" { + status.state == "inactive" + } else { + status.state == "active" + || status.state == "failed" + || status.state == "inactive" && status.code.is_some() + }; + if done { + return Ok(Reply { + code: if action != "stop" && status.state == "failed" { + 1 + } else { + 0 + }, + text: String::new(), + }); + } + } + if std::time::Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "init did not reply", + )); + } + // The short timer also handles a missing or failed init. + let _ = watcher.changes(); + std::thread::sleep(Duration::from_millis(20)); + } +} +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn unit_sections_and_defaults() { + let unit = Unit::parse("[Unit]\nAfter=prepare\nRequires=prepare\n[Service]\nType=oneshot\nExecStart=/bin/task.wasm\n".into()).unwrap(); + assert_eq!(unit.words("Unit.Requires"), ["prepare.service"]); + assert_eq!( + unit.seconds("Service.TimeoutStopSec", 10.), + Duration::from_secs(10) + ); + assert!(Unit::parse("[Unit]\nDescription=missing command".into()).is_err()); + } +} diff --git a/userland/init/src/main.rs b/userland/init/src/main.rs new file mode 100644 index 0000000..7a49b80 --- /dev/null +++ b/userland/init/src/main.rs @@ -0,0 +1,421 @@ +//! Runs the world's unit state machine, journal drains, and file-based control requests. +use sausage_init::*; +use sausage_spawn::{Child, Command, Stdio}; +use std::{ + collections::{BTreeMap, BTreeSet, VecDeque}, + fs, + io::{self, Write}, + time::{Duration, Instant}, +}; +struct Service { + unit: Unit, + child: Option, + state: String, + code: Option, + desired: bool, + starts: VecDeque, + next: Instant, + stopping: Option, +} +impl Service { + fn new(unit: Unit) -> Self { + Self { + unit, + child: None, + state: "inactive".into(), + code: None, + desired: false, + starts: VecDeque::new(), + next: Instant::now(), + stopping: None, + } + } + fn stop(&mut self) { + self.desired = false; + if let Some(child) = &mut self.child { + child.stdin.take(); + self.stopping + .get_or_insert(Instant::now() + self.unit.seconds("Service.TimeoutStopSec", 10.)); + self.state = "deactivating".into(); + } else { + self.state = "inactive".into(); + } + } + fn start(&mut self) { + let now = Instant::now(); + let interval = self.unit.seconds("Service.StartLimitIntervalSec", 10.); + while self + .starts + .front() + .is_some_and(|t| now.duration_since(*t) >= interval) + { + self.starts.pop_front(); + } + let burst = self + .unit + .get("Service.StartLimitBurst") + .parse() + .unwrap_or(5); + if burst != 0 && self.starts.len() >= burst { + self.state = "failed".into(); + self.desired = false; + return; + } + self.starts.push_back(now); + let Some(argv) = shlex::split(self.unit.get("Service.ExecStart")).filter(|a| !a.is_empty()) + else { + self.state = "failed".into(); + self.desired = false; + return; + }; + let mut command = Command::new(&argv[0]); + command + .args(&argv[1..]) + .new_group() + .stdin(Stdio::Pipe) + .stdout(Stdio::Pipe) + .stderr(Stdio::Pipe); + let cwd = self.unit.get("Service.WorkingDirectory"); + command.current_dir(if cwd.is_empty() { "/" } else { cwd }); + for assignment in shlex::split(self.unit.get("Service.Environment")).unwrap_or_default() { + if let Some((key, value)) = assignment.split_once('=') { + command.env(key, value); + } + } + match command.spawn() { + Ok(child) => { + self.child = Some(child); + self.state = if self.unit.get("Service.Type") == "oneshot" { + "activating" + } else { + "active" + } + .into(); + self.code = None; + } + Err(_) => { + self.code = Some(127); + self.exited(127); + } + } + } + fn exited(&mut self, code: i32) { + self.code = Some(code); + if self.stopping.take().is_some() { + self.state = if self.desired { + "activating" + } else { + "inactive" + } + .into(); + if self.desired { + self.code = None; + } + return; + } + self.state = if code != 0 { + "failed" + } else if matches!( + self.unit.get("Service.RemainAfterExit"), + "yes" | "true" | "1" + ) { + "active" + } else { + "inactive" + } + .into(); + let restart = self.unit.get("Service.Restart"); + if restart == "always" || restart == "on-failure" && code != 0 { + self.next = Instant::now() + self.unit.seconds("Service.RestartSec", 0.1); + } else { + self.desired = false; + } + } + fn tick(&mut self, name: &str) { + if let Some(child) = &mut self.child { + for slot in [&mut child.stdout, &mut child.stderr] { + if let Some(stream) = slot { + loop { + match stream.read(8192) { + Ok(bytes) if !bytes.is_empty() => journal(name, &bytes), + Ok(_) => break, + Err(_) => { + *slot = None; + break; + } + } + } + } + } + if child.try_wait().is_none() && self.stopping.is_some_and(|t| Instant::now() >= t) { + let _ = sausage_spawn::kill(child.id()); + } + if let Some(code) = child.try_wait() { + if child.stdout.is_none() && child.stderr.is_none() { + self.child = None; + self.exited(code); + } + } + } + } + fn publish(&self, name: &str) { + let status = Status { + state: self.state.clone(), + pid: self.child.as_ref().map(Child::id), + code: self.code, + }; + let bytes = serde_json::to_vec(&status).unwrap(); + let path = format!("{RUN}/state/{name}"); + if fs::read(&path).ok().as_deref() != Some(bytes.as_slice()) { + let _ = atomic(&path, &bytes); + } + } +} +fn journal(name: &str, bytes: &[u8]) { + let path = format!("/var/log/journal/{name}.log"); + if fs::metadata(&path).is_ok_and(|m| m.len() + bytes.len() as u64 > 1024 * 1024) { + let _ = fs::rename(&path, format!("{path}.1")); + } + if let Ok(mut f) = fs::OpenOptions::new().create(true).append(true).open(path) { + let _ = f.write_all(bytes); + } +} +fn activate(name: &str, services: &mut BTreeMap, seen: &mut BTreeSet) { + if !seen.insert(name.into()) { + return; + } + let Some(service) = services.get(name) else { + return; + }; + let deps = [ + service.unit.words("Unit.Requires"), + service.unit.words("Unit.Wants"), + ] + .concat(); + for dep in deps { + activate(&dep, services, seen); + } + let service = services.get_mut(name).unwrap(); + service.desired = true; + if service.child.is_none() { + service.state = "activating".into(); + service.code = None; + } + service.next = Instant::now(); +} +fn predecessors(name: &str, services: &BTreeMap) -> Vec { + let mut deps = services[name].unit.words("Unit.After"); + deps.extend(services[name].unit.words("Unit.Requires")); + for (other, service) in services { + if service.unit.words("Unit.Before").iter().any(|s| s == name) { + deps.push(other.clone()); + } + } + deps +} +fn scan(services: &mut BTreeMap) { + let mut found = BTreeMap::new(); + let mut present = BTreeSet::new(); + for dir in [TRANSIENT, UNITS] { + for entry in fs::read_dir(dir).into_iter().flatten().flatten() { + let name = entry.file_name().to_string_lossy().into_owned(); + if !name.ends_with(".service") { + continue; + } + present.insert(name.clone()); + if let Ok(source) = fs::read_to_string(entry.path()) { + if let Ok(unit) = Unit::parse(source) { + found.insert(name, unit); + } + } + } + } + for (name, service) in services.iter_mut() { + if !present.contains(name) { + service.stop(); + } + } + services.retain(|name, service| present.contains(name) || service.child.is_some()); + let mut new_enabled = Vec::new(); + for (name, unit) in found { + if !services.contains_key(&name) && enabled(&name) { + new_enabled.push(name.clone()); + } + let service = services + .entry(name.clone()) + .or_insert_with(|| Service::new(unit.clone())); + if service.unit != unit { + let desired = service.desired || service.state == "active"; + service.stop(); + service.unit = unit; + service.desired = desired; + service.starts.clear(); + service.next = Instant::now(); + } + } + for name in new_enabled { + activate(&name, services, &mut BTreeSet::new()); + } +} +fn controls(services: &mut BTreeMap) { + for entry in fs::read_dir(format!("{RUN}/requests")) + .into_iter() + .flatten() + .flatten() + { + let filename = entry.file_name().to_string_lossy().into_owned(); + if !filename.ends_with(".json") { + continue; + } + let Ok(bytes) = fs::read(entry.path()) else { + continue; + }; + let Ok(request) = serde_json::from_slice::(&bytes) else { + continue; + }; + let mut reply = Reply { + code: 0, + text: String::new(), + }; + if !services.contains_key(&request.unit) { + reply.code = 5; + reply.text = "unit not found".into(); + } else { + match request.action.as_str() { + "start" => { + services.get_mut(&request.unit).unwrap().starts.clear(); + activate(&request.unit, services, &mut BTreeSet::new()); + } + "stop" => services.get_mut(&request.unit).unwrap().stop(), + "restart" => { + let service = services.get_mut(&request.unit).unwrap(); + service.stop(); + service.desired = true; + if service.child.is_none() { + service.state = "activating".into(); + service.code = None; + } + service.starts.clear(); + service.next = Instant::now(); + } + _ => { + reply.code = 1; + reply.text = "unknown action".into(); + } + } + } + if let Some(service) = services.get(&request.unit) { + service.publish(&request.unit); + } + let id = filename.trim_end_matches(".json"); + let _ = atomic( + &format!("{RUN}/replies/{id}"), + &serde_json::to_vec(&reply).unwrap(), + ); + let _ = fs::remove_file(entry.path()); + } +} +fn main() -> io::Result<()> { + for dir in [ + UNITS, + TRANSIENT, + "/run/systemd/requests", + "/run/systemd/replies", + "/run/systemd/state", + "/var/log/journal", + ] { + fs::create_dir_all(dir)?; + } + let watcher = sausage_spawn::Watcher::new(&["/".into()]); + let stdin = wasi::cli::stdin::get_stdin(); + let mut services = BTreeMap::new(); + scan(&mut services); + loop { + if stdin.read(1).is_err() { + break; + } + 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 names: Vec<_> = services.keys().cloned().collect(); + for name in &names { + services.get_mut(name).unwrap().tick(name); + } + for name in &names { + let service = &services[name]; + if !service.desired + || service.child.is_some() + || Instant::now() < service.next + || service.stopping.is_some() + { + continue; + } + let required = service.unit.words("Unit.Requires"); + if required + .iter() + .any(|dep| !services.contains_key(dep) || services[dep].state == "failed") + { + let service = services.get_mut(name).unwrap(); + service.state = "failed".into(); + service.desired = false; + continue; + } + if predecessors(name, &services).iter().any(|dep| { + services.get(dep).is_some_and(|s| { + s.state == "activating" || s.desired && s.child.is_none() && s.code.is_none() + }) + }) { + continue; + } + services.get_mut(name).unwrap().start(); + } + for (name, service) in &services { + service.publish(name); + } + std::thread::sleep(Duration::from_millis(20)); + } + // Stop dependents before their prerequisites. A cycle stops together. + while services.values().any(|s| s.child.is_some()) { + let live: Vec<_> = services + .iter() + .filter(|(_, s)| s.child.is_some()) + .map(|(n, _)| n.clone()) + .collect(); + let mut leaves: Vec<_> = live + .iter() + .filter(|name| { + !live + .iter() + .any(|other| other != *name && predecessors(other, &services).contains(name)) + }) + .cloned() + .collect(); + if leaves.is_empty() { + leaves = live; + } + 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)); + } + Ok(()) +} diff --git a/userland/init/src/probe.rs b/userland/init/src/probe.rs new file mode 100644 index 0000000..bba6356 --- /dev/null +++ b/userland/init/src/probe.rs @@ -0,0 +1,96 @@ +//! Acceptance guest for unit lifecycle and process-interface readiness tests. +use std::{ + fs, + io::{Read, Write}, +}; +fn main() { + let args: Vec<_> = std::env::args().collect(); + match args.get(1).map(String::as_str).unwrap_or("") { + "kill-init" => { + let table = fs::read_to_string("/proc/processes").unwrap(); + let row = table + .lines() + .find(|line| line.ends_with("/bin/init.wasm")) + .unwrap(); + sausage_spawn::kill(row.split('\t').next().unwrap().parse().unwrap()).unwrap(); + } + "watch" => { + use sausage_spawn::{ChangeKind, Watcher}; + let watcher = Watcher::new(&["/run/watched".into()]); + let poll = watcher.subscribe(); + assert!(!poll.ready()); + fs::create_dir_all("/run/watched").unwrap(); + poll.block(); + assert!( + watcher + .changes() + .iter() + .any(|c| matches!(c.kind, ChangeKind::Created)) + ); + fs::write("/run/watched/a", "a").unwrap(); + watcher.changes(); + fs::write("/run/watched/a", "b").unwrap(); + assert!( + watcher + .changes() + .iter() + .any(|c| c.path == "/run/watched/a" && matches!(c.kind, ChangeKind::Modified)) + ); + fs::remove_file("/run/watched/a").unwrap(); + assert!( + watcher + .changes() + .iter() + .any(|c| matches!(c.kind, ChangeKind::Removed)) + ); + for i in 0..300 { + fs::write(format!("/run/watched/{i}"), "").unwrap(); + } + assert!( + watcher + .changes() + .iter() + .any(|c| matches!(c.kind, ChangeKind::Overflow)) + ); + drop(poll); + let mut cmd = sausage_spawn::Command::new("/bin/probe.wasm"); + cmd.arg("exit").arg("7"); + let child = cmd.spawn().unwrap(); + let exit = child.subscribe(); + exit.block(); + assert_eq!(child.try_wait(), Some(7)); + } + "exit" => sausage_init::exit(args.get(2).and_then(|a| a.parse().ok()).unwrap_or(0)), + "oneshot" | "server" | "crash" | "after" | "linger" => { + if args[1] == "after" { + assert!( + !fs::read_to_string("/proc/processes") + .unwrap() + .contains("oneshot /order.log") + ); + } + let mut log = fs::OpenOptions::new() + .create(true) + .append(true) + .open(&args[2]) + .unwrap(); + writeln!(log, "start").unwrap(); + for n in 1..=10 { + println!("line {n}"); + } + if args[1] == "linger" { + loop { + std::thread::sleep(std::time::Duration::from_secs(1)); + } + } + if args[1] == "crash" { + std::process::exit(1); + } + if args[1] == "server" || args[1] == "after" { + let _ = std::io::stdin().read_to_end(&mut Vec::new()); + writeln!(log, "stop").unwrap(); + } + } + _ => panic!("unknown probe mode"), + } +} diff --git a/userland/init/src/systemctl.rs b/userland/init/src/systemctl.rs new file mode 100644 index 0000000..2a44a3c --- /dev/null +++ b/userland/init/src/systemctl.rs @@ -0,0 +1,123 @@ +//! Controls init and reads its public unit state files. +use sausage_init::*; +use std::{fs, io}; +fn run() -> io::Result { + let args: Vec<_> = std::env::args().skip(1).collect(); + let action = args.first().map(String::as_str).unwrap_or("list-units"); + if action == "daemon-reload" { + return Ok(0); + } + if action == "list-units" { + for entry in fs::read_dir(format!("{RUN}/state"))? { + let entry = entry?; + let status: Status = serde_json::from_slice(&fs::read(entry.path())?)?; + println!("{} {}", entry.file_name().to_string_lossy(), status.state); + } + return Ok(0); + } + let now = args.iter().any(|a| a == "--now"); + let units: Vec<_> = args + .iter() + .skip(1) + .filter(|a| !a.starts_with('-')) + .map(|a| name(a)) + .collect(); + if units.is_empty() { + return Ok(1); + } + let mut result = 0; + for unit in units { + let status = fs::read(format!("{RUN}/state/{unit}")) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()) + .unwrap_or_default(); + let code = match action { + "is-active" => { + println!( + "{}", + if status.state.is_empty() { + "unknown" + } else { + &status.state + } + ); + if status.state == "active" { 0 } else { 3 } + } + "is-failed" => { + println!( + "{}", + if status.state.is_empty() { + "unknown" + } else { + &status.state + } + ); + if status.state == "failed" { 0 } else { 1 } + } + "is-enabled" => { + println!( + "{}", + if enabled(&unit) { + "enabled" + } else { + "disabled" + } + ); + if enabled(&unit) { 0 } else { 1 } + } + "status" => { + println!("{unit}: {}", status.state); + if status.state == "active" { + 0 + } else if status.state.is_empty() { + 4 + } else { + 3 + } + } + "cat" => { + print!("{}", fs::read_to_string(unit_path(&unit))?); + 0 + } + "enable" | "disable" => { + let spec = Unit::parse(fs::read_to_string(unit_path(&unit))?)?; + let targets: Vec<_> = spec + .get("Install.WantedBy") + .split_whitespace() + .filter(|t| matches!(*t, "default.target" | "multi-user.target")) + .collect(); + if action == "enable" && targets.is_empty() { + return Ok(1); + } + for target in ["default.target", "multi-user.target"] { + let dir = format!("{UNITS}/{target}.wants"); + let path = format!("{dir}/{unit}"); + if action == "enable" && targets.contains(&target) { + fs::create_dir_all(dir)?; + fs::write(path, b"")?; + } else if action == "disable" { + let _ = fs::remove_file(path); + } + } + if now { + request(if action == "enable" { "start" } else { "stop" }, &unit)?.code + } else { + 0 + } + } + "start" | "stop" | "restart" => { + let reply = request(action, &unit)?; + if !reply.text.is_empty() { + println!("{}", reply.text); + } + reply.code + } + _ => 1, + }; + result = result.max(code); + } + Ok(result) +} +fn main() { + sausage_init::exit(run().unwrap_or(1)); +} diff --git a/userland/init/src/systemd-run.rs b/userland/init/src/systemd-run.rs new file mode 100644 index 0000000..968face --- /dev/null +++ b/userland/init/src/systemd-run.rs @@ -0,0 +1,82 @@ +//! Creates transient unit files and optionally waits for their final state. +use sausage_init::*; +use std::{fs, io, time::Duration}; +fn run() -> io::Result { + let mut args = std::env::args().skip(1).peekable(); + let mut unit = format!( + "run-{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let (mut wait, mut properties, mut command) = (false, Vec::new(), Vec::new()); + while let Some(arg) = args.next() { + match arg.as_str() { + "--unit" => { + unit = args + .next() + .ok_or_else(|| io::Error::other("missing unit"))? + } + "--wait" => wait = true, + "-p" => properties.push( + args.next() + .ok_or_else(|| io::Error::other("missing property"))?, + ), + "--working-directory" => properties.push(format!( + "WorkingDirectory={}", + args.next() + .ok_or_else(|| io::Error::other("missing directory"))? + )), + "-E" => properties.push(format!( + "Environment={}", + args.next() + .ok_or_else(|| io::Error::other("missing environment"))? + )), + "--" => { + command.extend(args); + break; + } + _ if arg.starts_with("--unit=") => unit = arg[7..].into(), + _ if arg.starts_with('-') => return Ok(1), + _ => { + command.push(arg); + command.extend(args); + break; + } + } + } + if command.is_empty() { + return Ok(1); + } + let unit = name(&unit); + let mut text = format!( + "[Service]\nExecStart={}\n", + shlex::try_join(command.iter().map(String::as_str)).map_err(io::Error::other)? + ); + for property in properties { + text.push_str(&property); + text.push('\n'); + } + fs::create_dir_all(TRANSIENT)?; + atomic(&format!("{TRANSIENT}/{unit}"), text.as_bytes())?; + let reply = request("start", &unit)?; + if reply.code != 0 { + return Ok(reply.code); + } + if wait { + loop { + if let Ok(bytes) = fs::read(format!("{RUN}/state/{unit}")) { + let status: Status = serde_json::from_slice(&bytes)?; + if let Some(code) = status.code { + return Ok(code); + } + } + std::thread::sleep(Duration::from_millis(20)); + } + } + Ok(0) +} +fn main() { + sausage_init::exit(run().unwrap_or(1)); +} diff --git a/userland/spawn/src/lib.rs b/userland/spawn/src/lib.rs index 004ee87..a8b4020 100644 --- a/userland/spawn/src/lib.rs +++ b/userland/spawn/src/lib.rs @@ -7,7 +7,10 @@ mod bindings { } use bindings::sausage::process::host; pub use bindings::wasi::io::streams::{InputStream, OutputStream}; -pub use host::Stdio; +pub use host::{Stdio, Watcher, Change, ChangeKind}; +pub use bindings::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) } /// A prepared child process. pub struct Command { @@ -63,6 +66,7 @@ pub struct Child { } impl Child { pub fn id(&self) -> u32 { self.child.pid() } + pub fn subscribe(&self) -> Pollable { self.child.subscribe() } pub fn try_wait(&self) -> Option { self.child.status() } pub fn wait(&self) -> i32 { self.child.wait() } /// Stop every process in this child's group. diff --git a/vfs/src/lib.rs b/vfs/src/lib.rs index 03d816c..0902c87 100644 --- a/vfs/src/lib.rs +++ b/vfs/src/lib.rs @@ -67,6 +67,10 @@ pub trait File: Send + Sync { /// Implementations must retain open file identity and return portable IO errors. #[async_trait] pub trait Layer: Send + Sync { + /// Shared observation of mutations in this namespace, when available. + fn observer(&self) -> Option { + None + } async fn stat(&self, path: &str, follow: bool) -> Result; /// Return the target of a symbolic link, or InvalidInput for another node. async fn readlink(&self, _path: &str) -> Result { diff --git a/vfs/src/mounts.rs b/vfs/src/mounts.rs index c31207c..d062cc9 100644 --- a/vfs/src/mounts.rs +++ b/vfs/src/mounts.rs @@ -6,15 +6,24 @@ struct Mount { readonly: bool, } /// A per-process mount namespace. Mount the root before use. -#[derive(Default)] pub struct Mounts { mounts: BTreeMap, + observer: observe::Observer, +} +impl Default for Mounts { + fn default() -> Self { + Self { + mounts: BTreeMap::new(), + observer: observe::Observer::new(|_| {}), + } + } } impl Mounts { /// Add or replace a mount. Read-only mounts reject every mutation. pub fn mount(&mut self, path: &str, layer: Arc, readonly: bool) { - self.mounts - .insert(resolve("/", path), Mount { layer, readonly }); + let path = resolve("/", path); + let layer = self.observer.layer(layer, &path, Process::default()); + self.mounts.insert(path, Mount { layer, readonly }); } pub(crate) async fn metadata_file(&self, path: &str, follow: bool) -> Result> { let (m, p) = self.route(path, true)?; @@ -49,6 +58,9 @@ impl Mounts { } #[async_trait] impl Layer for Mounts { + fn observer(&self) -> Option { + Some(self.observer.clone()) + } async fn stat(&self, path: &str, follow: bool) -> Result { let (m, p) = self.route(path, false)?; m.layer.stat(&p, follow).await diff --git a/vfs/src/observe.rs b/vfs/src/observe.rs index e7f67d0..1738164 100644 --- a/vfs/src/observe.rs +++ b/vfs/src/observe.rs @@ -6,6 +6,14 @@ use std::{ }; use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock}; +/// The successful namespace or content mutation observed at a world path. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ChangeKind { + Created, + Modified, + Removed, +} + type Namespace = Mutex>>>; enum Held { Read { _guard: OwnedRwLockReadGuard<()> }, @@ -20,6 +28,7 @@ type Callback = dyn Fn(&str) + Send + Sync; #[derive(Clone)] pub struct Observer { changed: Arc, + events: tokio::sync::broadcast::Sender<(String, ChangeKind)>, 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. @@ -29,10 +38,19 @@ 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, 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() + } + fn notify(&self, path: &str, kind: ChangeKind) { + (self.changed)(path); + let _ = self.events.send((path.to_owned(), kind)); + } /// Wrap a layer at its canonical namespace prefix (e.g. `/run`). pub fn layer(&self, layer: Arc, prefix: &str, process: Process) -> Arc { Arc::new(Observed { @@ -149,7 +167,7 @@ impl Handle { let result = f.await?; let path = path.lock().unwrap().clone(); if let Some(path) = path { - (observer.changed)(&path); + observer.notify(&path, ChangeKind::Modified); } Ok(result) }) @@ -178,6 +196,7 @@ impl Layer for Observed { self.process .operation(async move { let order = observer.order(&[(&path, false)]).await; + let existed = !options.create || layer.stat(&local, options.follow).await.is_ok(); let file = layer.open(&local, options).await?; let handle = Arc::new(Handle { file, @@ -187,7 +206,14 @@ impl Layer for Observed { }); drop(order); if options.create || options.truncate { - (observer.changed)(&path); + observer.notify( + &path, + if existed { + ChangeKind::Modified + } else { + ChangeKind::Created + }, + ); } Ok(handle as Arc) }) @@ -203,7 +229,7 @@ impl Layer for Observed { self.process .operation(async move { layer.mkdir(&local).await?; - (observer.changed)(&path); + observer.notify(&path, ChangeKind::Created); Ok(()) }) .await @@ -226,13 +252,22 @@ impl Layer for Observed { self.process .operation(async move { let order = observer.order(&[(&from, true), (&to, true)]).await; + let replaced = layer.stat(&b, false).await.is_ok(); layer.rename(&a, &b).await?; - if from != to { - observer.moved(&from, Some(&to)); + if from == to { + return Ok(()); } + observer.moved(&from, Some(&to)); drop(order); - (observer.changed)(&from); - (observer.changed)(&to); + observer.notify(&from, ChangeKind::Removed); + observer.notify( + &to, + if replaced { + ChangeKind::Modified + } else { + ChangeKind::Created + }, + ); Ok(()) }) .await @@ -256,7 +291,7 @@ impl Observed { } observer.moved(&path, None); drop(order); - (observer.changed)(&path); + observer.notify(&path, ChangeKind::Removed); Ok(()) }) .await