From 9cbd4f94aabab440773756af2c3655206e6053a7 Mon Sep 17 00:00:00 2001 From: SAU-357 implementer Date: Fri, 30 Oct 2026 20:03:12 +0000 Subject: [PATCH 1/4] Use five-second upgrade grace for every plugin Ignore legacy grace metadata in both activation paths. Keep superseded delivery, admission ordering, discovery exclusion and the separate EOF stop budget. Document the uniform policy and cover both windows and opt-in drain. Card: SAU-357 --- PROTOCOL.md | 10 +- cli/src/activation.rs | 5 - cli/src/graph.rs | 11 +-- cli/src/upgrades.rs | 14 +-- cli/src/upgrades/tests.rs | 113 ++++++++++++++++++++++ cli/tests/host/activation.rs | 114 +++++++++++++++++++++-- cli/tests/host/sdk.rs | 50 ++++++++++ cli/tests/plugins/activation/src/main.rs | 8 ++ sdk/src/lib.rs | 4 +- sdk/tests/plugins/probe/src/main.rs | 6 ++ 10 files changed, 302 insertions(+), 33 deletions(-) create mode 100644 cli/src/upgrades/tests.rs diff --git a/PROTOCOL.md b/PROTOCOL.md index e9e6b6b..ca9b5bc 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -159,8 +159,11 @@ active (section 10.3): continue to work. It can stop accepting work, finish its in-flight calls, and exit early. Ignoring the notice changes nothing. The grace period is the upper bound before the core stops the process (section 3.6); its - default is 60 seconds. A plugin can declare its own value in its old - `rpc.discover` document (`x-sausage-grace`). + duration is 5 seconds for every plugin. Discovery metadata, including an + explicit `x-sausage-grace`, does not change it. After grace, stdin closes + and the separate 5-second stop budget applies (section 3.6): at most two + 5-second windows. Discovery-only processes do not receive this notice or + enter upgrade grace; they follow the discovery lifetime (section 10.1). 3. When the old process of a service has ended, the core starts the service again with the new version. 4. When the last process of an older version has ended, the core starts a @@ -391,7 +394,8 @@ Params are by name. The document can have these extension fields: `methods`. - `x-sausage-imports`: for each provider, the part of its document that the plugin uses. -- `x-sausage-grace`: the grace period in seconds. +- `x-sausage-grace`: legacy grace declaration in seconds. The core ignores + this field; upgrade grace is always 5 seconds (section 3.7). The schema of params describes what the provider can read. The schema of a result describes what the provider writes. Descriptions are part of the diff --git a/cli/src/activation.rs b/cli/src/activation.rs index 2729ff3..cbd561e 100644 --- a/cli/src/activation.rs +++ b/cli/src/activation.rs @@ -279,15 +279,10 @@ impl Host { .get(&candidate.plugin) .filter(|old| !old.hash.is_empty() && old.hash != candidate.interface.hash) { - let grace = old.document["x-sausage-grace"] - .as_f64() - .filter(|value| value.is_finite() && *value >= 0.0) - .unwrap_or(60.0); upgrades.push(self.snapshot_upgrade( &candidate.plugin, old.hash.clone(), &candidate.interface.hash, - grace, )); } self.0.journal.write( diff --git a/cli/src/graph.rs b/cli/src/graph.rs index fb551c7..38ed297 100644 --- a/cli/src/graph.rs +++ b/cli/src/graph.rs @@ -223,16 +223,7 @@ impl Host { if let Some(old) = old_plugins.get(plugin).filter(|old| { !old.hash.is_empty() && old.hash != current.hash && current.error.is_none() }) { - let grace = old.document["x-sausage-grace"] - .as_f64() - .filter(|value| value.is_finite() && *value >= 0.0) - .unwrap_or(60.0); - upgrades.push(self.snapshot_upgrade( - plugin, - old.hash.clone(), - ¤t.hash, - grace, - )); + upgrades.push(self.snapshot_upgrade(plugin, old.hash.clone(), ¤t.hash)); } } self.publish_graph(plugins); diff --git a/cli/src/upgrades.rs b/cli/src/upgrades.rs index bfa1783..3bbc866 100644 --- a/cli/src/upgrades.rs +++ b/cli/src/upgrades.rs @@ -4,6 +4,12 @@ use serde_json::json; use std::{collections::BTreeMap, sync::atomic::Ordering, time::Duration}; use tokio::sync::watch; +/// Uniform host policy (PROTOCOL 3.7), independent of discovery metadata. +const UPGRADE_GRACE: Duration = Duration::from_secs(5); + +#[cfg(test)] +mod tests; + pub(crate) struct OldProcess { pid: String, version: String, @@ -13,7 +19,6 @@ pub(crate) struct Upgrade { plugin: String, previous: String, current: String, - grace: Duration, processes: Vec, services: Vec<(Service, Option>)>, } @@ -24,7 +29,6 @@ impl Host { plugin: &str, previous: String, current: &str, - grace: f64, ) -> Upgrade { let mut state = self.0.state.lock().unwrap(); let processes = state @@ -69,7 +73,6 @@ impl Host { plugin: plugin.into(), previous, current: current.into(), - grace: Duration::from_secs_f64(grace), processes, services, } @@ -92,7 +95,7 @@ impl Host { "core", &upgrade.plugin, "core", - &format!("upgrade grace {} seconds", upgrade.grace.as_secs_f64()), + &format!("upgrade grace {} seconds", UPGRADE_GRACE.as_secs()), ); let mut versions: BTreeMap>> = BTreeMap::new(); versions.entry(upgrade.previous).or_default(); @@ -102,12 +105,11 @@ impl Host { .or_default() .push(process.ended.clone()); let host = host.clone(); - let grace = upgrade.grace; tokio::spawn(async move { let mut ended = process.ended; tokio::select! { _ = ended.wait_for(|ended|*ended) => (), - _ = tokio::time::sleep(grace) => { host.begin_stop(&process.pid); } + _ = tokio::time::sleep(UPGRADE_GRACE) => { host.begin_stop(&process.pid); } } }); } diff --git a/cli/src/upgrades/tests.rs b/cli/src/upgrades/tests.rs new file mode 100644 index 0000000..c2e6f81 --- /dev/null +++ b/cli/src/upgrades/tests.rs @@ -0,0 +1,113 @@ +//! Controlled-clock coverage of both upgrade windows and discovery exclusion. +use super::*; +use crate::{ + process::{Process, ProcessInfo}, + runtime::Runtime, + world::World, +}; +use tokio::{ + sync::mpsc, + task::JoinHandle, + time::{Instant, advance}, +}; + +/// A process with active work that ignores both superseded and EOF. +fn process( + host: &Host, + pid: &str, + method: &str, +) -> (mpsc::UnboundedReceiver, JoinHandle<()>) { + let (input, incoming) = mpsc::unbounded_channel(); + let task = tokio::spawn(std::future::pending()); + host.0.state.lock().unwrap().processes.insert( + pid.into(), + Process { + info: ProcessInfo { + plugin: "notes".into(), + version: "old".into(), + method: method.into(), + params: json!({}), + parent: None, + detached: true, + start_time: 0, + state: "running".into(), + }, + memory: sausage_vfs::Memory::default(), + files: sausage_vfs::Process::default(), + store_ended: watch::channel(true).0, + input: Some(input), + abort: Some(task.abort_handle()), + caller: "console".into(), + id: None, + answer: None, + ended: watch::channel(false).0, + }, + ); + (incoming, task) +} + +/// PROTOCOL 3.7/3.6: five seconds running, then five seconds stopping; discovery is untouched. +#[tokio::test] +async fn two_five_second_windows_and_discovery_exclusion() { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().join("world"); + World::create(&root).unwrap(); + let host = Host::test_world(&root, Runtime::new(&dir.path().join("cache")).unwrap()) + .await + .unwrap(); + let (mut input, work) = process(&host, "old-work", "work"); + let (mut discovery_input, discovery) = process(&host, "old-discovery", "rpc.discover"); + let upgrade = host.snapshot_upgrade("notes", "old".into(), "new"); + assert_eq!(upgrade.processes.len(), 1); + tokio::time::pause(); + let start = Instant::now(); + host.begin_upgrade(upgrade); + assert_eq!( + input.recv().await.unwrap(), + json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":"new"}}) + ); + // Run the upgrade dispatcher, then its per-process task, so the deadline is armed. + tokio::task::yield_now().await; + tokio::task::yield_now().await; + let window = Duration::from_secs(5); + assert_eq!(UPGRADE_GRACE, window); + // tokio::time::sleep documents millisecond granularity; the paused driver + // may round a deadline up by one tick. Check each window independently. + let tick = Duration::from_millis(1); + let before_deadline = window - tick; + advance(before_deadline).await; + tokio::task::yield_now().await; + assert_eq!(host.processes()["old-work"].state, "running"); + assert!(!work.is_finished()); + advance(tick).await; + assert!( + input.recv().await.is_none(), + "EOF at grace, even with active work" + ); + let stopped = Instant::now(); + assert!((window..=window + tick).contains(&(stopped - start))); + assert_eq!(host.processes()["old-work"].state, "stopping"); + // begin_stop schedules the independent PROTOCOL 3.6 deadline. + tokio::task::yield_now().await; + advance(before_deadline).await; + tokio::task::yield_now().await; + assert!( + !work.is_finished(), + "on_stop still has its full five seconds" + ); + advance(tick).await; + assert!(work.await.unwrap_err().is_cancelled()); + assert!((window..=window + tick).contains(&(Instant::now() - stopped))); + assert_eq!(host.processes()["old-discovery"].state, "running"); + assert!(matches!( + discovery_input.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + assert!(!discovery.is_finished()); + discovery.abort(); + let _ = discovery.await; + // Synthetic processes have no execution cleanup task; discard only this test table. + host.0.shutdown.store(true, Ordering::SeqCst); + host.0.state.lock().unwrap().processes.clear(); + host.stop().await; +} diff --git a/cli/tests/host/activation.rs b/cli/tests/host/activation.rs index dd20dab..5f5c692 100644 --- a/cli/tests/host/activation.rs +++ b/cli/tests/host/activation.rs @@ -160,9 +160,9 @@ impl Drop for TestWorld { let _ = std::fs::remove_dir_all(&self.0); } } -/// Builds a method-only document without an upgrade delay. +/// Builds a method-only document; all plugins use the host's five-second grace. fn document(name: &str) -> Value { - json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}],"x-sausage-grace":0}) + json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}]}) } /// Adds an import of a provider's method list. fn consumer(provider: &str, expected: Value) -> Value { @@ -484,7 +484,7 @@ async fn review_dropped_activation_completes() { host.stop().await; } /// PROTOCOL 10.1: discovery fails after its fixed ten-second deadline and receives EOF. -/// PROTOCOL 3.7: old processes retain their declared grace while new calls use new code. +/// PROTOCOL 3.7: old processes get five seconds regardless of declarations; new calls use new code. /// PROTOCOL 3.7: retirement waits for all live admissions, including reverted versions. /// PROTOCOL 3.7: rejected discovery candidates never receive retirement hooks. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -604,7 +604,7 @@ async fn source_changes_do_not_change_candidate() { ); host.stop().await; } -/// PROTOCOL 3.7: plugins without a declared grace period retain old calls before the default deadline. +/// PROTOCOL 3.7: missing metadata also retains old calls until the uniform deadline. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn default_grace_retains_old_process() { let world = TestWorld::new(); @@ -623,6 +623,104 @@ async fn default_grace_retains_old_process() { assert_eq!(host.processes()[&pid].state, "running"); host.stop().await; } +/// PROTOCOL 3.7: metadata cannot shorten or extend the actual grace for active requests. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn uniform_grace_ignores_discovery_declarations() { + let world = TestWorld::new(); + let declarations = [ + ("missing", None), + ("default", Some(json!(60))), + ("long", Some(json!(3600))), + ("zero", Some(json!(0))), + ("negative", Some(json!(-1))), + ("invalid", Some(json!("long"))), + ("huge", Some(json!(1e300))), + ]; + for (name, grace) in &declarations { + let mut old = document("work"); + if let Some(grace) = grace { + old["x-sausage-grace"] = grace.clone(); + } + world.plugin(name, old, document("work")); + } + let host = world.start().await; + let mut ready = host.subscribe(); + let mut calls = Vec::new(); + for (name, _) in &declarations { + let host = host.clone(); + let method = format!("{name}.work"); + calls.push(tokio::spawn(async move { + let result = host.call(&method, json!({})).await; + (result, std::time::Instant::now()) + })); + } + // Guest notifications are events, not process/file polling. + for _ in &declarations { + let event = tokio::time::timeout(Duration::from_secs(30), ready.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(event["params"]["name"], "ready"); + } + let before = std::time::Instant::now(); + let plugins: Vec<_> = declarations + .iter() + .map(|(name, _)| json!({"plugin":name,"wasm":format!("/state/{name}/next.wasm")})) + .collect(); + call(&host, "core.activate", json!({"plugins":plugins})).await; + // Use the existing discovery+stop test budget, not a new execution limit. + tokio::time::timeout(Duration::from_secs(15), async { + for task in calls { + let (result, ended) = task.await.unwrap(); + assert_eq!(result.unwrap_err()["code"], -32002); + assert!( + ended.duration_since(before) >= Duration::from_secs(5), + "every declaration gets the full grace" + ); + } + }) + .await + .unwrap(); + host.stop().await; +} + +/// PROTOCOL 10.4/3.7: a watcher refresh uses the same policy as explicit activation. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn manual_copy_uses_uniform_grace() { + let world = TestWorld::new(); + let mut old = document("work"); + old["x-sausage-grace"] = json!(3600); + world.plugin("notes", old, document("work")); + let host = world.start().await; + // Wait for initial discovery without changing the installed version. + call( + &host, + "core.activate", + json!({"plugin":"notes","wasm":"/plugins/notes/plugin.wasm"}), + ) + .await; + let mut ready = host.subscribe(); + let work_host = host.clone(); + let work = tokio::spawn(async move { work_host.call("notes.work", json!({})).await }); + let event = tokio::time::timeout(Duration::from_secs(30), ready.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(event["params"]["name"], "ready"); + let before = std::time::Instant::now(); + std::fs::copy(fixture(2), world.0.join("plugins/notes/plugin.wasm")).unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(15), work) + .await + .unwrap() + .unwrap() + .unwrap_err()["code"], + -32002 + ); + assert!(before.elapsed() >= Duration::from_secs(5)); + host.stop().await; +} + /// PROTOCOL 10.3: candidate imports must fit installed providers. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn checks_candidate_imports() { @@ -818,11 +916,11 @@ async fn graph_missing_import() { assert!(dot.as_str().unwrap().contains("absent")); host.stop().await; } -/// Checks the declared upgrade grace and dispatch to the new version. +/// Checks that an explicit long declaration cannot extend grace; new calls use the new version. async fn old_process_grace_and_new_calls() { let world = TestWorld::new(); let mut old = document("add"); - old["x-sausage-grace"] = json!(0.5); + old["x-sausage-grace"] = json!(3600); world.plugin("notes", old, document("add")); let host = world.start().await; call(&host, "notes.hold", json!({})).await; @@ -868,7 +966,7 @@ async fn superseded_notifies_each_old_process() { json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":fixture_hash(2)}}) .to_string(); for pid in pids { - // marker's existing 30-second bound is shorter than the old 60-second grace. + // The notice arrives while the old process is still in its five-second window. marker( world.0.join(format!("state/notes/superseded-{pid}")), &expected, @@ -993,7 +1091,7 @@ async fn upgrades_enter_the_journal() { .as_array() .unwrap() .iter() - .any(|e| e["line"] == "upgrade grace 60 seconds") + .any(|e| e["line"] == "upgrade grace 5 seconds") { break entries; } diff --git a/cli/tests/host/sdk.rs b/cli/tests/host/sdk.rs index dae662c..f405086 100644 --- a/cli/tests/host/sdk.rs +++ b/cli/tests/host/sdk.rs @@ -178,6 +178,56 @@ async fn stop_runs_hook() { assert_eq!(std::fs::read(world.0.join("stopped")).unwrap(), b"yes"); host.stop().await; } +/// PROTOCOL 3.7: an idle SDK listener opts into drain and exits before EOF/grace. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn superseded_idle_listener_drains_before_grace() { + let world = TestWorld::new(); + let host = world.start().await; + let (notice, mut notices) = tokio::sync::mpsc::unbounded_channel(); + host.register_native( + "other", + std::sync::Arc::new(move |_, method, _| { + let notice = notice.clone(); + Box::pin(async move { + assert_eq!(method, "superseded"); + notice.send(std::time::Instant::now()).unwrap(); + Ok(json!({})) + }) + }), + ); + let pid = call(&host, "probe.drain_service", json!({})).await; + // Replace with a different real component; force only skips unrelated test consumers. + std::fs::copy( + fixtures().join("notes.wasm"), + world.0.join("state/next.wasm"), + ) + .unwrap(); + call( + &host, + "core.activate", + json!({"plugin":"probe","wasm":"/state/next.wasm","force":true}), + ) + .await; + // Activation preparation/discovery is outside upgrade grace. Time from the + // hook's outgoing call, not from the start of candidate compilation. + let before = tokio::time::timeout(Duration::from_secs(60), notices.recv()) + .await + .unwrap() + .unwrap(); + // Waiting on its lifetime lock is an exit event, not process-liveness polling. + call(&host, "core.lock", json!({"path":"/drain-owner"})).await; + assert!(before.elapsed() < Duration::from_secs(5)); + assert!(!host.processes().contains_key(pid.as_str().unwrap())); + assert!( + !world + .0 + .join(format!("stopped-{}", pid.as_str().unwrap())) + .exists(), + "drain must not run on_stop (discovery processes may run their own hooks)" + ); + host.stop().await; +} + /// SDK framing carries payloads above 64 KiB in both directions. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn large_messages() { diff --git a/cli/tests/plugins/activation/src/main.rs b/cli/tests/plugins/activation/src/main.rs index 60c0c05..2059faf 100644 --- a/cli/tests/plugins/activation/src/main.rs +++ b/cli/tests/plugins/activation/src/main.rs @@ -84,6 +84,14 @@ fn main() { .unwrap(); json!({"retired":version}) } + "work" => { + // Keep an actual request unanswered through grace; notices do not drain it. + send( + json!({"jsonrpc":"2.0","method":"core.notify","params":{"name":"ready","params":{}}}), + ); + for _ in lines {} + return; + } "serve" | "hold" => { send(json!({"jsonrpc":"2.0","id":request["id"],"result":{"version":version}})); if request["method"] == "serve" { diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index d8865d9..582c79a 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -356,7 +356,9 @@ //! //! The interface check covers calls, not state. Old and new processes share files in `/state`. //! `core.revert` restores code, not state. Preserve compatibility or migrate the state. -//! `#[rpc(grace_seconds = N)]` declares the grace period for old processes. The default is 60 seconds. +//! Core gives every old process 5 seconds of upgrade grace, then closes stdin and allows +//! a separate 5-second stop budget before killing it. Discovery-only processes are excluded. +//! `#[rpc(grace_seconds = N)]` still emits legacy `x-sausage-grace` metadata, but core ignores it. //! New calls use the new version immediately. Old processes can still use old imports during their grace period. //! [`macro@on_superseded`] handles the notice at grace start, with normal calls still available. //! It is opt-in: without a hook the process continues as before until core stops it. diff --git a/sdk/tests/plugins/probe/src/main.rs b/sdk/tests/plugins/probe/src/main.rs index a4b013b..3de1c20 100644 --- a/sdk/tests/plugins/probe/src/main.rs +++ b/sdk/tests/plugins/probe/src/main.rs @@ -60,6 +60,11 @@ impl Probe { sausage_plugin::listen("/state/control", std::rc::Rc::new(Control)).await?; Ok(sausage_plugin::pid()) } + /// Keep an idle listener and ownership lock until opt-in drain ends this process. + async fn drain_service(&self) -> Result { + lock("/drain-owner").await?.keep(); + self.service().await + } /// Add two numbers. /// /// The result is their sum. @@ -170,6 +175,7 @@ impl Probe { #[sausage_plugin::on_stop] async fn stop(&self) { std::fs::write("/stopped", b"yes").unwrap(); + std::fs::write(format!("/stopped-{}", sausage_plugin::pid()), b"yes").unwrap(); Ctx::default().notify("stopped", json!({})).unwrap(); assert!(call("other.no", json!({})).await.is_err()); } From ff6fc62f05be144ac62377d00e99e9e79c5b9253 Mon Sep 17 00:00:00 2001 From: SAU-359 implementer Date: Fri, 30 Oct 2026 20:16:20 +0000 Subject: [PATCH 2/4] Add armed plugin-originated file watches across VFS mutations Register core.watch before its armed notification, coalesce later changes without losing post-notification writes, and clean up on caller death. Observe mounted layers and open file handles, including descriptor/stream writes, truncation, both rename paths and deletion. Track cancelled mutations through process idle and retain visible paths across renames. Document the protocol, expose native core discovery, and exercise watches through SDK callbacks and raw WASI filesystem operations. Card: SAU-359 --- PROTOCOL.md | 26 ++ cli/src/core.json | 19 ++ cli/src/core.rs | 2 + cli/src/filesystem.rs | 35 ++- cli/src/host.rs | 8 + cli/src/lib.rs | 1 + cli/src/routing.rs | 19 +- cli/src/watches.rs | 178 ++++++++++++++ cli/tests/host/main.rs | 1 + cli/tests/host/watches.rs | 195 +++++++++++++++ cli/tests/plugins/probe/src/main.rs | 66 ++++++ sdk/src/calls.rs | 5 + sdk/tests/plugins/probe/src/main.rs | 33 +++ vfs/src/lib.rs | 1 + vfs/src/observe.rs | 355 ++++++++++++++++++++++++++++ vfs/src/process.rs | 17 +- 16 files changed, 946 insertions(+), 15 deletions(-) create mode 100644 cli/src/core.json create mode 100644 cli/src/watches.rs create mode 100644 cli/tests/host/watches.rs create mode 100644 vfs/src/observe.rs diff --git a/PROTOCOL.md b/PROTOCOL.md index e9e6b6b..bd5635c 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -219,6 +219,7 @@ these areas fails. /proc/ core/ PROTOCOL.md this file + discover.json core.rpc.discover document graph.json plugins, imports, and the result of each check graph.dot the same graph for Graphviz locks.json held locks and waiting processes @@ -246,6 +247,29 @@ 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. The watch ends with its caller; there +is no cancellation API or polling/retry loop. + ## 5. Endpoints An endpoint is a JSON-RPC interface that a running process serves over TCP, @@ -333,7 +357,9 @@ At start, the core starts each service in `/config/boot.yaml`, in order: | Method | Params | Result | | --- | --- | --- | +| `core.rpc.discover` | none | OpenRPC core API document (also `/proc/core/discover.json`) | | `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.json b/cli/src/core.json new file mode 100644 index 0000000..6e05c04 --- /dev/null +++ b/cli/src/core.json @@ -0,0 +1,19 @@ +{ + "openrpc": "1.2.6", + "info": {"title":"core", "version":"0.1.0"}, + "methods": [ + {"name": "rpc.discover", "summary": "Describe the native core API.", "description": "Describe the native core API.", "paramStructure": "by-name", "params": [], "result": {"name": "result", "schema": {"type": "object"}}}, + {"name": "watch", "summary": "Keep this call open until the caller process ends. First core.notify is armed, then changed after successful plugin-originated file mutations. Scan only after armed; rescan after changed. Events may coalesce; files are the truth. No polling or cancellation API. See PROTOCOL section 4.", "description": "Keep this call open until the caller process ends. First core.notify is armed, then changed after successful plugin-originated file mutations. Scan only after armed; rescan after changed. Events may coalesce; files are the truth. No polling or cancellation API. See PROTOCOL section 4.", "paramStructure": "by-name", "params": [{"name": "path", "required": true, "description": "Virtual file or directory path, including descendants; need not exist. /self is relative to the caller.", "schema": {"type": "string"}}], "result": {"name": "result", "schema": {"type": "null"}}}, + {"name": "call", "summary": "Make a call and return its result; detach lets the new process outlive its caller.", "description": "Make a call and return its result; detach lets the new process outlive its caller.", "paramStructure": "by-name", "params": [{"name": "method", "required": true, "description": "Qualified method.", "schema": {"type": "string"}}, {"name": "params", "required": true, "description": "Named parameters.", "schema": {"type": "object"}}, {"name": "detach", "required": false, "description": "Detach the new process.", "schema": {"type": "boolean"}}], "result": {"name": "result", "schema": {}}}, + {"name": "notify", "summary": "Notify this process's caller while its call is open.", "description": "Notify this process's caller while its call is open.", "paramStructure": "by-name", "params": [{"name": "name", "required": true, "description": "Event name.", "schema": {"type": "string"}}, {"name": "params", "required": false, "description": "Event payload.", "schema": {}}], "result": {"name": "result", "schema": {"type": "null"}}}, + {"name": "stop", "summary": "End a process and await cleanup.", "description": "End a process and await cleanup.", "paramStructure": "by-name", "params": [{"name": "pid", "required": true, "description": "Process ID.", "schema": {"type": "string"}}], "result": {"name": "result", "schema": {"type": "object"}}}, + {"name": "kill", "summary": "End a process and await cleanup.", "description": "End a process and await cleanup.", "paramStructure": "by-name", "params": [{"name": "pid", "required": true, "description": "Process ID.", "schema": {"type": "string"}}], "result": {"name": "result", "schema": {"type": "object"}}}, + {"name": "lock", "summary": "Acquire a FIFO process-lifetime lock.", "description": "Acquire a FIFO process-lifetime lock.", "paramStructure": "by-name", "params": [{"name": "path", "required": true, "description": "Lock key.", "schema": {"type": "string"}}, {"name": "wait", "required": false, "description": "Wait for acquisition (default true).", "schema": {"type": "boolean"}}], "result": {"name": "result", "schema": {"type": "object", "properties": {"held": {"type": "boolean"}}, "required": ["held"]}}}, + {"name": "unlock", "summary": "Release an owned lock.", "description": "Release an owned lock.", "paramStructure": "by-name", "params": [{"name": "path", "required": true, "description": "Lock key.", "schema": {"type": "string"}}], "result": {"name": "result", "schema": {"type": "null"}}}, + {"name": "activate", "summary": "Install one or more checked plugin versions. See PROTOCOL section 10.", "description": "Install one or more checked plugin versions. See PROTOCOL section 10.", "paramStructure": "by-name", "params": [{"name": "plugin", "required": false, "description": "Single plugin name.", "schema": {"type": "string"}}, {"name": "wasm", "required": false, "description": "Candidate path.", "schema": {"type": "string"}}, {"name": "plugins", "required": false, "description": "Batch candidates.", "schema": {"type": "array", "items": {"type": "object", "properties": {"plugin": {"type": "string"}, "wasm": {"type": "string"}}, "required": ["plugin", "wasm"]}}}, {"name": "force", "required": false, "description": "Skip compatibility rejection.", "schema": {"type": "boolean"}}], "result": {"name": "result", "schema": {"type": "object"}}}, + {"name": "revert", "summary": "Restore the previous plugin version.", "description": "Restore the previous plugin version.", "paramStructure": "by-name", "params": [{"name": "plugin", "required": true, "description": "Plugin name.", "schema": {"type": "string"}}, {"name": "force", "required": false, "description": "Skip compatibility rejection.", "schema": {"type": "boolean"}}], "result": {"name": "result", "schema": {"type": "object"}}}, + {"name": "log", "summary": "Read the central journal (PROTOCOL section 7).", "description": "Read the central journal (PROTOCOL section 7).", "paramStructure": "by-name", "params": [{"name": "pid", "required": false, "description": "Process ID.", "schema": {"type": "string"}}, {"name": "plugin", "required": false, "description": "Plugin name.", "schema": {"type": "string"}}, {"name": "since", "required": false, "description": "ISO 8601 start time.", "schema": {"type": "string"}}, {"name": "until", "required": false, "description": "ISO 8601 end time.", "schema": {"type": "string"}}, {"name": "grep", "required": false, "description": "Line substring.", "schema": {"type": "string"}}, {"name": "limit", "required": false, "description": "Maximum entries (default 100).", "schema": {"type": "integer"}}], "result": {"name": "result", "schema": {"type": "array", "items": {"type": "object"}}}}, + {"name": "restart", "summary": "Request clean host restart through the loop supervisor; needs explicit operator approval.", "description": "Request clean host restart through the loop supervisor; needs explicit operator approval.", "paramStructure": "by-name", "params": [], "result": {"name": "result", "schema": {"type": "object"}}}, + {"name": "exit", "summary": "Request clean host restart through the loop supervisor; needs explicit operator approval.", "description": "Request clean host restart through the loop supervisor; needs explicit operator approval.", "paramStructure": "by-name", "params": [], "result": {"name": "result", "schema": {"type": "object"}}} + ] +} diff --git a/cli/src/core.rs b/cli/src/core.rs index 7bc863f..12c32fa 100644 --- a/cli/src/core.rs +++ b/cli/src/core.rs @@ -12,6 +12,8 @@ impl Host { p: Value, ) -> RpcResult { match method { + "rpc.discover" => Ok(serde_json::from_str(include_str!("core.json")).unwrap()), + "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..a74848f 100644 --- a/cli/src/filesystem.rs +++ b/cli/src/filesystem.rs @@ -47,27 +47,48 @@ 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 { let mut files = Generated::default(); + files.insert( + "/discover.json", + || Ok(include_bytes!("core.json").to_vec()), + ); let protocol = include_bytes!("../../PROTOCOL.md"); files.insert("/PROTOCOL.md", move || Ok(protocol.to_vec())); let graph = self.graph_json(); 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 4650c85..738e3f2 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/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..92c3a77 --- /dev/null +++ b/cli/tests/host/watches.rs @@ -0,0 +1,195 @@ +//! 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; +} + +/// PROTOCOL 9: core discovery includes the armed watch and matches its proc projection. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discovery() { + let (_dir, h) = start().await; + let doc = call(&h, "core.rpc.discover", json!({})).await; + let projected = call(&h, "probe.read", json!({"path":"/proc/core/discover.json"})).await; + assert_eq!( + doc, + serde_json::from_str::(projected.as_str().unwrap()).unwrap() + ); + let watch = doc["methods"] + .as_array() + .unwrap() + .iter() + .find(|m| m["name"] == "watch") + .unwrap(); + assert_eq!(watch["params"][0]["name"], "path"); + assert!(watch["description"].as_str().unwrap().contains("armed")); + h.stop().await; +} diff --git a/cli/tests/plugins/probe/src/main.rs b/cli/tests/plugins/probe/src/main.rs index c20bd7b..972553c 100644 --- a/cli/tests/plugins/probe/src/main.rs +++ b/cli/tests/plugins/probe/src/main.rs @@ -468,6 +468,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 +519,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..bff9005 100644 --- a/sdk/src/calls.rs +++ b/sdk/src/calls.rs @@ -26,6 +26,11 @@ impl Call { self } /// Receive progress notifications before the result arrives. + /// + /// For `core.watch`, await this call in a background task for the process's + /// lifetime. Scan only after `armed`, and rescan on `changed`; events can + /// coalesce. The call has no cancellation API. 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 a4b013b..581a946 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,33 @@ 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(()) + } + /// Publish a control endpoint until EOF. async fn serve(&self) -> Result<()> { sausage_plugin::listen("/endpoint", std::rc::Rc::new(Control)).await 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..be72b99 --- /dev/null +++ b/vfs/src/observe.rs @@ -0,0 +1,355 @@ +//! Observes successful namespace and open-file mutations, even if their caller is cancelled. +use crate::*; +use std::sync::{Mutex, Weak}; + +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>>>, + // Keep path rebasing and mutations in the same order across all namespaces. + mutations: Arc>, +} +impl Observer { + pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self { + Self { + changed: Arc::new(changed), + paths: Arc::default(), + mutations: 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, + }) + } + 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 order = observer.mutations.lock().await; + let result = f.await?; + let path = path.lock().unwrap().clone(); + drop(order); + 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.mutations.lock().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 { + let order = observer.mutations.lock().await; + layer.mkdir(&local).await?; + drop(order); + (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.mutations.lock().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.mutations.lock().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()); + } + } + /// 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.mutations.lock().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.mkdir("/published").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. From 0c1ef74f6efbdf0912bae786ba95e45408fc10f6 Mon Sep 17 00:00:00 2001 From: SAU-359 implementer Date: Fri, 30 Oct 2026 21:06:10 +0000 Subject: [PATCH 3/4] Keep file-watch observation from serializing unrelated IO Scope open-registration and rename/remove ordering to overlapping paths; let descriptor mutations and directory creation proceed without that lock. Prove blocked opens/writes cannot stall unrelated namespaces, including a queued rename, and retain post-rename descriptor notifications. Document handler-owned watch lifetimes and narrow paths; test natural idle exit. Card: SAU-359 --- PROTOCOL.md | 11 +- sdk/src/calls.rs | 14 +- sdk/tests/plugins/probe/src/main.rs | 35 ++++ sdk/tests/wasi.rs | 9 + vfs/src/observe.rs | 286 ++++++++++++++++++++++++++-- 5 files changed, 336 insertions(+), 19 deletions(-) diff --git a/PROTOCOL.md b/PROTOCOL.md index bd5635c..9adef6d 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -267,8 +267,15 @@ 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. The watch ends with its caller; there -is no cancellation API or polling/retry loop. +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 diff --git a/sdk/src/calls.rs b/sdk/src/calls.rs index bff9005..04e2c6e 100644 --- a/sdk/src/calls.rs +++ b/sdk/src/calls.rs @@ -27,10 +27,16 @@ impl Call { } /// Receive progress notifications before the result arrives. /// - /// For `core.watch`, await this call in a background task for the process's - /// lifetime. Scan only after `armed`, and rescan on `changed`; events can - /// coalesce. The call has no cancellation API. Do not use endpoint connect - /// retries to wait for file publication. + /// 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 581a946..ba8caa4 100644 --- a/sdk/tests/plugins/probe/src/main.rs +++ b/sdk/tests/plugins/probe/src/main.rs @@ -63,6 +63,41 @@ impl Probe { 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/observe.rs b/vfs/src/observe.rs index be72b99..e7f67d0 100644 --- a/vfs/src/observe.rs +++ b/vfs/src/observe.rs @@ -1,6 +1,16 @@ //! Observes successful namespace and open-file mutations, even if their caller is cancelled. use crate::*; -use std::sync::{Mutex, Weak}; +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; @@ -11,15 +21,16 @@ type Callback = dyn Fn(&str) + Send + Sync; pub struct Observer { changed: Arc, paths: Arc>>>, - // Keep path rebasing and mutations in the same order across all namespaces. - mutations: 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(), - mutations: Arc::default(), + namespace: Arc::default(), } } /// Wrap a layer at its canonical namespace prefix (e.g. `/run`). @@ -31,6 +42,56 @@ impl Observer { 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(); @@ -85,10 +146,8 @@ impl Handle { let (path, observer) = (self.path.clone(), self.observer.clone()); self.process .operation(async move { - let order = observer.mutations.lock().await; let result = f.await?; let path = path.lock().unwrap().clone(); - drop(order); if let Some(path) = path { (observer.changed)(&path); } @@ -118,7 +177,7 @@ impl Layer for Observed { ); self.process .operation(async move { - let order = observer.mutations.lock().await; + let order = observer.order(&[(&path, false)]).await; let file = layer.open(&local, options).await?; let handle = Arc::new(Handle { file, @@ -143,9 +202,7 @@ impl Layer for Observed { ); self.process .operation(async move { - let order = observer.mutations.lock().await; layer.mkdir(&local).await?; - drop(order); (observer.changed)(&path); Ok(()) }) @@ -168,7 +225,7 @@ impl Layer for Observed { ); self.process .operation(async move { - let order = observer.mutations.lock().await; + let order = observer.order(&[(&from, true), (&to, true)]).await; layer.rename(&a, &b).await?; if from != to { observer.moved(&from, Some(&to)); @@ -191,7 +248,7 @@ impl Observed { ); self.process .operation(async move { - let order = observer.mutations.lock().await; + let order = observer.order(&[(&path, true)]).await; if directory { layer.remove_dir(&local).await?; } else { @@ -324,6 +381,200 @@ mod tests { 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] @@ -334,14 +585,23 @@ mod tests { }); let process = Process::default(); let layer = observer.layer(Arc::new(Memory::default()), "/", process.clone()); - let order = observer.mutations.lock().await; + 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.mkdir("/published").await + layer + .open( + "/published", + Open { + create: true, + write: true, + ..Open::default() + }, + ) + .await }) .await }); From 95b27f094895237d14f9c23b785e1afa086ed325 Mon Sep 17 00:00:00 2001 From: SAU-359 implementer Date: Fri, 30 Oct 2026 21:17:28 +0000 Subject: [PATCH 4/4] Remove fixture races exposed by concurrent file IO Read only atomically published journal batches, not in-progress spool tempfiles. Wait for a guest-ready notification before closing stdin in the EOF test, rather than treating process admission as handler readiness. Card: SAU-359 --- cli/tests/host/core.rs | 19 +++++++++++-------- cli/tests/plugins/probe/src/main.rs | 3 +++ 2 files changed, 14 insertions(+), 8 deletions(-) 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/plugins/probe/src/main.rs b/cli/tests/plugins/probe/src/main.rs index 972553c..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(); }