surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge 'b696e57dc5ff6393c39daee0c639bf66ce9da06e'

Browse Source
This commit is contained in:
surma
2026-10-31 02:41:40 +00:00
parent 0d2d633447 b696e57dc5
commit 205e5d7775
27 changed files with 1588 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
PROTOCOL.md
+40 -3
View File
@@ -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
@@ -219,6 +222,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 +250,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/<pid>` 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,
@@ -352,7 +386,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` | `{}` |
@@ -410,7 +446,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
cli/src/activation.rs
-5
View File
@@ -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(
cli/src/core.json
+19
View File
@@ -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"}}}
]
}
cli/src/core.rs
+2
View File
@@ -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(),
cli/src/filesystem.rs
+28 -7
View File
@@ -47,27 +47,48 @@ impl Host {
tracker: Process,
) -> anyhow::Result<Mounts> {
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();
cli/src/graph.rs
+1 -10
View File
@@ -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(),
&current.hash,
grace,
));
upgrades.push(self.snapshot_upgrade(plugin, old.hash.clone(), &current.hash));
}
}
self.publish_graph(plugins);
cli/src/host.rs
+8
View File
@@ -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<crate::graph::Graph>,
pub(crate) activations: tokio::sync::Mutex<()>,
pub(crate) watcher: Mutex<Option<notify::RecommendedWatcher>>,
@@ -91,6 +93,10 @@ impl Host {
))
}
fn with_parts(world: World, runtime: Arc<Runtime>, 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),
cli/src/lib.rs
+1
View File
@@ -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};
cli/src/routing.rs
+12 -7
View File
@@ -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,
cli/src/services/tests.rs
+7 -13
View File
@@ -164,7 +164,7 @@ async fn fifth_restarts_sixth_gives_up_and_journals_once() {
rig.no_start();
// Neither time passing nor an upgrade re-arms a given-up service.
tokio::time::advance(CRASH_WINDOW).await;
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new", 0.0);
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
rig.host.begin_upgrade(upgrade);
tokio::task::yield_now().await;
rig.no_start();
@@ -228,7 +228,7 @@ async fn services_have_independent_crash_budgets() {
async fn upgrade_crash_race_has_one_restart_and_retired_fallback_cannot_duplicate_it() {
let mut rig = Rig::new().await;
let (pid, end) = rig.start("demo.serve").await;
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new", 0.0);
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
rig.crash(&pid).await; // Exits between snapshot and begin_upgrade.
drop(end);
rig.host.begin_upgrade(upgrade);
@@ -271,9 +271,7 @@ async fn crash_before_upgrade_snapshot_also_has_one_restart() {
rig.crash(&pid).await;
drop(end);
// Restart may be pending admission or already admitted: snapshot handles both.
let upgrade = rig
.host
.snapshot_upgrade("demo", "old".into(), "native", 0.0);
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native");
rig.host.begin_upgrade(upgrade);
let (pid, end) = rig.next().await;
rig.assert_single(&pid, 1);
@@ -290,9 +288,7 @@ async fn pending_admission_is_single_owned_across_upgrade() {
let admission = host.0.admission.write().await;
rig.crash(&pid).await;
drop(end);
let upgrade = rig
.host
.snapshot_upgrade("demo", "old".into(), "native", 0.0);
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native");
rig.host.begin_upgrade(upgrade);
assert!(rig.host.0.state.lock().unwrap().services[0].pending);
drop(admission);
@@ -324,7 +320,7 @@ async fn clean_service_stays_dormant_until_upgrade_and_explicit_stop_stays_disab
let (pid, end) = rig.start("demo.serve").await;
rig.clean(&pid, end).await;
rig.no_start();
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new", 0.0);
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
rig.host.begin_upgrade(upgrade);
let (pid, end) = rig.next().await;
rig.assert_single(&pid, 1);
@@ -332,9 +328,7 @@ async fn clean_service_stays_dormant_until_upgrade_and_explicit_stop_stays_disab
let mut ended = rig.host.begin_stop(&pid).unwrap();
end.send(()).unwrap();
ended.wait_for(|done| *done).await.unwrap();
let upgrade = rig
.host
.snapshot_upgrade("demo", "old".into(), "newer", 0.0);
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "newer");
rig.host.begin_upgrade(upgrade);
tokio::task::yield_now().await;
rig.no_start();
@@ -345,7 +339,7 @@ async fn clean_service_stays_dormant_until_upgrade_and_explicit_stop_stays_disab
async fn intentional_stop_and_shutdown_never_restart_even_during_upgrade() {
let mut rig = Rig::new().await;
let (pid, end) = rig.start("demo.serve").await;
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new", 0.0);
let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
// An explicit kill after supersession overrides the automatic upgrade restart.
rig.host
.call("core.kill", json!({"pid":pid}))
cli/src/upgrades.rs
+8 -6
View File
@@ -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<OldProcess>,
}
impl Host {
@@ -23,7 +28,6 @@ impl Host {
plugin: &str,
previous: String,
current: &str,
grace: f64,
) -> Upgrade {
let mut state = self.0.state.lock().unwrap();
let processes = state
@@ -65,7 +69,6 @@ impl Host {
plugin: plugin.into(),
previous,
current: current.into(),
grace: Duration::from_secs_f64(grace),
processes,
}
}
@@ -87,7 +90,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<String, Vec<watch::Receiver<bool>>> = BTreeMap::new();
versions.entry(upgrade.previous).or_default();
@@ -97,12 +100,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_retire(&process.pid); }
_ = tokio::time::sleep(UPGRADE_GRACE) => { host.begin_retire(&process.pid); }
}
});
}
cli/src/upgrades/tests.rs
+115
View File
@@ -0,0 +1,115 @@
//! 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<serde_json::Value>, 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,
stopped: false,
superseded: false,
},
);
(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_retire 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;
}
cli/src/watches.rs
+178
View File
@@ -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<Notify>,
}
pub(crate) type Watches = Arc<Mutex<HashMap<Uuid, Watch>>>;
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<Value>, 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;
}
}
cli/tests/host/activation.rs
+106 -8
View File
@@ -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;
}
cli/tests/host/core.rs
+11 -8
View File
@@ -208,7 +208,11 @@ async fn journal_recovers_after_sqlite_write_failure() {
call(&h, "core.log", json!({})).await;
let batches: Vec<Value> = 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
cli/tests/host/main.rs
+1
View File
@@ -7,3 +7,4 @@ mod common;
mod console;
mod core;
mod sdk;
mod watches;
cli/tests/host/sdk.rs
+50
View File
@@ -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() {
cli/tests/host/watches.rs
+195
View File
@@ -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<Value>, 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<sausage::RpcResult> {
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/<pid>.
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::<Value>(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;
}
cli/tests/plugins/activation/src/main.rs
+8
View File
@@ -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" => {
// One genuine process failure, followed by a healthy service, without
// timers or liveness polling in either the fixture or supervisor.
cli/tests/plugins/probe/src/main.rs
+69
View File
@@ -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<Value> {
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!({})
}
sdk/src/calls.rs
+11
View File
@@ -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
sdk/src/lib.rs
+3 -1
View File
@@ -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.
sdk/tests/plugins/probe/src/main.rs
+74
View File
@@ -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::<Waker>)));
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
@@ -60,6 +128,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<String> {
lock("/drain-owner").await?.keep();
self.service().await
}
/// Add two numbers.
///
/// The result is their sum.
@@ -170,6 +243,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());
}
sdk/tests/wasi.rs
+9
View File
@@ -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();
vfs/src/lib.rs
+1
View File
@@ -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;
vfs/src/observe.rs
+615
View File
@@ -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<BTreeMap<String, Weak<RwLock<()>>>>;
enum Held {
Read { _guard: OwnedRwLockReadGuard<()> },
Write { _guard: OwnedRwLockWriteGuard<()> },
}
type TrackedPath = Mutex<Option<String>>;
type Path = Arc<TrackedPath>;
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<Callback>,
paths: Arc<Mutex<Vec<Weak<TrackedPath>>>>,
// 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<Namespace>,
}
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<dyn Layer>, prefix: &str, process: Process) -> Arc<dyn Layer> {
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<Held> {
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<dyn Layer>,
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<dyn File>,
path: Path,
observer: Observer,
process: Process,
}
impl Handle {
async fn mutate<T: Send + 'static>(
&self,
f: impl Future<Output = Result<T>> + Send + 'static,
) -> Result<T> {
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<Stat> {
self.layer.stat(path, follow).await
}
async fn readlink(&self, path: &str) -> Result<String> {
self.layer.readlink(path).await
}
async fn list(&self, path: &str) -> Result<Vec<Entry>> {
self.layer.list(path).await
}
async fn open(&self, path: &str, options: Open) -> Result<Arc<dyn File>> {
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<dyn File>)
})
.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<Stat> {
self.file.stat().await
}
async fn read(&self, offset: u64, length: usize) -> Result<Vec<u8>> {
self.file.read(offset, length).await
}
async fn write(&self, offset: u64, bytes: Vec<u8>) -> Result<usize> {
let f = self.file.clone();
self.mutate(async move { f.write(offset, bytes).await })
.await
}
async fn append(&self, bytes: Vec<u8>) -> Result<usize> {
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<SystemTime>,
modified: Option<SystemTime>,
) -> 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<dyn Layer> = 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<T>(future: impl Future<Output = T>) -> 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<Memory>,
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<Stat> {
self.memory.stat(p, follow).await
}
async fn list(&self, p: &str) -> Result<Vec<Entry>> {
self.memory.list(p).await
}
async fn open(&self, p: &str, options: Open) -> Result<Arc<dyn File>> {
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");
}
}
vfs/src/process.rs
+16 -1
View File
@@ -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<T: Send + 'static>(
&self,
operation: impl std::future::Future<Output = std::io::Result<T>> + Send + 'static,
) -> std::io::Result<T> {
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.