diff --git a/PROTOCOL.md b/PROTOCOL.md index 341701b..ac52b40 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -249,6 +249,36 @@ Rules: Principle: **inspect through files, act through calls.** +### File watching + +`core.watch {"path": "/state/session/mailbox"}` stays open for the caller +process's lifetime. It needs a request ID and a running plugin caller. +Its first `core.notify` is `"armed"`, with empty params. Only then scan the +files. Writes before arming are visible in that scan; writes after arming +produce `"changed"` notifications (also with empty params). Rescan after +each notification; files, not event counts, are the truth. + +Notifications follow successful plugin-originated file mutations: creation, +truncation, descriptor/stream writes, rename (both source and destination), +and deletion, including mounted `/run` and `/self` files. Paths are normalized +in the virtual namespace; `/self` means the caller's `/proc/` folder. +A watch matches itself and descendants, not sibling prefixes. Moving or +removing an ancestor also invalidates it. Paths need not exist when armed. +Open file writes follow plugin renames; unlinked/replaced handles no longer +name visible files. External host writes are outside this contract. + +Changes may coalesce, including a change already visible in the initial +scan, but a change after the last notification always produces another. +Delivery holds no filesystem locks. Observation does not serialize descriptor +IO; only overlapping opens and rename/removal wait for one another. +Watch the narrowest useful path: delivery is not rate-limited, so watching a +busy parent can cause a rescan for every write. + +The watch ends with its caller; there is no cancellation API or polling/retry +loop. SDK callers should poll the watch future alongside their handler's work +and drop it on return, not keep it in a background task that prevents an idle +process from exiting. Dropping the future does not cancel the host watch. + ## 5. Endpoints An endpoint is a JSON-RPC interface that a running process serves over TCP, @@ -356,6 +386,7 @@ At start, the core starts each service in `/config/boot.yaml`, in order: | Method | Params | Result | | --- | --- | --- | | `core.call` | `method`, `params`, `detach`? | the result of the method | +| `core.watch` | `path` | stays open; `armed`, then `changed` notifications (section 4) | | `core.notify` | `name`, `params` | none (a notification) | | `core.stop` | `pid` | `{}` when the process has ended | | `core.kill` | `pid` | `{}` | diff --git a/cli/src/core.rs b/cli/src/core.rs index 7bc863f..0f6fc0c 100644 --- a/cli/src/core.rs +++ b/cli/src/core.rs @@ -12,6 +12,7 @@ impl Host { p: Value, ) -> RpcResult { match method { + "watch" => self.watch(caller, id, required_str(&p, "path")?).await, "call" => { self.route( caller.into(), diff --git a/cli/src/filesystem.rs b/cli/src/filesystem.rs index 30be907..d0d83f5 100644 --- a/cli/src/filesystem.rs +++ b/cli/src/filesystem.rs @@ -47,23 +47,40 @@ impl Host { tracker: Process, ) -> anyhow::Result { let root = self.0.world.root.clone(); + let operations = tracker.clone(); let host = tracker .clone() .blocking(move || sausage_vfs::Host::new(root, tracker)) .await?; let mut mounts = Mounts::default(); - mounts.mount("/", Arc::new(host), false); + mounts.mount( + "/", + self.0 + .observer + .layer(Arc::new(host), "/", operations.clone()), + false, + ); mounts.mount("/proc", Arc::new(Proc(self.clone())), true); mounts.mount( "/self", - Arc::new(SelfFolder { - host: self.clone(), - pid, - memory, - }), + self.0.observer.layer( + Arc::new(SelfFolder { + host: self.clone(), + pid: pid.clone(), + memory, + }), + &format!("/proc/{}", pid.as_deref().unwrap_or("console")), + operations.clone(), + ), + false, + ); + mounts.mount( + "/run", + self.0 + .observer + .layer(Arc::new(self.0.run.clone()), "/run", operations), false, ); - mounts.mount("/run", Arc::new(self.0.run.clone()), false); Ok(mounts) } fn core_files(&self) -> Generated { diff --git a/cli/src/host.rs b/cli/src/host.rs index e0f4785..4b23ed4 100644 --- a/cli/src/host.rs +++ b/cli/src/host.rs @@ -37,6 +37,8 @@ pub(crate) struct Inner { /// processes, and upgrade tasks do nothing. pub(crate) shutdown: std::sync::atomic::AtomicBool, pub(crate) run: sausage_vfs::Memory, + pub(crate) watches: crate::watches::Watches, + pub(crate) observer: sausage_vfs::observe::Observer, pub(crate) graph: Mutex, pub(crate) activations: tokio::sync::Mutex<()>, pub(crate) watcher: Mutex>, @@ -91,6 +93,10 @@ impl Host { )) } fn with_parts(world: World, runtime: Arc, journal: journal::Journal) -> Self { + let watches = crate::watches::Watches::default(); + let events = watches.clone(); + let observer = + sausage_vfs::observe::Observer::new(move |path| crate::watches::changed(&events, path)); Self(Arc::new(Inner { world, runtime, @@ -109,6 +115,8 @@ impl Host { admission: tokio::sync::RwLock::new(()), shutdown: std::sync::atomic::AtomicBool::new(false), run: sausage_vfs::Memory::default(), + watches, + observer, graph: Mutex::new(crate::graph::Graph::default()), activations: tokio::sync::Mutex::new(()), watcher: Mutex::new(None), diff --git a/cli/src/lib.rs b/cli/src/lib.rs index ed684f5..b0d5e25 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -18,6 +18,7 @@ pub mod runtime; mod services; mod upgrades; mod watcher; +mod watches; pub mod world; pub use host::Host; pub use native::{NativeContext, NativeFuture, NativePlugin}; diff --git a/cli/src/routing.rs b/cli/src/routing.rs index 15a7a89..9489f9a 100644 --- a/cli/src/routing.rs +++ b/cli/src/routing.rs @@ -51,15 +51,20 @@ impl Host { .processes .get(pid) .and_then(|p| p.answer.as_ref().map(|_| (p.caller.clone(), p.id.clone()))); - if let Some((caller, id)) = target { - let notification = json!({ - "jsonrpc": "2.0", - "method": "core.notify", - "params": { "call": id, "name": name, "params": params } - }); - self.send(&caller, notification); + if let Some((caller, Some(id))) = target { + self.call_notify(&caller, &id, name, params); } } + /// Deliver a notification on an open core call using the same envelope as plugin progress. + pub(crate) fn call_notify(&self, caller: &str, id: &Value, name: &str, params: Value) { + self.send( + caller, + json!({ + "jsonrpc":"2.0", "method":"core.notify", + "params": {"call":id, "name":name, "params":params} + }), + ); + } /// Validate and route a request, then start its native or WASI process. pub(crate) fn route( &self, diff --git a/cli/src/watches.rs b/cli/src/watches.rs new file mode 100644 index 0000000..b6a1ab6 --- /dev/null +++ b/cli/src/watches.rs @@ -0,0 +1,178 @@ +//! Armed plugin-file watches. A Notify permit coalesces changes without losing a later change. +use crate::{Host, RpcResult, error}; +use sausage_vfs::observe::under; +use serde_json::{Value, json}; +use std::{ + collections::HashMap, + sync::{Arc, Mutex}, +}; +use tokio::sync::Notify; +use uuid::Uuid; + +pub(crate) struct Watch { + path: String, + changed: Arc, +} +pub(crate) type Watches = Arc>>; +pub(crate) fn changed(watches: &Watches, path: &str) { + let targets: Vec<_> = watches + .lock() + .unwrap() + .values() + // Removing/moving an ancestor also invalidates a watch inside that subtree. + .filter(|w| under(path, &w.path) || under(&w.path, path)) + .map(|w| w.changed.clone()) + .collect(); + for changed in targets { + changed.notify_one(); + } +} +struct Registration { + watches: Watches, + key: Uuid, +} +impl Drop for Registration { + fn drop(&mut self) { + self.watches.lock().unwrap().remove(&self.key); + } +} +fn path(caller: &str, path: &str) -> String { + let path = sausage_vfs::resolve("/", path); + let path = path.trim_end_matches('/'); + if path.is_empty() { + return "/".into(); + } + if under(path, "/self") { + format!("/proc/{caller}{}", &path[5..]) + } else { + path.into() + } +} +impl Host { + pub(crate) async fn watch(&self, caller: &str, id: Option, target: &str) -> RpcResult { + let id = id.ok_or_else(|| error(-32600, "watch needs a request ID"))?; + let mut ended = self + .0 + .state + .lock() + .unwrap() + .processes + .get(caller) + .filter(|p| p.info.state == "running") + .map(|p| p.ended.subscribe()) + .ok_or_else(|| error(-32600, "watch needs a running caller process"))?; + let changed = Arc::new(Notify::new()); + let registration = Registration { + watches: self.0.watches.clone(), + key: Uuid::now_v7(), + }; + self.0.watches.lock().unwrap().insert( + registration.key, + Watch { + path: path(caller, target), + changed: changed.clone(), + }, + ); + self.call_notify(caller, &id, "armed", json!({})); + loop { + tokio::select! { + biased; + _ = ended.wait_for(|ended| *ended) => return Ok(Value::Null), + _ = changed.notified() => self.call_notify(caller, &id, "changed", json!({})), + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + /// Pending changes coalesce; a change after consuming the last permit is retained. + #[tokio::test] + async fn coalescing_and_cleanup() { + let watches = Watches::default(); + let n = Arc::new(Notify::new()); + let r = Registration { + watches: watches.clone(), + key: Uuid::now_v7(), + }; + watches.lock().unwrap().insert( + r.key, + Watch { + path: "/mailbox".into(), + changed: n.clone(), + }, + ); + changed(&watches, "/mailboxes/a"); + assert!( + tokio::time::timeout(std::time::Duration::ZERO, n.notified()) + .await + .is_err() + ); + changed(&watches, "/mailbox/a"); + changed(&watches, "/mailbox/b"); + n.notified().await; + assert!( + tokio::time::timeout(std::time::Duration::ZERO, n.notified()) + .await + .is_err() + ); + changed(&watches, "/mailbox/c"); + n.notified().await; + drop(r); + assert!(watches.lock().unwrap().is_empty()); + changed(&watches, "/mailbox/d"); + assert!( + tokio::time::timeout(std::time::Duration::ZERO, n.notified()) + .await + .is_err() + ); + } + #[test] + fn self_alias_and_normalization() { + assert_eq!(path("writer", "/self/../self/a/"), "/proc/writer/a"); + assert_eq!(path("writer", "/selfish/a"), "/selfish/a"); + assert_eq!(path("writer", "/"), "/"); + } +} + +#[cfg(test)] +mod lifetime_tests { + use super::*; + /// The real caller's ended event releases the registration, without a cancellation RPC. + #[tokio::test] + async fn caller_death_removes_watch() { + let temp = tempfile::tempdir().unwrap(); + let runtime = crate::runtime::Runtime::new(&temp.path().join("cache")).unwrap(); + let host = Host::test_world(&temp.path().join("world"), runtime) + .await + .unwrap(); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + host.register_native( + "holder", + Arc::new(move |ctx, _, _| { + let tx = tx.clone(); + Box::pin(async move { + tx.send(ctx.pid.clone()).unwrap(); + std::future::pending().await + }) + }), + ); + let h = host.clone(); + let task = tokio::spawn(async move { h.call("holder.hold", json!({})).await }); + let pid = rx.recv().await.unwrap(); + let mut watch = Box::pin(host.watch(&pid, Some(json!("watch")), "/mailbox")); + assert!( + watch + .as_mut() + .poll(&mut std::task::Context::from_waker(std::task::Waker::noop())) + .is_pending() + ); + assert_eq!(host.0.watches.lock().unwrap().len(), 1); + host.call("core.kill", json!({"pid":pid})).await.unwrap(); + watch.await.unwrap(); + assert!(host.0.watches.lock().unwrap().is_empty()); + assert!(task.await.unwrap().is_err()); + host.stop().await; + } +} diff --git a/cli/tests/host/core.rs b/cli/tests/host/core.rs index 95707ad..1ba0285 100644 --- a/cli/tests/host/core.rs +++ b/cli/tests/host/core.rs @@ -208,7 +208,11 @@ async fn journal_recovers_after_sqlite_write_failure() { call(&h, "core.log", json!({})).await; let batches: Vec = std::fs::read_dir(w.0.join("logs/core/failed-batches")) .unwrap() - .map(|file| serde_json::from_slice(&std::fs::read(file.unwrap().path()).unwrap()).unwrap()) + .map(Result::unwrap) + // The writer atomically publishes .json files; its temporary files can + // still be incomplete while later process-exit entries are spooling. + .filter(|file| file.path().extension().is_some_and(|ext| ext == "json")) + .map(|file| serde_json::from_slice(&std::fs::read(file.path()).unwrap()).unwrap()) .collect(); assert!( batches @@ -1104,17 +1108,16 @@ async fn stop_closes_stdin_and_waits_for_exit() { let w = TestWorld::new(); let h = w.start().await; let hc = h.clone(); + let mut events = h.subscribe(); let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await }); + // Admission is not readiness: stop may otherwise close stdin before the + // request reaches the guest, so wait for the handler to enter its EOF loop. let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async { loop { - if let Some((pid, _)) = h - .processes() - .into_iter() - .find(|(_, p)| p.method == "wait_eof") - { - break pid; + let event = events.recv().await.unwrap(); + if event["params"]["name"] == "waiting-eof" { + break event["params"]["params"]["pid"].clone(); } - tokio::task::yield_now().await; } }) .await diff --git a/cli/tests/host/main.rs b/cli/tests/host/main.rs index f99b7a8..d0d6732 100644 --- a/cli/tests/host/main.rs +++ b/cli/tests/host/main.rs @@ -7,3 +7,4 @@ mod common; mod console; mod core; mod sdk; +mod watches; diff --git a/cli/tests/host/watches.rs b/cli/tests/host/watches.rs new file mode 100644 index 0000000..03c2a0f --- /dev/null +++ b/cli/tests/host/watches.rs @@ -0,0 +1,174 @@ +//! Real SDK watch callbacks and raw WASI filesystem mutations share the host namespace. +use super::common::{CACHE, probe, sdk_probe}; +use sausage::{Host, world::World}; +use serde_json::{Value, json}; +use std::time::Duration; + +// Same call bound as core.rs; no sleeps or polling. +async fn call(h: &Host, method: &str, p: Value) -> Value { + tokio::time::timeout(Duration::from_secs(20), h.call(method, p)) + .await + .unwrap() + .unwrap() +} +async fn event(h: &Host, rx: &mut tokio::sync::broadcast::Receiver, name: &str) -> Value { + let result = tokio::time::timeout(Duration::from_secs(20), async { + loop { + let e = rx.recv().await.unwrap(); + if e["params"]["name"] == name { + return e["params"]["params"].clone(); + } + } + }) + .await; + match result { + Ok(v) => v, + Err(e) => { + eprintln!( + "JOURNAL {}", + h.call("core.log", json!({"limit":100})).await.unwrap() + ); + panic!("{e}"); + } + } +} +async fn start() -> (tempfile::TempDir, Host) { + let dir = tempfile::tempdir().unwrap(); + let world = World::create(&dir.path().join("world")).unwrap(); + for (name, wasm) in [("probe", probe()), ("sdk", sdk_probe())] { + std::fs::create_dir_all(world.plugin_dir(name)).unwrap(); + std::fs::copy(wasm, world.plugin_dir(name).join("plugin.wasm")).unwrap(); + } + let host = Host::start_with_cache(&world.root, std::path::Path::new(CACHE)) + .await + .unwrap(); + (dir, host) +} +fn watch(h: &Host, path: &str) -> tokio::task::JoinHandle { + let (h, path) = (h.clone(), path.to_owned()); + tokio::spawn(async move { h.call("sdk.watch", json!({"path":path})).await }) +} +async fn mutate(h: &Host, op: &str, path: &str) { + call(h, "probe.mutate", json!({"op":op,"path":path})).await; +} +/// PROTOCOL 4: pre-arm state is in the first scan; each post-scan mutation wakes +/// the SDK without completing the watch. Covers descriptor/stream/truncate paths. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn armed_scan_and_later_changes() { + let (_dir, h) = start().await; + mutate(&h, "mkdir", "/state/mailbox").await; + call( + &h, + "probe.write", + json!({"path":"/state/mailbox/before", "text":"before"}), + ) + .await; + let mut rx = h.subscribe(); + let task = watch(&h, "/state/mailbox/../mailbox/"); + let armed = event(&h, &mut rx, "armed").await; + assert_eq!(armed["entries"], json!(["before"])); + assert!(!task.is_finished()); + for (op, expected) in [ + ("descriptor", "descriptor"), + ("stream", "streamptor"), + ("append", "streamptorappend"), + ("size", ""), + ("descriptor", "descriptor"), + ("truncate", ""), + ] { + mutate(&h, op, "/state/mailbox/before").await; + let mut scan = event(&h, &mut rx, "changed").await; + while scan["text"] != expected { + scan = event(&h, &mut rx, "changed").await; + } + } + mutate(&h, "burst", "/state/mailbox").await; + let mut scanned = event(&h, &mut rx, "changed").await; + while !scanned["entries"].as_array().unwrap().contains(&json!("4")) { + scanned = event(&h, &mut rx, "changed").await; + } + while rx.try_recv().is_ok() {} + mutate(&h, "delete", "/state/mailbox/4").await; + let mut scanned = event(&h, &mut rx, "changed").await; + while scanned["entries"].as_array().unwrap().contains(&json!("4")) { + scanned = event(&h, &mut rx, "changed").await; + } + call(&h, "core.kill", json!({"pid":armed["pid"]})).await; + assert!(task.await.unwrap().is_err()); + h.stop().await; +} +/// PROTOCOL 4: atomic publication, rename source/destination, mounted paths, +/// absent paths, and caller-death cleanup. No external host writes are required. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn rename_delete_and_mounts() { + let (_dir, h) = start().await; + for base in ["/state/inbox", "/run/inbox"] { + let mut rx = h.subscribe(); + let task = watch(&h, base); + let armed = event(&h, &mut rx, "armed").await; + assert_eq!(armed["entries"], json!([])); + let staging = format!("{base}-staging"); + mutate(&h, "mkdir", &staging).await; + call( + &h, + "probe.mutate", + json!({"op":"rename","path":staging,"to":base}), + ) + .await; + event(&h, &mut rx, "changed").await; + mutate(&h, "atomic", &format!("{base}/published")).await; + let mut scan = event(&h, &mut rx, "changed").await; + while !scan["entries"] + .as_array() + .unwrap() + .contains(&json!("published")) + { + scan = event(&h, &mut rx, "changed").await; + } + assert_eq!( + call( + &h, + "probe.read", + json!({"path":format!("{base}/published")}) + ) + .await, + "complete" + ); + while rx.try_recv().is_ok() {} + call( + &h, + "probe.mutate", + json!({"op":"rename","path":base,"to":format!("{base}-moved")}), + ) + .await; + let mut scan = event(&h, &mut rx, "changed").await; + while scan["entries"] != json!([]) { + scan = event(&h, &mut rx, "changed").await; + } + call(&h, "core.kill", json!({"pid":armed["pid"]})).await; + assert!(task.await.unwrap().is_err()); + } + // /self watches bind to this caller, and mutations are visible via /proc/. + let mut rx = h.subscribe(); + let task = watch(&h, "/self"); + let armed = event(&h, &mut rx, "armed").await; + assert!( + armed["entries"] + .as_array() + .unwrap() + .contains(&json!("owned")) + ); + event(&h, &mut rx, "changed").await; + assert_eq!( + call( + &h, + "probe.read", + json!({"path":format!("/proc/{}/owned", armed["pid"].as_str().unwrap())}) + ) + .await, + "visible via proc" + ); + call(&h, "core.kill", json!({"pid":armed["pid"]})).await; + assert!(task.await.unwrap().is_err()); + h.stop().await; +} diff --git a/cli/tests/plugins/probe/src/main.rs b/cli/tests/plugins/probe/src/main.rs index c20bd7b..7ba2cf9 100644 --- a/cli/tests/plugins/probe/src/main.rs +++ b/cli/tests/plugins/probe/src/main.rs @@ -83,6 +83,9 @@ fn unknown_id() { } fn wait_eof(lines: &mut Lines) { std::fs::write("/state/waiting", "yes").unwrap(); + send(json!({"jsonrpc":"2.0","method":"core.notify","params":{ + "name":"waiting-eof", "params":{"pid":std::env::var("SAUSAGE_PID").unwrap()} + }})); for _ in lines {} std::fs::write("/state/stopped", "yes").unwrap(); } @@ -468,6 +471,7 @@ fn dispatch(request: &Value, lines: &mut Lines) -> Option { return None; } "network" => network(params), + "mutate" => mutate(params), "read" => read_file(params), "write" => write_file(params), "raw_call" => { @@ -518,3 +522,68 @@ fn main() { })); } } + +/// Separate mutation calls let the host await a scan before each subsequent change. +fn mutate(p: &Value) -> Value { + use wasip2::filesystem::{ + preopens, + types::{DescriptorFlags as D, OpenFlags as O, PathFlags as P}, + }; + let path = p["path"].as_str().unwrap(); + match p["op"].as_str().unwrap() { + "mkdir" => std::fs::create_dir_all(path).unwrap(), + "rename" => std::fs::rename(path, p["to"].as_str().unwrap()).unwrap(), + "delete" => std::fs::remove_file(path).unwrap(), + "rmdir" => std::fs::remove_dir(path).unwrap(), + "atomic" => { + let temp = format!("{path}.tmp"); + std::fs::write(&temp, "complete").unwrap(); + std::fs::rename(temp, path).unwrap(); + } + "burst" => { + for i in 0..5 { + std::fs::write(format!("{path}/{i}"), "complete").unwrap(); + } + } + op => { + let root = preopens::get_directories().remove(0).0; + let file = root + .open_at( + P::empty(), + path.trim_start_matches('/'), + O::empty(), + D::READ | D::WRITE, + ) + .unwrap(); + match op { + "descriptor" => { + file.write(b"descriptor", 0).unwrap(); + } + "stream" => { + file.write_via_stream(0) + .unwrap() + .blocking_write_and_flush(b"stream") + .unwrap(); + } + "append" => { + file.append_via_stream() + .unwrap() + .blocking_write_and_flush(b"append") + .unwrap(); + } + "size" => file.set_size(0).unwrap(), + "truncate" => { + root.open_at( + P::empty(), + path.trim_start_matches('/'), + O::TRUNCATE, + D::WRITE, + ) + .unwrap(); + } + _ => panic!("unknown mutation"), + } + } + } + json!({}) +} diff --git a/sdk/src/calls.rs b/sdk/src/calls.rs index c8be755..04e2c6e 100644 --- a/sdk/src/calls.rs +++ b/sdk/src/calls.rs @@ -26,6 +26,17 @@ impl Call { self } /// Receive progress notifications before the result arrives. + /// + /// For `core.watch`, poll the call future alongside work in the handler, then + /// drop it when the handler returns. Do not put it in `background()` just to + /// keep the watch alive: background tasks keep the process alive, while the + /// watch only ends when the process ends. Dropping the future does not cancel + /// the host watch, but it lets an otherwise idle process exit normally. + /// + /// Scan only after `armed`, and rescan on `changed`. Events may coalesce, but + /// are not rate-limited: watch the narrowest useful path to avoid rescanning + /// for unrelated writes (for example, watch a mailbox, not all session data). + /// Do not use endpoint connect retries to wait for file publication. pub fn on_notify(mut self, f: impl FnMut(&str, Value) + 'static) -> Self { self.notify = Some(Box::new(f)); self diff --git a/sdk/tests/plugins/probe/src/main.rs b/sdk/tests/plugins/probe/src/main.rs index 3de1c20..5ee6049 100644 --- a/sdk/tests/plugins/probe/src/main.rs +++ b/sdk/tests/plugins/probe/src/main.rs @@ -2,6 +2,12 @@ use sausage_plugin::serde_json::{Value, json}; use sausage_plugin::{Ctx, Result, call, lock, rpc, schema}; struct Probe; +/// A virtual path to observe. +#[schema] +struct WatchPath { + /// Directory or file, not necessarily present yet. + path: String, +} /// The values to add. #[schema] struct Add { @@ -30,6 +36,68 @@ struct Echo { } #[rpc(grace_seconds = 12,endpoints(control=Control))] impl Probe { + /// Scan only after the watch is armed; keep the call open until process death. + async fn watch(&self, p: WatchPath) -> Result<()> { + let path = p.path.clone(); + call("core.watch", json!({"path":p.path})) + .on_notify(move |name, _| { + if name == "armed" && path == "/self" { + std::fs::write("/self/owned", "visible via proc").unwrap(); + } + let mut entries: Vec<_> = std::fs::read_dir(&path) + .into_iter() + .flatten() + .filter_map(std::result::Result::ok) + .map(|e| e.file_name().into_string().unwrap()) + .collect(); + entries.sort(); + let text = std::fs::read_to_string(format!("{path}/before")).ok(); + Ctx::default() + .notify( + name, + json!({"pid":sausage_plugin::pid(), "entries":entries, "text":text}), + ) + .unwrap(); + }) + .await?; + Ok(()) + } + + /// Poll the watch in the handler, then return idle without a background task. + async fn watch_once(&self, p: WatchPath) -> Result<()> { + use std::{ + cell::RefCell, + rc::Rc, + task::{Poll, Waker}, + }; + let state = Rc::new(RefCell::new((false, None::))); + let received = state.clone(); + let mut watch = std::pin::pin!(std::future::IntoFuture::into_future( + call("core.watch", json!({"path":p.path})).on_notify(move |name, _| { + if name == "armed" { + let mut state = received.borrow_mut(); + state.0 = true; + if let Some(waker) = state.1.take() { + waker.wake(); + } + } + }) + )); + std::future::poll_fn(|cx| { + if let Poll::Ready(result) = watch.as_mut().poll(cx) { + return Poll::Ready(result.map(|_| ())); + } + let mut state = state.borrow_mut(); + state.1 = Some(cx.waker().clone()); + if state.0 { + Poll::Ready(Ok(())) + } else { + Poll::Pending + } + }) + .await + } + /// Publish a control endpoint until EOF. async fn serve(&self) -> Result<()> { sausage_plugin::listen("/endpoint", std::rc::Rc::new(Control)).await diff --git a/sdk/tests/wasi.rs b/sdk/tests/wasi.rs index 251a3de..1c71860 100644 --- a/sdk/tests/wasi.rs +++ b/sdk/tests/wasi.rs @@ -108,6 +108,15 @@ fn protocol() { } p.finishes(); } + // A handler-owned watch does not keep an idle process alive after return, + // even though the host never answers the watch and stdin remains open. + let mut p = start(&wasm, &dir); + p.request("watch_once", json!({"path":"/mailbox"})); + let watch = p.read(); + assert_eq!(watch["method"], "core.watch"); + p.send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":watch["id"],"name":"armed","params":{}}})); + assert_eq!(p.read()["result"], Value::Null); + p.finishes(); let mut p = start(&wasm, &dir); p.request("rpc.discover", json!({})); let d = p.read(); diff --git a/vfs/src/lib.rs b/vfs/src/lib.rs index 324a83c..b6065d4 100644 --- a/vfs/src/lib.rs +++ b/vfs/src/lib.rs @@ -5,6 +5,7 @@ pub mod generated; pub mod host; pub mod memory; pub mod mounts; +pub mod observe; pub mod process; mod readonly; mod streams; diff --git a/vfs/src/observe.rs b/vfs/src/observe.rs new file mode 100644 index 0000000..e7f67d0 --- /dev/null +++ b/vfs/src/observe.rs @@ -0,0 +1,615 @@ +//! Observes successful namespace and open-file mutations, even if their caller is cancelled. +use crate::*; +use std::{ + collections::BTreeMap, + sync::{Mutex, Weak}, +}; +use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock}; + +type Namespace = Mutex>>>; +enum Held { + Read { _guard: OwnedRwLockReadGuard<()> }, + Write { _guard: OwnedRwLockWriteGuard<()> }, +} + +type TrackedPath = Mutex>; +type Path = Arc; +type Callback = dyn Fn(&str) + Send + Sync; +/// 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, + 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. + namespace: Arc, +} +impl Observer { + pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self { + Self { + changed: Arc::new(changed), + paths: Arc::default(), + namespace: Arc::default(), + } + } + /// 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 { + layer, + prefix: prefix.trim_end_matches('/').into(), + observer: self.clone(), + process, + }) + } + /// Opens share their path and ancestors; rename/remove exclusively hold their + /// endpoints, sharing ancestors. Sorted acquisition avoids cross-rename deadlocks. + /// No descriptor IO or notification holds these guards. + async fn order(&self, endpoints: &[(&str, bool)]) -> Vec { + let mut modes = BTreeMap::new(); + for &(path, exclusive) in endpoints { + modes + .entry(path.to_owned()) + .and_modify(|mode| *mode |= exclusive) + .or_insert(exclusive); + let mut ancestor = path; + while let Some((parent, _)) = ancestor.rsplit_once('/') { + let parent = if parent.is_empty() { "/" } else { parent }; + modes.entry(parent.to_owned()).or_insert(false); + if parent == "/" { + break; + } + ancestor = parent; + } + } + let locks: Vec<_> = { + let mut namespace = self.namespace.lock().unwrap(); + namespace.retain(|_, lock| lock.strong_count() != 0); + modes + .into_iter() + .map(|(path, exclusive)| { + let lock = namespace.entry(path).or_default(); + let strong = lock.upgrade().unwrap_or_else(|| { + let strong = Arc::new(RwLock::new(())); + *lock = Arc::downgrade(&strong); + strong + }); + (strong, exclusive) + }) + .collect() + }; + let mut held = Vec::new(); + for (lock, exclusive) in locks { + held.push(if exclusive { + Held::Write { + _guard: lock.write_owned().await, + } + } else { + Held::Read { + _guard: lock.read_owned().await, + } + }); + } + held + } + fn path(&self, path: String) -> Path { + let path = Arc::new(Mutex::new(Some(path))); + let mut paths = self.paths.lock().unwrap(); + paths.retain(|p| p.strong_count() != 0); + paths.push(Arc::downgrade(&path)); + path + } + fn moved(&self, from: &str, to: Option<&str>) { + let mut paths = self.paths.lock().unwrap(); + paths.retain(|p| { + let Some(p) = p.upgrade() else { return false }; + let mut path = p.lock().unwrap(); + if let Some(old) = path.as_deref() { + if under(old, from) { + *path = to.map(|to| format!("{to}{}", &old[from.len()..])); + } else if to.is_some_and(|to| under(old, to)) { + // A replaced destination's old descriptor no longer names a visible file. + *path = None; + } + } + true + }); + } +} +/// Component-aware descendant matching, including the path itself. +pub fn under(path: &str, base: &str) -> bool { + base == "/" || path == base || path.strip_prefix(base).is_some_and(|s| s.starts_with('/')) +} +struct Observed { + layer: Arc, + prefix: String, + observer: Observer, + process: Process, +} +impl Observed { + fn path(&self, path: &str) -> String { + let path = format!("{}{}", self.prefix, path.trim_end_matches('/')); + if path.is_empty() { "/".into() } else { path } + } +} +struct Handle { + file: Arc, + path: Path, + observer: Observer, + process: Process, +} +impl Handle { + async fn mutate( + &self, + f: impl Future> + Send + 'static, + ) -> Result { + let (path, observer) = (self.path.clone(), self.observer.clone()); + self.process + .operation(async move { + let result = f.await?; + let path = path.lock().unwrap().clone(); + if let Some(path) = path { + (observer.changed)(&path); + } + Ok(result) + }) + .await + } +} +#[async_trait] +impl Layer for Observed { + async fn stat(&self, path: &str, follow: bool) -> Result { + self.layer.stat(path, follow).await + } + async fn readlink(&self, path: &str) -> Result { + self.layer.readlink(path).await + } + async fn list(&self, path: &str) -> Result> { + self.layer.list(path).await + } + async fn open(&self, path: &str, options: Open) -> Result> { + let (layer, local, path, observer, process) = ( + self.layer.clone(), + path.to_owned(), + self.path(path), + self.observer.clone(), + self.process.clone(), + ); + self.process + .operation(async move { + let order = observer.order(&[(&path, false)]).await; + let file = layer.open(&local, options).await?; + let handle = Arc::new(Handle { + file, + path: observer.path(path.clone()), + observer: observer.clone(), + process, + }); + drop(order); + if options.create || options.truncate { + (observer.changed)(&path); + } + Ok(handle as Arc) + }) + .await + } + async fn mkdir(&self, path: &str) -> Result<()> { + let (layer, local, path, observer) = ( + self.layer.clone(), + path.to_owned(), + self.path(path), + self.observer.clone(), + ); + self.process + .operation(async move { + layer.mkdir(&local).await?; + (observer.changed)(&path); + Ok(()) + }) + .await + } + async fn remove_file(&self, path: &str) -> Result<()> { + self.remove(path, false).await + } + async fn remove_dir(&self, path: &str) -> Result<()> { + self.remove(path, true).await + } + async fn rename(&self, from: &str, to: &str) -> Result<()> { + let (layer, a, b, from, to, observer) = ( + self.layer.clone(), + from.to_owned(), + to.to_owned(), + self.path(from), + self.path(to), + self.observer.clone(), + ); + self.process + .operation(async move { + let order = observer.order(&[(&from, true), (&to, true)]).await; + layer.rename(&a, &b).await?; + if from != to { + observer.moved(&from, Some(&to)); + } + drop(order); + (observer.changed)(&from); + (observer.changed)(&to); + Ok(()) + }) + .await + } +} +impl Observed { + async fn remove(&self, path: &str, directory: bool) -> Result<()> { + let (layer, local, path, observer) = ( + self.layer.clone(), + path.to_owned(), + self.path(path), + self.observer.clone(), + ); + self.process + .operation(async move { + let order = observer.order(&[(&path, true)]).await; + if directory { + layer.remove_dir(&local).await?; + } else { + layer.remove_file(&local).await?; + } + observer.moved(&path, None); + drop(order); + (observer.changed)(&path); + Ok(()) + }) + .await + } +} +#[async_trait] +impl File for Handle { + async fn stat(&self) -> Result { + self.file.stat().await + } + async fn read(&self, offset: u64, length: usize) -> Result> { + self.file.read(offset, length).await + } + async fn write(&self, offset: u64, bytes: Vec) -> Result { + let f = self.file.clone(); + self.mutate(async move { f.write(offset, bytes).await }) + .await + } + async fn append(&self, bytes: Vec) -> Result { + let f = self.file.clone(); + self.mutate(async move { f.append(bytes).await }).await + } + async fn set_size(&self, size: u64) -> Result<()> { + let f = self.file.clone(); + self.mutate(async move { f.set_size(size).await }).await + } + async fn sync(&self) -> Result<()> { + self.file.sync().await + } + async fn set_times( + &self, + accessed: Option, + modified: Option, + ) -> Result<()> { + let f = self.file.clone(); + self.mutate(async move { f.set_times(accessed, modified).await }) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::sync::mpsc; + #[test] + fn component_matching() { + assert!(under("/mailbox/a", "/mailbox")); + assert!(under("/mailbox", "/mailbox")); + assert!(under("/anything", "/")); + assert!(!under("/mailboxes/a", "/mailbox")); + } + /// Every visible mutation is delivered after the layer call, including rebased + /// open handles in another process's namespace. Removed handles stay usable. + #[tokio::test] + async fn handles_follow_rename_and_unlink() { + for host in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let process = Process::default(); + let layer: Arc = if host { + Arc::new(Host::new(temp.path(), process.clone()).unwrap()) + } else { + Arc::new(Memory::default()) + }; + let (tx, mut rx) = mpsc::unbounded_channel(); + let observer = Observer::new(move |p| { + tx.send(p.to_owned()).unwrap(); + }); + let a = observer.layer(layer.clone(), "/run", process.clone()); + let b = observer.layer(layer, "/run", process); + a.mkdir("/dir").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/dir"); + let f = a + .open( + "/dir/a", + Open { + create: true, + write: true, + ..Open::default() + }, + ) + .await + .unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/dir/a"); + b.rename("/dir", "/moved").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/dir"); + assert_eq!(rx.recv().await.unwrap(), "/run/moved"); + f.write(0, b"new".to_vec()).await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/moved/a"); + assert_eq!( + b.open("/moved/a", Open::default()) + .await + .unwrap() + .read(0, 3) + .await + .unwrap(), + b"new" + ); + let old = b + .open( + "/old", + Open { + create: true, + write: true, + ..Open::default() + }, + ) + .await + .unwrap(); + rx.recv().await.unwrap(); + b.rename("/moved/a", "/old").await.unwrap(); + rx.recv().await.unwrap(); + rx.recv().await.unwrap(); + old.write(0, b"invisible".to_vec()).await.unwrap(); + assert!(rx.try_recv().is_err()); + f.append(b"!".to_vec()).await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/old"); + b.remove_file("/old").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/old"); + f.set_size(0).await.unwrap(); + assert!(rx.try_recv().is_err()); + assert!(b.remove_file("/missing").await.is_err()); + assert!(rx.try_recv().is_err()); + } + } + /// A pending exclusive namespace change must not block a sibling's opens, + /// even though Tokio's RwLock prefers queued writers over new readers. + #[tokio::test] + async fn namespace_order_is_scoped_and_shared() { + async fn ready(future: impl Future) -> T { + let mut future = std::pin::pin!(future); + std::future::poll_fn(|cx| match future.as_mut().poll(cx) { + std::task::Poll::Ready(value) => std::task::Poll::Ready(value), + std::task::Poll::Pending => panic!("unrelated namespace operation blocked"), + }) + .await + } + let observer = Observer::new(|_| {}); + let open = observer.order(&[("/run/slow/file", false)]).await; + let concurrent_open = ready(observer.order(&[("/run/slow/file", false)])).await; + let mut rename = + std::pin::pin!(observer.order(&[("/run/slow", true), ("/run/moved", true)])); + std::future::poll_fn(|cx| { + assert!(rename.as_mut().poll(cx).is_pending()); + std::task::Poll::Ready(()) + }) + .await; + // Same mounted layer, different subtree; also a different mounted layer. + let sibling_open = ready(observer.order(&[("/run/sibling/file", false)])).await; + let host_open = ready(observer.order(&[("/state/file", false)])).await; + drop(sibling_open); + let sibling_rename = + ready(observer.order(&[("/run/sibling", true), ("/run/other", true)])).await; + drop((sibling_rename, host_open, concurrent_open, open)); + let renamed = ready(rename).await; + // A descendant open waits for its ancestor's rename until rebasing is done. + let mut descendant = std::pin::pin!(observer.order(&[("/run/moved/file", false)])); + std::future::poll_fn(|cx| { + assert!(descendant.as_mut().poll(cx).is_pending()); + std::task::Poll::Ready(()) + }) + .await; + drop(renamed); + drop(ready(descendant).await); + drop( + observer + .order(&[("/run/moved/file", true), ("/run/moved", true)]) + .await, + ); + drop( + observer + .order(&[("/run/moved", true), ("/run/moved", true)]) + .await, + ); + // Dead weak entries are pruned, rather than retaining every path ever opened. + observer.order(&[("/", false)]).await; + assert_eq!(observer.namespace.lock().unwrap().len(), 1); + } + + /// Pause an open after obtaining its real identity, before registering its + /// observed handle, like a slow host syscall or /self/log query. + struct SlowOpen { + memory: Arc, + gate: Mutex< + Option<( + tokio::sync::oneshot::Sender<()>, + tokio::sync::oneshot::Receiver<()>, + )>, + >, + } + #[async_trait] + impl Layer for SlowOpen { + async fn stat(&self, p: &str, follow: bool) -> Result { + self.memory.stat(p, follow).await + } + async fn list(&self, p: &str) -> Result> { + self.memory.list(p).await + } + async fn open(&self, p: &str, options: Open) -> Result> { + let f = self.memory.open(p, options).await?; + let gate = self.gate.lock().unwrap().take(); + if let Some((started, release)) = gate { + started.send(()).unwrap(); + release.await.unwrap(); + } + Ok(f) + } + async fn mkdir(&self, p: &str) -> Result<()> { + self.memory.mkdir(p).await + } + async fn remove_file(&self, p: &str) -> Result<()> { + self.memory.remove_file(p).await + } + async fn remove_dir(&self, p: &str) -> Result<()> { + self.memory.remove_dir(p).await + } + async fn rename(&self, a: &str, b: &str) -> Result<()> { + self.memory.rename(a, b).await + } + } + #[tokio::test] + async fn blocked_io_does_not_stall_other_processes() { + let memory = Arc::new(Memory::default()); + memory.mkdir("/slow").await.unwrap(); + let options = Open { + create: true, + write: true, + ..Open::default() + }; + let raw = memory.open("/slow/file", options).await.unwrap(); + let (started, opening) = tokio::sync::oneshot::channel(); + let (release, released) = tokio::sync::oneshot::channel(); + let (tx, mut rx) = mpsc::unbounded_channel(); + let observer = Observer::new(move |p| { + tx.send(p.to_owned()).unwrap(); + }); + let slow = observer.layer( + Arc::new(SlowOpen { + memory: memory.clone(), + gate: Mutex::new(Some((started, released))), + }), + "/run", + Process::default(), + ); + let other = observer.layer(memory, "/run", Process::default()); + let opening_task = tokio::spawn(async move { + slow.open( + "/slow/file", + Open { + write: true, + ..Open::default() + }, + ) + .await + }); + opening.await.unwrap(); + let renamer = other.clone(); + let renamed = tokio::spawn(async move { renamer.rename("/slow", "/moved").await }); + // These must complete while the first process's open is still blocked, + // and a conflicting rename is waiting to rebase its registered handle. + other + .open( + "/", + Open { + directory: true, + ..Open::default() + }, + ) + .await + .unwrap(); + let f = other.open("/other", options).await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/other"); + f.write(0, b"visible".to_vec()).await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/other"); + other.mkdir("/sibling").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/sibling"); + other.rename("/other", "/sibling/file").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/other"); + assert_eq!(rx.recv().await.unwrap(), "/run/sibling/file"); + other.remove_file("/sibling/file").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/sibling/file"); + assert!(!renamed.is_finished()); + release.send(()).unwrap(); + let opened = opening_task.await.unwrap().unwrap(); + renamed.await.unwrap().unwrap(); + rx.recv().await.unwrap(); + rx.recv().await.unwrap(); + opened.set_size(0).await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/moved/file"); + + // No namespace guard surrounds descriptor IO: a rename completes while + // this write is held in flight. Its notification uses the rebased path. + let handle = Handle { + file: raw.clone(), + path: observer.path("/run/moved/file".into()), + observer, + process: Process::default(), + }; + let (started, writing) = tokio::sync::oneshot::channel(); + let (release, released) = tokio::sync::oneshot::channel(); + let write = tokio::spawn(async move { + handle + .mutate(async move { + started.send(()).unwrap(); + released.await.unwrap(); + raw.write(0, b"after rename".to_vec()).await + }) + .await + }); + writing.await.unwrap(); + other.rename("/moved", "/again").await.unwrap(); + rx.recv().await.unwrap(); + rx.recv().await.unwrap(); + other.open("/again/file", Open::default()).await.unwrap(); + release.send(()).unwrap(); + write.await.unwrap().unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/again/file"); + } + + /// Killing a waiting writer does not discard observation of a completed mutation; + /// the process's idle barrier includes the callback, not only the system call. + #[tokio::test] + async fn cancellation_keeps_mutation_and_observation() { + let (tx, mut rx) = mpsc::unbounded_channel(); + let observer = Observer::new(move |p| { + tx.send(p.to_owned()).unwrap(); + }); + let process = Process::default(); + let layer = observer.layer(Arc::new(Memory::default()), "/", process.clone()); + let order = observer.order(&[("/published", true)]).await; + let (started, ready) = tokio::sync::oneshot::channel(); + let p = process.clone(); + let task = tokio::spawn(async move { + // Signal from inside the tracked operation, not before its guard exists. + p.operation(async move { + started.send(()).unwrap(); + layer + .open( + "/published", + Open { + create: true, + write: true, + ..Open::default() + }, + ) + .await + }) + .await + }); + ready.await.unwrap(); + task.abort(); + let _ = task.await; + drop(order); + process.idle().await; + assert_eq!(rx.try_recv().unwrap(), "/published"); + } +} diff --git a/vfs/src/process.rs b/vfs/src/process.rs index 2cffbd8..a73735a 100644 --- a/vfs/src/process.rs +++ b/vfs/src/process.rs @@ -36,7 +36,22 @@ impl Process { .await .map_err(std::io::Error::other)? } - /// Wait for all outstanding system calls. + /// Complete an asynchronous mutation and its observation even if the guest is cancelled. + /// Its guard participates in idle(), so no mutation outlives process cleanup. + pub async fn operation( + &self, + operation: impl std::future::Future> + Send + 'static, + ) -> std::io::Result { + self.0.count.fetch_add(1, Ordering::SeqCst); + let guard = Guard(self.0.clone()); + tokio::spawn(async move { + let _guard = guard; + operation.await + }) + .await + .map_err(std::io::Error::other)? + } + /// Wait for all outstanding system calls and observed mutations. /// First abort the guest task. Then await its JoinHandle, or drop its store. /// Only then await idle before releasing locks. Abort alone does not stop a running poll. /// This ordering also prevents later memory writes, which need no blocking guard.