diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 15c86e1..674cb0d 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -26,3 +26,9 @@ wasmtime-wasi-http = "49" uuid = { version = "1", features = ["v7"] } sha2 = "0.10" rusqlite = { version = "0.38", features = ["bundled"] } +sausage-vfs = { version = "0.1.0", path = "../vfs" } +sausage-interface = { version = "0.1.0", path = "../interface" } +notify = "8.2.0" + +[dev-dependencies] +sausage-interface = { path = "../interface" } diff --git a/cli/src/activation.rs b/cli/src/activation.rs index b99d6f1..2aca707 100644 --- a/cli/src/activation.rs +++ b/cli/src/activation.rs @@ -1,15 +1,210 @@ -//! Activation and revert of plugin versions (PROTOCOL.md section 10.3). -use crate::{Host, RpcResult, error}; -use serde_json::Value; +//! Serialized candidate-copy activation and revert under PROTOCOL.md section 10.3. +use crate::{ + Host, RpcResult, error, + graph::{Interface, io_error, problems}, +}; +use serde_json::{Value, json}; +use std::{collections::BTreeMap, path::PathBuf}; +struct Candidate { + plugin: String, + copy: PathBuf, + interface: Interface, +} impl Host { - /// `core.activate` and `core.revert`. - pub(crate) async fn activation( + /// Copies first, discovers that copy, checks the combined graph, then installs it. + pub(crate) async fn activation(&self, _caller: &str, method: &str, params: Value) -> RpcResult { + let _activation = self.0.activations.lock().await; + let mut candidates = Vec::new(); + let result = self + .activate_candidates(method, ¶ms, &mut candidates) + .await; + for candidate in candidates { + let _ = tokio::fs::remove_file(candidate.copy).await; + } + result + } + /// Prepares every member before graph checks or changes to installed files. + async fn activate_candidates( &self, - _caller: &str, - _method: &str, - _params: Value, + method: &str, + params: &Value, + candidates: &mut Vec, ) -> RpcResult { - Err(error(-32601, "method not found")) + let requests = params + .get("plugins") + .and_then(Value::as_array) + .cloned() + .unwrap_or_else(|| vec![params.clone()]); + if requests.is_empty() { + return Err(error(-32602, "plugins must not be empty")); + } + for request in requests { + let plugin = request["plugin"] + .as_str() + .ok_or_else(|| error(-32602, "plugin must be a string"))?; + if plugin.is_empty() + || !plugin.bytes().all(|byte| { + byte.is_ascii_lowercase() + || byte.is_ascii_digit() + || byte == b'_' + || byte == b'-' + }) + || candidates + .iter() + .any(|candidate| candidate.plugin == plugin) + { + return Err(error(-32602, "invalid or duplicate plugin")); + } + let directory = self.0.world.plugin_dir(plugin); + tokio::fs::create_dir_all(&directory) + .await + .map_err(io_error)?; + let source = if method == "revert" { + directory.join("plugin.prev.wasm") + } else { + let source = PathBuf::from( + request["wasm"] + .as_str() + .ok_or_else(|| error(-32602, "wasm must be a string"))?, + ); + if source.starts_with(&self.0.world.root) { + source + } else { + self.0 + .world + .root + .join(source.strip_prefix("/").unwrap_or(&source)) + } + }; + let copy = directory.join(format!("activate-{}.wasm", uuid::Uuid::now_v7())); + // Read once. Discovery and installation both use the same private bytes. + let bytes = tokio::fs::read(source).await.map_err(io_error)?; + tokio::fs::write(©, &bytes).await.map_err(io_error)?; + self.0 + .journal + .write("core", plugin, "core", "activation copied candidate"); + let discovery = self.discover_copy(plugin, ©, &bytes).await; + let interface = match discovery { + Ok(interface) => interface, + Err(error) => { + let _ = tokio::fs::remove_file(copy).await; + return Err(error); + } + }; + self.0 + .journal + .write("core", plugin, "core", "activation discovered candidate"); + candidates.push(Candidate { + plugin: plugin.into(), + copy, + interface, + }); + } + self.refresh_graph().await?; + let previous = self.0.graph.lock().unwrap().plugins.clone(); + if previous.values().any(|entry| entry.error.is_some()) { + return Err(error( + -32003, + "activation waits for successful discovery of every plugin", + )); + } + let mut proposed = previous.clone(); + let names: Vec<_> = candidates + .iter() + .map(|candidate| candidate.plugin.clone()) + .collect(); + for candidate in candidates.iter() { + proposed.insert(candidate.plugin.clone(), candidate.interface.clone()); + } + if params["force"] != true { + let failures = problems(&proposed, Some(&names)); + if !failures.is_empty() { + self.0.journal.write( + "core", + "core", + "core", + &format!("activation rejected {}", json!(failures)), + ); + let details = failures + .iter() + .map(|problem| { + format!( + "- {} imports {}: {}", + problem["consumer"], + problem["provider"], + problem["display"] + .as_str() + .unwrap_or_else(|| problem["message"] + .as_str() + .unwrap_or("incompatible import")) + ) + }) + .collect::>() + .join("\n"); + return Err( + json!({"code":-32003,"message":format!("{} was not activated: consumers break.\n{details}\nUpdate consumers first, or activate both together.",names.join(", ")),"data":failures}), + ); + } + } + self.install_candidates(candidates, previous, proposed) + .await + } + /// Installs staged files and snapshots all old admissions under the shared write gate. + async fn install_candidates( + &self, + candidates: &[Candidate], + previous: BTreeMap, + proposed: BTreeMap, + ) -> RpcResult { + let admission = self.0.admission.write().await; + // Every candidate already has a temporary name before any installed file changes. + for candidate in candidates { + let directory = self.0.world.plugin_dir(&candidate.plugin); + let current = directory.join("plugin.wasm"); + if tokio::fs::try_exists(¤t).await.map_err(io_error)? { + tokio::fs::rename(¤t, directory.join("plugin.prev.wasm")) + .await + .map_err(io_error)?; + } + tokio::fs::rename(&candidate.copy, ¤t) + .await + .map_err(io_error)?; + } + let mut upgrades = Vec::new(); + let mut documents = serde_json::Map::new(); + for candidate in candidates { + documents.insert( + candidate.plugin.clone(), + candidate.interface.document.clone(), + ); + if let Some(old) = previous + .get(&candidate.plugin) + .filter(|old| 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( + "core", + &candidate.plugin, + "core", + &format!("activation installed {}", candidate.interface.hash), + ); + } + self.publish_graph(proposed); + drop(admission); + for upgrade in upgrades { + self.begin_upgrade(upgrade); + } + Ok(Value::Object(documents)) } } diff --git a/cli/src/core.rs b/cli/src/core.rs index 96d0810..cc74f67 100644 --- a/cli/src/core.rs +++ b/cli/src/core.rs @@ -31,7 +31,19 @@ impl Host { Ok(Value::Null) } "kill" => { - self.kill(required_str(&p, "pid")?); + let pid = required_str(&p, "pid")?; + let ended = self + .0 + .state + .lock() + .unwrap() + .processes + .get(pid) + .map(|p| p.ended.subscribe()); + self.kill(pid); + if let Some(mut ended) = ended { + let _ = ended.wait_for(|ended| *ended).await; + } Ok(json!({})) } "stop" => { diff --git a/cli/src/execution.rs b/cli/src/execution.rs index f5af1e0..dadd8fa 100644 --- a/cli/src/execution.rs +++ b/cli/src/execution.rs @@ -1,7 +1,7 @@ //! Execution of native futures and WASI processes, including transport drainage. use crate::{ Host, NativeContext, NativePlugin, - process::{ProcessGuard, ProcessInfo}, + process::ProcessInfo, runtime::{Instance, InstanceConfig}, }; use serde_json::{Value, json}; @@ -17,11 +17,7 @@ impl Host { native: Option, incoming: mpsc::UnboundedReceiver, has_id: bool, - ) { - let _guard = ProcessGuard { - host: self.clone(), - pid: pid.clone(), - }; + ) -> String { if let Some(native) = native { let context = NativeContext { host: self.clone(), @@ -29,13 +25,12 @@ impl Host { }; let result = native(context, info.method, info.params).await; self.answer(&pid, result); - self.finish(&pid, "exit 0"); - return; + return "exit 0".into(); } let result = self .run_guest(&pid, &info, &component.unwrap(), incoming, has_id) .await; - self.finish(&pid, &result.unwrap_or_else(|error| error.to_string())); + result.unwrap_or_else(|error| error.to_string()) } /// Drain stdout before deciding whether the guest returned an answer. @@ -47,8 +42,14 @@ impl Host { mut incoming: mpsc::UnboundedReceiver, has_id: bool, ) -> anyhow::Result { + let (memory, tracker, store_ended) = { + let state = self.0.state.lock().unwrap(); + let p = state.processes.get(pid).unwrap(); + (p.memory.clone(), p.files.clone(), p.store_ended.clone()) + }; let config = InstanceConfig { - world: self.0.world.root.clone(), + filesystem: self.mounts(Some(pid.into()), memory, tracker).await?, + store_ended, network: true, env: vec![ ("SAUSAGE_PID".into(), pid.into()), diff --git a/cli/src/filesystem.rs b/cli/src/filesystem.rs new file mode 100644 index 0000000..bc70e5b --- /dev/null +++ b/cli/src/filesystem.rs @@ -0,0 +1,301 @@ +//! Per-process mount tables, reserved self files, and the live process namespace. +use crate::Host; +use sausage_vfs::{ + Entry, File, Generated, Layer, Memory, Mounts, Open, Process, Result, Stat, async_trait, +}; +use std::{io, sync::Arc}; + +fn denied() -> Result { + Err(io::ErrorKind::ReadOnlyFilesystem.into()) +} +fn missing() -> Result { + Err(io::ErrorKind::NotFound.into()) +} +fn directory() -> Stat { + Stat { + directory: true, + symlink: false, + size: 0, + accessed: None, + modified: None, + } +} +fn entry(name: String, directory: bool) -> Entry { + Entry { + name, + directory, + symlink: false, + } +} +fn reserved(path: &str) -> bool { + matches!( + path.trim_start_matches('/').split('/').next(), + Some("info.json" | "log" | "fs") + ) +} + +impl Host { + /// Create a console filesystem view without adding a process. Its `/self` is private and ephemeral. + pub async fn filesystem(&self) -> anyhow::Result { + self.mounts(None, Memory::default(), Process::default()) + .await + } + pub(crate) async fn mounts( + &self, + pid: Option, + memory: Memory, + tracker: Process, + ) -> anyhow::Result { + let root = self.0.world.root.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("/proc", Arc::new(Proc(self.clone())), true); + mounts.mount( + "/self", + Arc::new(SelfFolder { + host: self.clone(), + pid, + memory, + }), + false, + ); + mounts.mount("/run", Arc::new(self.0.run.clone()), false); + Ok(mounts) + } + fn core_files(&self) -> Generated { + let mut files = Generated::default(); + let protocol = include_bytes!("../../PROTOCOL.md"); + files.insert("/PROTOCOL.md", move || Ok(protocol.to_vec())); + let graph = self.graph_json(); + files.insert("/graph.json", move || Ok(graph.clone())); + let dot = self.graph_dot(); + files.insert("/graph.dot", move || Ok(dot.clone())); + let locks = { + let state = self.0.state.lock().unwrap(); + let locks: std::collections::BTreeMap<_, _> = state + .locks + .iter() + .map(|(path, lock)| { + let waiting: Vec<_> = lock.queue.iter().map(|(pid, _)| pid).collect(); + let info = serde_json::json!({ + "owner": lock.owner, + "waiting": waiting, + }); + (path.clone(), info) + }) + .collect(); + serde_json::to_vec(&locks).unwrap() + }; + files.insert("/locks.json", move || Ok(locks.clone())); + for name in self.discovered_plugins() { + if let Some(doc) = self.discover_document(&name) { + let bytes = serde_json::to_vec(&doc).unwrap(); + files.insert(&format!("/plugins/{name}/discover.json"), move || { + Ok(bytes.clone()) + }); + files.insert(&format!("/plugins/{name}/client.rs"), move || { + sausage_interface::client_rs(&name, &doc) + .map(String::into_bytes) + .map_err(io::Error::other) + }); + } + } + files + } +} +struct Proc(Host); +impl Proc { + fn folder(&self, path: &str) -> Result<(SelfFolder, String)> { + let (pid, rest) = path + .trim_start_matches('/') + .split_once('/') + .unwrap_or((path.trim_start_matches('/'), "")); + let memory = self + .0 + .0 + .state + .lock() + .unwrap() + .processes + .get(pid) + .map(|p| p.memory.clone()) + .ok_or(io::ErrorKind::NotFound)?; + Ok(( + SelfFolder { + host: self.0.clone(), + pid: Some(pid.into()), + memory, + }, + format!("/{rest}"), + )) + } +} +#[async_trait] +impl Layer for Proc { + async fn stat(&self, path: &str, follow: bool) -> Result { + if matches!(path.trim_end_matches('/'), "" | "/core" | "/core/plugins") { + return Ok(directory()); + } + if let Some(path) = path.strip_prefix("/core/") { + return self.0.core_files().stat(&format!("/{path}"), follow).await; + } + let (folder, path) = self.folder(path)?; + folder.stat(&path, follow).await + } + async fn list(&self, path: &str) -> Result> { + if path == "/" { + let mut entries = vec![entry("core".into(), true)]; + entries.extend(self.0.processes().into_keys().map(|pid| entry(pid, true))); + entries.sort_by(|a, b| a.name.cmp(&b.name)); + return Ok(entries); + } + if path.trim_end_matches('/') == "/core/plugins" { + return Ok(self + .0 + .discovered_plugins() + .into_iter() + .filter(|name| self.0.discover_document(name).is_some()) + .map(|name| entry(name, true)) + .collect()); + } + if path.trim_end_matches('/') == "/core" { + let mut entries = self.0.core_files().list("/").await?; + if !entries.iter().any(|e| e.name == "plugins") { + entries.push(entry("plugins".into(), true)); + } + return Ok(entries); + } + if let Some(path) = path.strip_prefix("/core/") { + return self.0.core_files().list(&format!("/{path}")).await; + } + let (folder, path) = self.folder(path)?; + folder.list(&path).await + } + async fn open(&self, path: &str, options: Open) -> Result> { + if options.write || options.create || options.truncate { + return denied(); + } + if matches!(path.trim_end_matches('/'), "" | "/core" | "/core/plugins") { + return Memory::default().open("/", options).await; + } + if let Some(path) = path.strip_prefix("/core/") { + return self.0.core_files().open(&format!("/{path}"), options).await; + } + let (folder, path) = self.folder(path)?; + folder.open(&path, options).await + } + async fn mkdir(&self, _: &str) -> Result<()> { + denied() + } + async fn remove_file(&self, _: &str) -> Result<()> { + denied() + } + async fn remove_dir(&self, _: &str) -> Result<()> { + denied() + } + async fn rename(&self, _: &str, _: &str) -> Result<()> { + denied() + } +} +struct SelfFolder { + host: Host, + pid: Option, + memory: Memory, +} +impl SelfFolder { + async fn generated(&self, path: &str) -> Result { + let pid = self.pid.as_ref().ok_or(io::ErrorKind::NotFound)?; + let bytes = match path { + "/info.json" => { + let info = self + .host + .0 + .state + .lock() + .unwrap() + .processes + .get(pid) + .map(|p| p.info.clone()) + .ok_or(io::ErrorKind::NotFound)?; + serde_json::to_vec(&info).unwrap() + } + "/log" => { + let lines = self + .host + .0 + .journal + .query(serde_json::json!({"pid":pid,"limit":i64::MAX})) + .await; + let mut bytes = Vec::new(); + for line in lines.as_array().unwrap() { + serde_json::to_writer(&mut bytes, line).unwrap(); + bytes.push(b'\n'); + } + bytes + } + _ => return missing(), + }; + let mut files = Generated::default(); + files.insert(path, move || Ok(bytes.clone())); + Ok(files) + } +} +#[async_trait] +impl Layer for SelfFolder { + async fn stat(&self, path: &str, follow: bool) -> Result { + if reserved(path) { + self.generated(path).await?.stat(path, follow).await + } else { + self.memory.stat(path, follow).await + } + } + async fn list(&self, path: &str) -> Result> { + let mut entries = self.memory.list(path).await?; + if path == "/" && self.pid.is_some() { + entries.extend([entry("info.json".into(), false), entry("log".into(), false)]); + } + Ok(entries) + } + async fn open(&self, path: &str, options: Open) -> Result> { + if reserved(path) { + if options.write || options.create || options.truncate { + return denied(); + } + self.generated(path).await?.open(path, options).await + } else { + self.memory.open(path, options).await + } + } + async fn mkdir(&self, path: &str) -> Result<()> { + if reserved(path) { + denied() + } else { + self.memory.mkdir(path).await + } + } + async fn remove_file(&self, path: &str) -> Result<()> { + if reserved(path) { + denied() + } else { + self.memory.remove_file(path).await + } + } + async fn remove_dir(&self, path: &str) -> Result<()> { + if reserved(path) || path == "/" { + denied() + } else { + self.memory.remove_dir(path).await + } + } + async fn rename(&self, from: &str, to: &str) -> Result<()> { + if reserved(from) || reserved(to) || from == "/" || to == "/" { + denied() + } else { + self.memory.rename(from, to).await + } + } +} diff --git a/cli/src/graph.rs b/cli/src/graph.rs index 7cf8f1d..6a438a9 100644 --- a/cli/src/graph.rs +++ b/cli/src/graph.rs @@ -1,31 +1,257 @@ -//! The interface graph (PROTOCOL.md section 10.4): the `rpc.discover` -//! documents of all plugins, their imports, and the result of each check. -//! -//! The `/proc` layer reads the files of `/proc/core/` through these -//! functions. They must be cheap and must not block: they return cached data. -use crate::Host; -use serde_json::Value; +//! Cached discovery documents and import checks under PROTOCOL.md section 10.4. +//! Refreshes use immutable copies. The proc layer only reads serialized cache entries. +use crate::{Host, RpcResult}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use std::{ + collections::{BTreeMap, HashMap}, + path::Path, +}; -// The /proc layer (T6) uses these functions. Remove this when it does. -#[allow(dead_code)] -impl Host { - /// The content of `/proc/core/graph.json`. - pub(crate) fn graph_json(&self) -> Vec { - b"{}\n".to_vec() - } +#[derive(Clone)] +pub(crate) struct Interface { + pub(crate) hash: String, + pub(crate) document: Value, + pub(crate) error: Option, +} +pub(crate) struct Graph { + pub(crate) plugins: BTreeMap, + pub(crate) versions: HashMap, + json: Vec, + dot: Vec, +} - /// The content of `/proc/core/graph.dot`. - pub(crate) fn graph_dot(&self) -> Vec { - b"digraph plugins {}\n".to_vec() - } - - /// The plugins that have an `rpc.discover` document, for `/proc/core/plugins/`. - pub(crate) fn discovered_plugins(&self) -> Vec { - Vec::new() - } - - /// The `rpc.discover` document of the current `plugin.wasm` of `plugin`. - pub(crate) fn discover_document(&self, _plugin: &str) -> Option { - None +impl Default for Graph { + /// Exposes a valid empty graph before background discovery completes. + fn default() -> Self { + Self { + plugins: BTreeMap::new(), + versions: HashMap::new(), + json: b"{\"plugins\":{}}".to_vec(), + dot: b"digraph plugins {}\n".to_vec(), + } } } + +/// Hashes exactly the bytes copied for discovery or installation. +pub(crate) fn hash(bytes: &[u8]) -> String { + format!("{:x}", Sha256::digest(bytes)) +} + +/// Lists broken imports, optionally restricted to changed consumers or providers. +pub(crate) fn problems( + plugins: &BTreeMap, + changed: Option<&[String]>, +) -> Vec { + let mut failures = Vec::new(); + for (consumer, interface) in plugins { + if let Some(imports) = interface.document["x-sausage-imports"].as_object() { + for (provider, expected) in imports { + if changed + .is_some_and(|names| !names.contains(consumer) && !names.contains(provider)) + { + continue; + } + let Some(actual) = plugins.get(provider) else { + failures.push(json!({"consumer":consumer,"provider":provider,"status":"missing","message":"the provider does not exist"})); + continue; + }; + if let Some(error) = &actual.error { + failures.push(json!({"consumer":consumer,"provider":provider,"status":"unavailable","message":"discovery failed","error":error})); + continue; + } + for problem in sausage_interface::compat::check(expected, &actual.document) { + failures.push(json!({"consumer":consumer,"provider":provider,"status":"broken","method":problem.method,"message":problem.message,"counterexample":problem.counterexample,"display":problem.to_string()})); + } + } + } + } + failures +} + +impl Host { + /// The cached JSON graph. This does not perform filesystem or RPC operations. + pub(crate) fn graph_json(&self) -> Vec { + self.0.graph.lock().unwrap().json.clone() + } + /// The cached DOT graph, with an edge for every import. + pub(crate) fn graph_dot(&self) -> Vec { + self.0.graph.lock().unwrap().dot.clone() + } + /// Plugin names currently present in the discovery cache. + pub(crate) fn discovered_plugins(&self) -> Vec { + self.0 + .graph + .lock() + .unwrap() + .plugins + .keys() + .cloned() + .collect() + } + /// The current document, if discovery succeeded. + pub(crate) fn discover_document(&self, plugin: &str) -> Option { + self.0 + .graph + .lock() + .unwrap() + .plugins + .get(plugin) + .filter(|entry| entry.error.is_none()) + .map(|entry| entry.document.clone()) + } + /// Discovers a private copy and caches successful documents by file hash. + pub(crate) async fn discover_copy( + &self, + plugin: &str, + path: &Path, + bytes: &[u8], + ) -> Result { + let hash = hash(bytes); + let cache_key = format!("{plugin}/{hash}"); + let document = self + .route_from( + "core".into(), + Some(json!(uuid::Uuid::now_v7().to_string())), + format!("{plugin}.rpc.discover"), + json!({}), + true, + Some(path.to_path_buf()), + ) + .await?; + if !document["methods"].is_array() { + return Err(crate::error( + -32003, + "rpc.discover did not return an OpenRPC document", + )); + } + self.0 + .graph + .lock() + .unwrap() + .versions + .insert(cache_key, document.clone()); + Ok(Interface { + hash, + document, + error: None, + }) + } + /// Refreshes every installed plugin while the caller holds the activation mutex. + pub(crate) async fn refresh_graph(&self) -> RpcResult { + let mut directory = tokio::fs::read_dir(self.0.world.root.join("plugins")) + .await + .map_err(io_error)?; + let mut plugins = BTreeMap::new(); + while let Some(entry) = directory.next_entry().await.map_err(io_error)? { + let name = entry.file_name().to_string_lossy().into_owned(); + let source = entry.path().join("plugin.wasm"); + let bytes = match tokio::fs::read(&source).await { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(io_error(error)), + }; + let hash = hash(&bytes); + let previous = self.0.graph.lock().unwrap().plugins.get(&name).cloned(); + let interface = if let Some(previous) = + previous.filter(|entry| entry.hash == hash && entry.error.is_none()) + { + previous + } else { + let copy = entry + .path() + .join(format!("discover-{}.wasm", uuid::Uuid::now_v7())); + tokio::fs::write(©, &bytes).await.map_err(io_error)?; + let result = self.discover_copy(&name, ©, &bytes).await; + let _ = tokio::fs::remove_file(©).await; + match result { + Ok(interface) => interface, + Err(error) => Interface { + hash, + document: Value::Null, + error: Some(error), + }, + } + }; + plugins.insert(name, interface); + } + let admission = self.0.admission.write().await; + let old_plugins = self.0.graph.lock().unwrap().plugins.clone(); + let mut upgrades = Vec::new(); + for (plugin, current) in &plugins { + if let Some(old) = old_plugins + .get(plugin) + .filter(|old| 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, + )); + } + } + self.publish_graph(plugins); + drop(admission); + for upgrade in upgrades { + self.begin_upgrade(upgrade); + } + Ok(json!({})) + } + /// Rebuilds serialized snapshots after a watcher refresh or an activation. + pub(crate) fn publish_graph(&self, plugins: BTreeMap) { + let failures = problems(&plugins, None); + for failure in &failures { + self.0 + .journal + .write("core", "core", "core", &format!("broken import {failure}")); + } + let mut entries = serde_json::Map::new(); + let mut dot = String::from("digraph plugins {\n"); + for (name, interface) in &plugins { + dot.push_str(&format!(" {};\n", json!(name))); + let mut imports = serde_json::Map::new(); + for provider in interface.document["x-sausage-imports"] + .as_object() + .into_iter() + .flat_map(|imports| imports.keys()) + { + let issues: Vec<_> = failures + .iter() + .filter(|problem| { + problem["consumer"] == *name && problem["provider"] == *provider + }) + .cloned() + .collect(); + let status = issues + .first() + .and_then(|problem| problem["status"].as_str()) + .unwrap_or("compatible"); + imports.insert(provider.clone(), json!({"status":status,"problems":issues})); + dot.push_str(&format!( + " {} -> {} [label={}];\n", + json!(name), + json!(provider), + json!(status) + )); + } + entries.insert( + name.clone(), + json!({"hash":interface.hash,"imports":imports,"error":interface.error}), + ); + } + dot.push_str("}\n"); + let mut graph = self.0.graph.lock().unwrap(); + graph.json = serde_json::to_vec(&json!({"plugins":entries})).unwrap(); + graph.dot = dot.into_bytes(); + graph.plugins = plugins; + } +} +/// Converts a filesystem failure into a JSON-RPC internal error. +pub(crate) fn io_error(error: std::io::Error) -> Value { + crate::error(-32603, &error.to_string()) +} diff --git a/cli/src/host.rs b/cli/src/host.rs index 6a9520a..7b0b4a9 100644 --- a/cli/src/host.rs +++ b/cli/src/host.rs @@ -17,6 +17,7 @@ pub(crate) struct State { pub(crate) processes: HashMap, pub(crate) locks: HashMap, pub(crate) native: HashMap, + pub(crate) services: Vec, } pub(crate) struct Inner { pub(crate) world: World, @@ -26,6 +27,15 @@ pub(crate) struct Inner { pub(crate) notifications: tokio::sync::broadcast::Sender, pub(crate) exit: watch::Sender, pub(crate) components: components::Components, + /// Calls hold it for reading from the choice of a version until their + /// process is in the table. Activation holds it for writing while it + /// installs a version, so that it sees every process of the old version. + pub(crate) admission: tokio::sync::RwLock<()>, + pub(crate) run: sausage_vfs::Memory, + pub(crate) graph: Mutex, + pub(crate) activations: tokio::sync::Mutex<()>, + pub(crate) watcher: Mutex>, + pub(crate) retirements: Mutex>, } #[derive(Clone)] /// A cloneable handle to a world and its running processes. @@ -60,10 +70,17 @@ impl Host { processes: HashMap::new(), locks: HashMap::new(), native: HashMap::new(), + services: Vec::new(), }), notifications: tokio::sync::broadcast::channel(1024).0, exit: watch::channel(false).0, components: components::Components::default(), + admission: tokio::sync::RwLock::new(()), + run: sausage_vfs::Memory::default(), + graph: Mutex::new(crate::graph::Graph::default()), + activations: tokio::sync::Mutex::new(()), + watcher: Mutex::new(None), + retirements: Mutex::new(std::collections::HashSet::new()), })); host.boot().await?; Ok(host) diff --git a/cli/src/lib.rs b/cli/src/lib.rs index 2f78937..2ebea71 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -5,6 +5,7 @@ mod components; pub mod console; mod core; mod execution; +mod filesystem; mod graph; mod host; mod journal; @@ -14,6 +15,8 @@ mod process; mod routing; pub mod runtime; mod services; +mod upgrades; +mod watcher; pub mod world; pub use host::Host; pub use native::{NativeContext, NativeFuture, NativePlugin}; diff --git a/cli/src/process.rs b/cli/src/process.rs index ca08bbd..ecce342 100644 --- a/cli/src/process.rs +++ b/cli/src/process.rs @@ -24,6 +24,9 @@ pub struct ProcessInfo { } pub(crate) struct Process { pub(crate) info: ProcessInfo, + pub(crate) memory: sausage_vfs::Memory, + pub(crate) files: sausage_vfs::Process, + pub(crate) store_ended: watch::Sender, pub(crate) input: Option>, pub(crate) abort: Option, pub(crate) caller: String, @@ -31,19 +34,19 @@ pub(crate) struct Process { pub(crate) answer: Option>, pub(crate) ended: watch::Sender, } -pub(crate) struct ProcessGuard { - pub(crate) host: Host, - pub(crate) pid: String, -} -impl Drop for ProcessGuard { - fn drop(&mut self) { - self.host.finish(&self.pid, "process task ended"); - } -} - impl Host { - /// Remove a process and release its locks before asynchronous journal completion. - pub(crate) fn finish(&self, pid: &str, status: &str) { + /// Called only after the execution task ends. Wait for the store and file operations before releasing locks. + pub(crate) async fn finish(&self, pid: &str, status: &str) { + let (files, mut store_ended) = { + let mut state = self.0.state.lock().unwrap(); + let Some(p) = state.processes.get_mut(pid) else { + return; + }; + p.info.state = "stopping".into(); + (p.files.clone(), p.store_ended.subscribe()) + }; + let _ = store_ended.wait_for(|ended| *ended).await; + files.idle().await; let mut state = self.0.state.lock().unwrap(); let process = state.processes.remove(pid); if let Some(p) = process { @@ -106,13 +109,14 @@ impl Host { } /// Abort a process task and perform its ended-state cleanup. pub(crate) fn kill(&self, pid: &str) { - if let Some(p) = self.0.state.lock().unwrap().processes.get(pid) { + if let Some(p) = self.0.state.lock().unwrap().processes.get_mut(pid) { + p.info.state = "stopping".into(); + p.input.take(); self.0.journal.write(pid, &p.info.plugin, "core", "kill"); if let Some(a) = &p.abort { a.abort(); } } - self.finish(pid, "killed"); } /// Stop every process in parallel and wait for journal writes to finish. pub async fn stop(&self) { diff --git a/cli/src/routing.rs b/cli/src/routing.rs index 511c439..16929d1 100644 --- a/cli/src/routing.rs +++ b/cli/src/routing.rs @@ -93,6 +93,7 @@ impl Host { { return Err(error(-32601, "invalid plugin name")); } + let admission = host.0.admission.read().await; let native = host.0.state.lock().unwrap().native.get(plugin).cloned(); let (component, version) = if native.is_some() { (None, "native".into()) @@ -141,6 +142,9 @@ impl Host { pid.clone(), Process { info: info.clone(), + memory: sausage_vfs::Memory::default(), + files: sausage_vfs::Process::default(), + store_ended: watch::channel(true).0, input: Some(input), abort: None, caller, @@ -153,15 +157,34 @@ impl Host { let h = host.clone(); let p = pid.clone(); let has_id = id.is_some(); + let (start, ready) = oneshot::channel(); let task = tokio::spawn(async move { + let _ = ready.await; h.run_process(p, info, component, native, incoming, has_id) - .await; + .await }); if let Some(p) = host.0.state.lock().unwrap().processes.get_mut(&pid) { p.abort = Some(task.abort_handle()); + if p.info.state == "stopping" { + task.abort(); + } } else { task.abort(); } + drop(admission); + let cleanup = host.clone(); + let finished_pid = pid.clone(); + tokio::spawn(async move { + let status = task.await.unwrap_or_else(|error| { + if error.is_cancelled() { + "killed".into() + } else { + format!("host task failed: {error}") + } + }); + cleanup.finish(&finished_pid, &status).await; + }); + let _ = start.send(()); if id.is_none() { return Ok(Value::Null); } diff --git a/cli/src/runtime.rs b/cli/src/runtime.rs index affe4a3..c3a7f2b 100644 --- a/cli/src/runtime.rs +++ b/cli/src/runtime.rs @@ -3,7 +3,7 @@ //! An instance is one run of a `wasi:cli/command` component. The host writes //! NDJSON lines to its stdin and reads NDJSON lines from its stdout. -use std::path::{Path, PathBuf}; +use std::path::Path; use std::sync::Arc; use std::time::Duration; @@ -17,7 +17,7 @@ use wasmtime_wasi::cli::{AsyncStdinStream, AsyncStdoutStream}; use wasmtime_wasi::p2::bindings::Command; -use wasmtime_wasi::{FsPerms, I32Exit, WasiCtx, WasiCtxView, WasiView}; +use wasmtime_wasi::{I32Exit, WasiCtx, WasiCtxView, WasiView}; use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpHooks, WasiHttpView}; /// The HTTP transport hooks supplied to WASI processes. @@ -61,6 +61,7 @@ pub struct ProcessState { table: ResourceTable, http: WasiHttpCtx, hooks: HttpHooks, + fs: sausage_vfs::adapter::Adapter, } impl WasiView for ProcessState { @@ -111,8 +112,10 @@ impl std::fmt::Display for Exit { /// The filesystem, network, and environment supplied to one process. pub struct InstanceConfig { - /// The world folder to preopen as the guest root. - pub world: PathBuf, + /// The virtual namespace exposed through the single root preopen. + pub filesystem: sausage_vfs::Mounts, + /// The store completion barrier used by process cleanup. + pub store_ended: tokio::sync::watch::Sender, /// Outgoing HTTP, and TCP, UDP, and DNS through `wasi:sockets`. pub network: bool, /// The environment variables supplied to the guest. @@ -127,12 +130,20 @@ pub struct Instance { pub stdout: mpsc::Receiver, /// The stderr lines that the process writes. pub stderr: mpsc::Receiver, - task: JoinHandle, + task: Option>, + store_ended: tokio::sync::watch::Sender, } impl Drop for Instance { fn drop(&mut self) { - self.task.abort(); + if let Some(task) = self.task.take() { + task.abort(); + let ended = self.store_ended.clone(); + tokio::spawn(async move { + let _ = task.await; + ended.send_replace(true); + }); + } } } @@ -166,7 +177,10 @@ impl Instance { /// Wait until the guest ends. pub async fn wait(&mut self) -> Exit { - match (&mut self.task).await { + let result = self.task.as_mut().unwrap().await; + self.task.take(); + self.store_ended.send_replace(true); + match result { Ok(exit) => exit, Err(e) if e.is_cancelled() => Exit::Trap("killed".into()), Err(e) => Exit::Trap(format!("host task failed: {e}")), @@ -198,6 +212,10 @@ impl Runtime { let mut linker: Linker = Linker::new(&engine); wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; + sausage_vfs::adapter::add_to_linker(&mut linker, |state| sausage_vfs::adapter::View { + adapter: &mut state.fs, + table: &mut state.table, + })?; wasmtime_wasi_http::p2::add_only_http_to_linker_async(&mut linker)?; // Time slices: every 10 ms each running guest yields to the scheduler. let runtime = Arc::new(Self { engine, linker }); @@ -220,7 +238,7 @@ impl Runtime { Ok(Component::new(&self.engine, bytes)?) } - /// Prepare a world preopen off the async workers, then start the WASI task. + /// Start a WASI task with the supplied virtual namespace. pub async fn start( self: &Arc, component: &Component, @@ -243,18 +261,13 @@ impl Runtime { for (k, v) in &cfg.env { b.env(k, v); } - // Filesystem setup lives here. T6 replaces this preopen with the VFS. - let world = cfg.world; - let mut b = tokio::task::spawn_blocking(move || -> Result<_> { - b.preopened_dir(world, "/", FsPerms::ReadWrite)?; - Ok(b) - }) - .await??; + let state = ProcessState { ctx: b.build(), table: ResourceTable::new(), http: WasiHttpCtx::new(), hooks: HttpHooks { allow: cfg.network }, + fs: sausage_vfs::adapter::Adapter::new(cfg.filesystem), }; let (line_tx, line_rx) = mpsc::channel(1024); @@ -285,6 +298,7 @@ impl Runtime { }); let rt = self.clone(); let component = component.clone(); + cfg.store_ended.send_replace(false); let task = tokio::spawn(async move { let mut store = Store::new(&rt.engine, state); store.epoch_deadline_async_yield_and_update(1); @@ -311,7 +325,8 @@ impl Runtime { stdin: Some(host_in), stdout: line_rx, stderr: err_rx, - task, + task: Some(task), + store_ended: cfg.store_ended, }) } } diff --git a/cli/src/services.rs b/cli/src/services.rs index 7faf67a..48ad2d8 100644 --- a/cli/src/services.rs +++ b/cli/src/services.rs @@ -1,31 +1,46 @@ -//! Services: the calls in `/config/boot.json` (PROTOCOL.md section 8). The -//! core starts them in order at boot. +//! Ordered boot services and their identities for upgrade restarts (PROTOCOL.md section 8). use crate::Host; use serde_json::{Value, json}; -use uuid::Uuid; - +#[derive(Clone)] +pub(crate) struct Service { + pub(crate) method: String, + pub(crate) params: Value, + pub(crate) request: Value, +} impl Host { - /// Start each service of `/config/boot.json` as a detached call with the - /// parent `core`, in order. A failure goes to the journal, and boot - /// continues. + /// Starts discovery in the background, then waits for each boot service in order. pub(crate) async fn boot(&self) -> anyhow::Result<()> { + self.watch_plugins()?; let boot: Vec = serde_json::from_slice( &tokio::fs::read(self.0.world.root.join("config/boot.json")).await?, )?; for service in boot { - let result = self - .route( - "core".into(), - Some(json!(Uuid::now_v7().to_string())), - service["method"].as_str().unwrap_or("").into(), - service["params"].clone(), - true, - ) - .await; - if let Err(e) = result { - self.0.journal.write("core", "core", "core", &e.to_string()); - } + self.start_service( + service["method"].as_str().unwrap_or("").into(), + service["params"].clone(), + ) + .await; } Ok(()) } + /// Records a unique request identity before the service can answer or exit. + pub(crate) async fn start_service(&self, method: String, params: Value) { + let request = json!(uuid::Uuid::now_v7().to_string()); + self.0.state.lock().unwrap().services.push(Service { + method: method.clone(), + params: params.clone(), + request: request.clone(), + }); + self.0 + .journal + .write("core", "core", "core", &format!("service start {method}")); + if let Err(error) = self + .route("core".into(), Some(request), method, params, true) + .await + { + self.0 + .journal + .write("core", "core", "core", &format!("service failed {error}")); + } + } } diff --git a/cli/src/upgrades.rs b/cli/src/upgrades.rs new file mode 100644 index 0000000..eec0a4b --- /dev/null +++ b/cli/src/upgrades.rs @@ -0,0 +1,157 @@ +//! Grace periods, service restarts, and retirement of replaced versions (PROTOCOL.md section 3.7). +use crate::{Host, services::Service}; +use serde_json::json; +use std::{collections::BTreeMap, time::Duration}; +use tokio::sync::watch; + +pub(crate) struct OldProcess { + pid: String, + version: String, + ended: watch::Receiver, +} +pub(crate) struct Upgrade { + plugin: String, + previous: String, + grace: Duration, + processes: Vec, + services: Vec<(Service, Option>)>, +} +impl Host { + /// Snapshots old admissions while activation holds the admission write guard. + pub(crate) fn snapshot_upgrade( + &self, + plugin: &str, + previous: String, + current: &str, + grace: f64, + ) -> Upgrade { + let mut state = self.0.state.lock().unwrap(); + let processes = state + .processes + .iter() + .filter(|(_, process)| process.info.plugin == plugin && process.info.version != current) + .map(|(pid, process)| OldProcess { + pid: pid.clone(), + version: process.info.version.clone(), + ended: process.ended.subscribe(), + }) + .collect(); + let mut services = Vec::new(); + let mut retained = Vec::new(); + for service in std::mem::take(&mut state.services) { + if service + .method + .split_once('.') + .is_some_and(|(name, _)| name == plugin) + { + let ended = state + .processes + .values() + .find(|process| process.id.as_ref() == Some(&service.request)) + .map(|process| process.ended.subscribe()); + services.push((service, ended)); + } else { + retained.push(service); + } + } + state.services = retained; + Upgrade { + plugin: plugin.into(), + previous, + grace: Duration::from_secs_f64(grace), + processes, + services, + } + } + /// Lets old calls finish, stops them after grace, and starts each lifecycle hook once. + pub(crate) fn begin_upgrade(&self, upgrade: Upgrade) { + let host = self.clone(); + tokio::spawn(async move { + host.0.journal.write( + "core", + &upgrade.plugin, + "core", + &format!("upgrade grace {} seconds", upgrade.grace.as_secs_f64()), + ); + let mut versions: BTreeMap>> = BTreeMap::new(); + versions.entry(upgrade.previous).or_default(); + for process in upgrade.processes { + versions + .entry(process.version) + .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); } + } + }); + } + for (service, ended) in upgrade.services { + let host = host.clone(); + tokio::spawn(async move { + if let Some(mut ended) = ended { + let _ = ended.wait_for(|ended| *ended).await; + } + host.start_service(service.method, service.params).await; + }); + } + for (version, processes) in versions { + let host = host.clone(); + let plugin = upgrade.plugin.clone(); + let retirement = (plugin.clone(), version.clone()); + if !host + .0 + .retirements + .lock() + .unwrap() + .insert(retirement.clone()) + { + continue; + } + tokio::spawn(async move { + for mut ended in processes { + let _ = ended.wait_for(|ended| *ended).await; + } + if host + .0 + .graph + .lock() + .unwrap() + .plugins + .get(&plugin) + .is_some_and(|current| current.hash == version) + { + host.0.retirements.lock().unwrap().remove(&retirement); + return; + } + host.0.journal.write( + "core", + &plugin, + "core", + &format!("retired start {version}"), + ); + let result = host + .route( + "core".into(), + Some(json!(uuid::Uuid::now_v7().to_string())), + format!("{plugin}.rpc.retired"), + json!({"previous":version}), + true, + ) + .await; + host.0.journal.write( + "core", + &plugin, + "core", + &format!("retired result {result:?}"), + ); + host.0.retirements.lock().unwrap().remove(&retirement); + }); + } + }); + } +} diff --git a/cli/src/watcher.rs b/cli/src/watcher.rs new file mode 100644 index 0000000..9eacde2 --- /dev/null +++ b/cli/src/watcher.rs @@ -0,0 +1,48 @@ +//! Background discovery and file notifications for installed plugin versions. +use crate::Host; +use notify::{RecursiveMode, Watcher}; +impl Host { + /// Starts background discovery and retains the watcher for this host's lifetime. + pub(crate) fn watch_plugins(&self) -> anyhow::Result<()> { + let (events, mut incoming) = tokio::sync::mpsc::unbounded_channel(); + let mut watcher = + notify::recommended_watcher(move |event: notify::Result| { + if let Ok(event) = event + && !matches!(event.kind, notify::EventKind::Access(_)) + && event + .paths + .iter() + .any(|path| path.file_name().is_some_and(|name| name == "plugin.wasm")) + { + let _ = events.send(()); + } + })?; + watcher.watch(&self.0.world.root.join("plugins"), RecursiveMode::Recursive)?; + *self.0.watcher.lock().unwrap() = Some(watcher); + let weak = std::sync::Arc::downgrade(&self.0); + tokio::spawn(async move { + loop { + let Some(inner) = weak.upgrade() else { + break; + }; + let host = Host(inner); + let _activation = host.0.activations.lock().await; + if let Err(error) = host.refresh_graph().await { + host.0.journal.write( + "core", + "core", + "core", + &format!("discovery failed {error}"), + ); + } + drop(_activation); + drop(host); + if incoming.recv().await.is_none() { + break; + } + while incoming.try_recv().is_ok() {} + } + }); + Ok(()) + } +} diff --git a/cli/tests/activation.rs b/cli/tests/activation.rs new file mode 100644 index 0000000..c1098e6 --- /dev/null +++ b/cli/tests/activation.rs @@ -0,0 +1,491 @@ +//! Activation, graph, and upgrade integration tests with raw WASI guests. +use sausage::{Host, world::World}; +use serde_json::{Value, json}; +use std::{ + path::{Path, PathBuf}, + sync::OnceLock, + time::Duration, +}; + +/// Builds two genuine guest versions in this executor's target folder. +fn fixture(version: usize) -> PathBuf { + static FILES: OnceLock> = OnceLock::new(); + FILES.get_or_init(|| { + let root = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap(); + let target = root.join(format!( + "target/compat/activation-guests/{}", + std::process::id() + )); + (1..=2) + .map(|version| { + let status = std::process::Command::new("cargo") + .env("FIXTURE_VERSION", version.to_string()) + .args([ + "build", + "--offline", + "--target", + "wasm32-wasip2", + "--manifest-path", + ]) + .arg(root.join("cli/tests/plugins/activation/Cargo.toml")) + .arg("--target-dir") + .arg(&target) + .status() + .unwrap(); + assert!(status.success()); + let copy = target.join(format!("v{version}.wasm")); + std::fs::copy( + target.join("wasm32-wasip2/debug/activation-guest.wasm"), + ©, + ) + .unwrap(); + copy + }) + .collect() + })[version - 1] + .clone() +} +struct TestWorld(PathBuf); +impl TestWorld { + /// Creates an isolated world with two version documents per plugin. + fn new() -> Self { + let path = + std::env::temp_dir().join(format!("sausage-activation-{}", uuid::Uuid::now_v7())); + World::create(&path).unwrap(); + Self(path) + } + /// Installs version one and stages version two under the world state. + fn plugin(&self, name: &str, old: Value, new: Value) { + std::fs::create_dir_all(self.0.join(format!("plugins/{name}"))).unwrap(); + std::fs::create_dir_all(self.0.join(format!("state/{name}"))).unwrap(); + std::fs::copy( + fixture(1), + self.0.join(format!("plugins/{name}/plugin.wasm")), + ) + .unwrap(); + std::fs::copy(fixture(2), self.0.join(format!("state/{name}/next.wasm"))).unwrap(); + std::fs::write(self.0.join(format!("state/{name}/1.json")), old.to_string()).unwrap(); + std::fs::write(self.0.join(format!("state/{name}/2.json")), new.to_string()).unwrap(); + } + /// Starts a host with an independent compile cache shared by these tests. + async fn start(&self) -> Host { + Host::start_with_cache( + &self.0, + &Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .unwrap() + .join("target/compat/runtime-cache"), + ) + .await + .unwrap() + } +} +impl Drop for TestWorld { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} +/// Builds a method-only document with a short upgrade grace period. +fn document(name: &str) -> Value { + json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}],"x-sausage-grace":0.05}) +} +/// Adds an import of a provider's method list. +fn consumer(provider: &str, expected: Value) -> Value { + let mut document = document("consume"); + document["x-sausage-imports"] = json!({provider:expected}); + document +} +/// Performs a bounded test call without imposing a host execution limit. +async fn call(host: &Host, method: &str, params: Value) -> Value { + tokio::time::timeout(Duration::from_secs(60), host.call(method, params)) + .await + .unwrap() + .unwrap() +} +/// Activates the staged candidate. +async fn activate(host: &Host, name: &str) -> Result { + tokio::time::timeout( + Duration::from_secs(60), + host.call( + "core.activate", + json!({"plugin":name,"wasm":format!("/state/{name}/next.wasm")}), + ), + ) + .await + .unwrap() +} +/// Waits for a lifecycle marker without a fixed delay. +async fn marker(path: PathBuf, expected: &str) { + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if std::fs::read_to_string(&path).ok().as_deref() == Some(expected) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} +/// Reads the cached graph through the actual proc adapter. +async fn graph(host: &Host) -> Value { + call(host, "notes.read", json!({"path":"/proc/core/graph.json"})).await +} + +/// PROTOCOL 10.4: proc graph files contain valid documents before discovery completes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn graph_is_valid_immediately() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("add")); + let host = world.start().await; + assert!(graph(&host).await["plugins"].is_object()); + let dot = call(&host, "notes.read", json!({"path":"/proc/core/graph.dot"})).await; + assert!(dot.as_str().unwrap().starts_with("digraph plugins")); + host.stop().await; +} +/// PROTOCOL 10.3: a rejected group leaves every installed file unchanged. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn rejects_group_without_changes() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("other")); + world.plugin( + "agent", + consumer("notes", document("add")), + consumer("notes", document("missing")), + ); + let host = world.start().await; + let result = host.call("core.activate",json!({"plugins":[{"plugin":"notes","wasm":"/state/notes/next.wasm"},{"plugin":"agent","wasm":"/state/agent/next.wasm"}]})).await; + assert_eq!(result.unwrap_err()["code"], -32003); + for name in ["notes", "agent"] { + assert_eq!( + std::fs::read(world.0.join(format!("plugins/{name}/plugin.wasm"))).unwrap(), + std::fs::read(fixture(1)).unwrap() + ); + } + host.stop().await; +} +/// PROTOCOL 10.3: activation discovers and installs a copied candidate. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn activates_copy() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("add")); + let host = world.start().await; + let result = activate(&host, "notes").await.unwrap(); + assert!(result["notes"]["methods"].is_array()); + assert_eq!( + call(&host, "notes.version", json!({})).await["version"], + "2" + ); + assert_eq!( + std::fs::read(world.0.join("plugins/notes/plugin.prev.wasm")).unwrap(), + std::fs::read(fixture(1)).unwrap() + ); + host.stop().await; +} +/// PROTOCOL 10.3: changes to the original after discovery starts cannot change the installed copy. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn source_changes_do_not_change_candidate() { + let world = TestWorld::new(); + let mut next = document("add"); + next["x-test-wait"] = json!(true); + world.plugin("notes", document("add"), next); + let host = world.start().await; + let activation_host = host.clone(); + let activation = tokio::spawn(async move { activate(&activation_host, "notes").await }); + marker(world.0.join("state/notes/discover-started"), "ready").await; + std::fs::copy(fixture(1), world.0.join("state/notes/next.wasm")).unwrap(); + std::fs::write(world.0.join("state/notes/discover-release"), "ready").unwrap(); + activation.await.unwrap().unwrap(); + assert_eq!( + std::fs::read(world.0.join("plugins/notes/plugin.wasm")).unwrap(), + std::fs::read(fixture(2)).unwrap() + ); + host.stop().await; +} +/// PROTOCOL 3.7: plugins without a declared grace period retain old calls before the default deadline. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn default_grace_retains_old_process() { + let world = TestWorld::new(); + let mut old = document("hold"); + old.as_object_mut().unwrap().remove("x-sausage-grace"); + world.plugin("notes", old, document("hold")); + let host = world.start().await; + call(&host, "notes.hold", json!({})).await; + let pid = host + .processes() + .into_iter() + .find(|(_, process)| process.method == "hold") + .unwrap() + .0; + activate(&host, "notes").await.unwrap(); + assert_eq!(host.processes()[&pid].state, "running"); + 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() { + let world = TestWorld::new(); + world.plugin( + "notes", + document("add"), + consumer("absent", document("add")), + ); + let host = world.start().await; + let failure = activate(&host, "notes").await.unwrap_err(); + assert_eq!(failure["code"], -32003); + assert_eq!(failure["data"][0]["status"], "missing"); + host.stop().await; +} +/// PROTOCOL 10.3: current consumers block an incompatible provider upgrade. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn checks_current_consumers() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("other")); + world.plugin( + "agent", + consumer("notes", document("add")), + document("consume"), + ); + let host = world.start().await; + let failure = activate(&host, "notes").await.unwrap_err(); + assert_eq!(failure["code"], -32003); + assert_eq!(failure["data"][0]["method"], "add"); + assert_eq!( + std::fs::read(world.0.join("plugins/notes/plugin.wasm")).unwrap(), + std::fs::read(fixture(1)).unwrap() + ); + host.stop().await; +} +/// PROTOCOL 10.3: a group checks the proposed set rather than intermediate versions. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn activates_group() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("other")); + world.plugin( + "agent", + consumer("notes", document("add")), + consumer("notes", document("other")), + ); + let host = world.start().await; + call(&host,"core.activate",json!({"plugins":[{"plugin":"notes","wasm":"/state/notes/next.wasm"},{"plugin":"agent","wasm":"/state/agent/next.wasm"}]})).await; + assert_eq!( + call(&host, "agent.version", json!({})).await["version"], + "2" + ); + host.stop().await; +} +/// PROTOCOL 10.3: force skips compatibility checks. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn force_skips_checks() { + let world = TestWorld::new(); + world.plugin( + "notes", + document("add"), + consumer("absent", document("add")), + ); + let host = world.start().await; + call( + &host, + "core.activate", + json!({"plugin":"notes","wasm":"/state/notes/next.wasm","force":true}), + ) + .await; + host.stop().await; +} +/// PROTOCOL 10.3: revert activates the previous copy and preserves the replaced version. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn reverts_previous_version() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("add")); + let host = world.start().await; + activate(&host, "notes").await.unwrap(); + call(&host, "core.revert", json!({"plugin":"notes"})).await; + assert_eq!( + call(&host, "notes.version", json!({})).await["version"], + "1" + ); + assert_eq!( + std::fs::read(world.0.join("plugins/notes/plugin.prev.wasm")).unwrap(), + std::fs::read(fixture(2)).unwrap() + ); + host.stop().await; +} +/// PROTOCOL 10.3: concurrent activations serialize their installation and revert state. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn serializes_activations() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("add")); + let host = world.start().await; + let (first, second) = tokio::join!(activate(&host, "notes"), activate(&host, "notes")); + first.unwrap(); + second.unwrap(); + assert_eq!( + std::fs::read(world.0.join("plugins/notes/plugin.prev.wasm")).unwrap(), + std::fs::read(fixture(2)).unwrap() + ); + host.stop().await; +} +/// PROTOCOL 10.4: startup fetches every installed document before activation checks consumers. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discovers_all_before_checks() { + checks_current_consumers_body().await; +} +/// Exercises a consumer whose discovery must finish before the provider can change. +async fn checks_current_consumers_body() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("other")); + world.plugin( + "agent", + consumer("notes", document("add")), + document("consume"), + ); + let host = world.start().await; + assert_eq!(activate(&host, "notes").await.unwrap_err()["code"], -32003); + host.stop().await; +} +/// PROTOCOL 10.4: manual copies update cached hashes and expose broken imports in the journal. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn watches_manual_copy() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("other")); + world.plugin( + "agent", + consumer("notes", document("add")), + document("consume"), + ); + let host = world.start().await; + // A same-version activation waits for initial discovery without changing the provider. + call( + &host, + "core.activate", + json!({"plugin":"notes","wasm":"/plugins/notes/plugin.wasm"}), + ) + .await; + std::fs::copy(fixture(2), world.0.join("plugins/notes/plugin.wasm")).unwrap(); + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let graph = graph(&host).await; + if graph["plugins"]["agent"]["imports"]["notes"]["status"] == "broken" { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!( + !call(&host, "core.log", json!({"grep":"broken import"})) + .await + .as_array() + .unwrap() + .is_empty() + ); + host.stop().await; +} +/// PROTOCOL 10.4: nonexistent providers appear as missing imports. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn graph_missing_import() { + let world = TestWorld::new(); + world.plugin( + "notes", + consumer("absent", document("add")), + document("add"), + ); + let host = world.start().await; + call( + &host, + "core.activate", + json!({"plugin":"notes","wasm":"/plugins/notes/plugin.wasm","force":true}), + ) + .await; + assert_eq!( + graph(&host).await["plugins"]["notes"]["imports"]["absent"]["status"], + "missing" + ); + let dot = call(&host, "notes.read", json!({"path":"/proc/core/graph.dot"})).await; + assert!(dot.as_str().unwrap().contains("absent")); + host.stop().await; +} +/// PROTOCOL 3.7: new calls use new code while old calls retain their grace period. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn old_process_grace_and_new_calls() { + let world = TestWorld::new(); + let mut old = document("add"); + old["x-sausage-grace"] = json!(0.5); + world.plugin("notes", old, document("add")); + let host = world.start().await; + call(&host, "notes.hold", json!({})).await; + let pid = host + .processes() + .into_iter() + .find(|(_, process)| process.method == "hold") + .unwrap() + .0; + activate(&host, "notes").await.unwrap(); + assert!(host.processes().contains_key(&pid)); + assert_eq!( + call(&host, "notes.version", json!({})).await["version"], + "2" + ); + marker(world.0.join("state/notes/ended-1"), "ended").await; + host.stop().await; +} +/// PROTOCOL 3.7: retirement receives the previous hash only after old processes end. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn retires_old_version() { + use sha2::{Digest, Sha256}; + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("add")); + let host = world.start().await; + call(&host, "notes.hold", json!({})).await; + activate(&host, "notes").await.unwrap(); + let previous = format!("{:x}", Sha256::digest(std::fs::read(fixture(1)).unwrap())); + marker(world.0.join("state/notes/retired"), &previous).await; + marker(world.0.join("state/notes/ended-1"), "ended").await; + assert!( + !call(&host, "core.log", json!({"grep":"retired start"})) + .await + .as_array() + .unwrap() + .is_empty() + ); + host.stop().await; +} +/// PROTOCOL 8 and 3.7: services restart after their old process ends. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn restarts_service() { + let world = TestWorld::new(); + world.plugin("notes", document("serve"), document("serve")); + std::fs::write( + world.0.join("config/boot.json"), + json!([{"method":"notes.serve","params":{}}]).to_string(), + ) + .unwrap(); + let host = world.start().await; + marker(world.0.join("state/notes/service"), "1").await; + activate(&host, "notes").await.unwrap(); + marker(world.0.join("state/notes/service"), "2").await; + host.stop().await; +} +/// PROTOCOL 8: ordinary detached processes do not become services after an upgrade. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn does_not_restart_detached_call() { + let world = TestWorld::new(); + world.plugin("notes", document("hold"), document("hold")); + let host = world.start().await; + call( + &host, + "core.call", + json!({"method":"notes.hold","params":{},"detach":true}), + ) + .await; + activate(&host, "notes").await.unwrap(); + marker(world.0.join("state/notes/ended-1"), "ended").await; + assert!( + !host + .processes() + .values() + .any(|process| process.method == "hold" && process.state == "running") + ); + host.stop().await; +} diff --git a/cli/tests/core.rs b/cli/tests/core.rs index 0973a3b..62eb581 100644 --- a/cli/tests/core.rs +++ b/cli/tests/core.rs @@ -190,7 +190,7 @@ async fn unimplemented_core_methods_return_method_not_found() { let w = TestWorld::new(); let h = w.start().await; assert_eq!( - h.call("core.activate", json!({})).await.unwrap_err()["code"], + h.call("core.unknown", json!({})).await.unwrap_err()["code"], -32601 ); @@ -1083,8 +1083,9 @@ async fn boot_services_are_detached() { .unwrap(); let h = w.start().await; let processes = h.processes(); - assert_eq!(processes.len(), 1); - assert!(processes.values().next().unwrap().detached); + let services: Vec<_> = processes.values().filter(|p| p.method == "serve").collect(); + assert_eq!(services.len(), 1); + assert!(services[0].detached); h.stop().await; } @@ -1110,3 +1111,256 @@ async fn core_lifetime_events_enter_the_journal() { } h.stop().await; } + +/// PROTOCOL 4: the root merges world files and mounts, and cross-mount rename fails. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_root_and_mount_boundaries() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "probe.filesystem", json!({"topic":"root"})).await; + assert_eq!( + std::fs::read_to_string(w.0.join("state/root")).unwrap(), + "world" + ); + h.stop().await; +} + +/// PROTOCOL 4: a process publishes self files read-only under its own live proc folder. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_self_and_proc_lifetime() { + use sausage_vfs::Layer; + let w = TestWorld::new(); + let h = w.start().await; + let view = h.filesystem().await.unwrap(); + let pid = call(&h, "probe.filesystem", json!({"topic":"self"})).await; + call(&h, "core.stop", json!({"pid":pid})).await; + assert!( + view.stat(&format!("/proc/{}", pid.as_str().unwrap()), true) + .await + .is_err() + ); + let live = call(&h, "probe.self_serve", json!({})).await; + call(&h, "probe.filesystem", json!({"topic":"isolated"})).await; + assert_eq!( + call( + &h, + "probe.read", + json!({"path":format!("/proc/{}/endpoint",live.as_str().unwrap())}) + ) + .await, + "endpoint" + ); + call(&h, "core.stop", json!({"pid":live})).await; + assert!( + view.stat(&format!("/proc/{}", live.as_str().unwrap()), true) + .await + .is_err() + ); + h.stop().await; +} + +/// PROTOCOL 4: core-owned self names reject creation and replacement. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_reserved_self_names() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "probe.filesystem", json!({"topic":"reserved"})).await; + h.stop().await; +} + +/// PROTOCOL 4: generated logs capture a finite snapshot at open. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_log_snapshot_ends() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "probe.filesystem", json!({"topic":"snapshot"})).await; + h.stop().await; +} + +/// PROTOCOL 4: proc/core provides protocol, graph, lock, and plugin inspection files. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_core_inspection_files() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "probe.filesystem", json!({"topic":"core"})).await; + h.stop().await; +} + +/// PROTOCOL 4: proc/core exposes each discovered document and its generated Rust client. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_discovery_document_and_client() { + use sausage_vfs::{Layer, Open}; + let w = TestWorld::new(); + let document = + json!({"methods":[{"name":"echo","params":[],"result":{"schema":{"type":"string"}}}]}); + std::fs::create_dir_all(w.0.join("state/probe")).unwrap(); + std::fs::write(w.0.join("state/probe/discover.json"), document.to_string()).unwrap(); + let h = w.start().await; + let view = h.filesystem().await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(20), async { + loop { + if view + .open("/proc/core/plugins/probe/discover.json", Open::default()) + .await + .is_ok() + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + let published = call( + &h, + "probe.read", + json!({"path":"/proc/core/plugins/probe/discover.json"}), + ) + .await; + assert_eq!( + serde_json::from_str::(published.as_str().unwrap()).unwrap(), + document + ); + let client = call( + &h, + "probe.read", + json!({"path":"/proc/core/plugins/probe/client.rs"}), + ) + .await; + assert_eq!( + client, + sausage_interface::client_rs("probe", &document).unwrap() + ); + h.stop().await; +} + +/// PROTOCOL 4: run is shared across processes but empty after a host restart. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_run_is_shared_and_ephemeral() { + use sausage_vfs::Layer; + let w = TestWorld::new(); + let h = w.start().await; + call( + &h, + "probe.write", + json!({"path":"/run/shared","text":"yes"}), + ) + .await; + assert_eq!( + call(&h, "probe.read", json!({"path":"/run/shared"})).await, + "yes" + ); + h.stop().await; + let h = w.start().await; + assert!( + h.filesystem() + .await + .unwrap() + .list("/run") + .await + .unwrap() + .is_empty() + ); + h.stop().await; +} + +/// PROTOCOL 6: no write from a killed process lands after the next holder acquires its lock. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_kill_drains_writes_before_lock_transfer() { + let w = TestWorld::new(); + let h = w.start().await; + let pid = call(&h, "probe.write_loop", json!({})).await; + let next = h.clone(); + let waiter = tokio::spawn(async move { call(&next, "probe.locked_write", json!({})).await }); + use sausage_vfs::{Layer, Open}; + let view = h.filesystem().await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(20), async { + loop { + let file = view + .open("/proc/core/locks.json", Open::default()) + .await + .unwrap(); + let locks: Value = serde_json::from_slice(&file.read(0, 65536).await.unwrap()).unwrap(); + if locks["/state/loop"]["waiting"] + .as_array() + .is_some_and(|queue| !queue.is_empty()) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + call(&h, "core.kill", json!({"pid":pid})).await; + waiter.await.unwrap(); + assert_eq!( + std::fs::read_to_string(w.0.join("state/loop")).unwrap(), + "new" + ); + h.stop().await; +} + +/// PROTOCOL 4: native processes appear in proc and locks.json identifies holders and waiters. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn filesystem_native_processes_and_lock_snapshot() { + use sausage_vfs::{Layer, Open}; + let w = TestWorld::new(); + let h = w.start().await; + h.register_native( + "native", + std::sync::Arc::new(|ctx, _, _| { + Box::pin(async move { + ctx.call("core.lock", json!({"path":"live"})).await.unwrap(); + ctx.notify("ready", json!({"pid":ctx.pid})); + std::future::pending().await + }) + }), + ); + let mut events = h.subscribe(); + let host = h.clone(); + let task = tokio::spawn(async move { host.call("native.wait", json!({})).await }); + let event = events.recv().await.unwrap(); + let pid = event["params"]["params"]["pid"].as_str().unwrap(); + let view = h.filesystem().await.unwrap(); + assert!( + view.list("/proc") + .await + .unwrap() + .iter() + .any(|e| e.name == pid) + ); + let info = call( + &h, + "probe.read", + json!({"path":format!("/proc/{pid}/info.json")}), + ) + .await; + assert_eq!( + serde_json::from_str::(info.as_str().unwrap()).unwrap()["version"], + "native" + ); + let host = h.clone(); + let waiter = tokio::spawn(async move { host.call("core.lock", json!({"path":"live"})).await }); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let file = view + .open("/proc/core/locks.json", Open::default()) + .await + .unwrap(); + let locks: Value = serde_json::from_slice(&file.read(0, 65536).await.unwrap()).unwrap(); + if locks["live"]["waiting"] == json!(["console"]) { + assert_eq!(locks["live"]["owner"], pid); + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + call(&h, "core.kill", json!({"pid":pid})).await; + assert_eq!(waiter.await.unwrap().unwrap(), json!({"held":true})); + assert!(task.await.unwrap().is_err()); + assert!(view.stat(&format!("/proc/{pid}"), true).await.is_err()); + h.stop().await; +} diff --git a/cli/tests/plugins/activation/Cargo.toml b/cli/tests/plugins/activation/Cargo.toml new file mode 100644 index 0000000..3dd20d1 --- /dev/null +++ b/cli/tests/plugins/activation/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "activation-guest" +version = "0.1.0" +edition = "2024" + +[workspace] + +[dependencies] +serde_json = "1" diff --git a/cli/tests/plugins/activation/src/main.rs b/cli/tests/plugins/activation/src/main.rs new file mode 100644 index 0000000..a514258 --- /dev/null +++ b/cli/tests/plugins/activation/src/main.rs @@ -0,0 +1,53 @@ +//! Raw discovery and lifecycle guest for activation protocol tests. +use serde_json::{Value, json}; +use std::io::{self, BufRead, Write}; +/// Emits one flushed JSON-RPC response. +fn send(value: Value) { + println!("{value}"); + io::stdout().flush().unwrap(); +} +/// Reads a version-specific discovery document without the SDK. +fn main() { + let version = env!("FIXTURE_VERSION"); + let plugin = std::env::var("SAUSAGE_PLUGIN").unwrap(); + let mut lines = io::stdin().lock().lines(); + let request: Value = serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap(); + let document: Value = + serde_json::from_slice(&std::fs::read(format!("/state/{plugin}/{version}.json")).unwrap()) + .unwrap(); + let result = match request["method"].as_str().unwrap() { + "rpc.discover" => { + if document["x-test-wait"] == true { + std::fs::write(format!("/state/{plugin}/discover-started"), "ready").unwrap(); + while !std::path::Path::new(&format!("/state/{plugin}/discover-release")).exists() { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + } + document + } + "read" => { + let text = + std::fs::read_to_string(request["params"]["path"].as_str().unwrap()).unwrap(); + serde_json::from_str(&text).unwrap_or(json!(text)) + } + "rpc.retired" => { + std::fs::write( + format!("/state/{plugin}/retired"), + request["params"]["previous"].as_str().unwrap(), + ) + .unwrap(); + json!({"retired":version}) + } + "serve" | "hold" => { + send(json!({"jsonrpc":"2.0","id":request["id"],"result":{"version":version}})); + if request["method"] == "serve" { + std::fs::write(format!("/state/{plugin}/service"), version).unwrap(); + } + for _ in lines {} + std::fs::write(format!("/state/{plugin}/ended-{version}"), "ended").unwrap(); + return; + } + _ => json!({"version":version}), + }; + send(json!({"jsonrpc":"2.0","id":request["id"],"result":result})); +} diff --git a/cli/tests/plugins/probe/Cargo.toml b/cli/tests/plugins/probe/Cargo.toml index b2499d8..0ad0294 100644 --- a/cli/tests/plugins/probe/Cargo.toml +++ b/cli/tests/plugins/probe/Cargo.toml @@ -5,3 +5,4 @@ edition = "2024" [workspace] [dependencies] serde_json = "1" +wasip2 = "1" diff --git a/cli/tests/plugins/probe/src/main.rs b/cli/tests/plugins/probe/src/main.rs index 5d7cf89..581f12e 100644 --- a/cli/tests/plugins/probe/src/main.rs +++ b/cli/tests/plugins/probe/src/main.rs @@ -14,6 +14,13 @@ fn receive(lines: &mut Lines) -> Value { fn echo(params: &Value) -> Value { params.clone() } +fn discover(params: &Value) -> Value { + if let Ok(bytes) = std::fs::read("/state/probe/discover.json") { + serde_json::from_slice(&bytes).unwrap() + } else { + params.clone() + } +} fn environment() -> Value { json!({ "pid": std::env::var("SAUSAGE_PID").unwrap(), @@ -155,10 +162,126 @@ fn outgoing_call(params: &Value, lines: &mut Lines) -> Value { fn lock(params: &Value, lines: &mut Lines) -> Value { outgoing_call(&json!({"method":"core.lock", "params":params}), lines) } +fn filesystem(params: &Value) -> Value { + use std::fs; + let pid = std::env::var("SAUSAGE_PID").unwrap(); + match params["topic"].as_str().unwrap() { + "root" => { + let preopens = wasip2::filesystem::preopens::get_directories(); + assert_eq!(preopens.len(), 1); + assert_eq!(preopens[0].1, "/"); + let names: Vec<_> = fs::read_dir("/") + .unwrap() + .map(|e| e.unwrap().file_name().into_string().unwrap()) + .collect(); + for name in ["proc", "self", "run"] { + assert!(names.iter().any(|n| n == name)); + } + fs::write("/state/root", "world").unwrap(); + fs::write("/run/shared", "shared").unwrap(); + assert!(fs::rename("/run/shared", "/state/moved").is_err()); + assert!(fs::rename("/state/root", "/self/moved").is_err()); + assert_eq!(fs::read_to_string("/run/../state/root").unwrap(), "world"); + } + "reserved" => { + for name in ["info.json", "log", "fs"] { + assert!(fs::write(format!("/self/{name}"), "bad").is_err()); + assert!(fs::create_dir(format!("/self/{name}")).is_err()); + fs::write("/self/temp", "ok").unwrap(); + assert!(fs::rename("/self/temp", format!("/self/{name}")).is_err()); + fs::remove_file("/self/temp").unwrap(); + } + } + "self" => { + fs::write("/self/endpoint.tmp", "endpoint").unwrap(); + fs::rename("/self/endpoint.tmp", "/self/endpoint").unwrap(); + assert_eq!( + fs::read_to_string(format!("/proc/{pid}/endpoint")).unwrap(), + "endpoint" + ); + assert!(fs::write(format!("/proc/{pid}/endpoint"), "bad").is_err()); + assert!(fs::remove_file(format!("/proc/{pid}/endpoint")).is_err()); + let info: Value = + serde_json::from_slice(&fs::read("/self/info.json").unwrap()).unwrap(); + assert_eq!(info["plugin"], "probe"); + assert_eq!(info["method"], "filesystem"); + return json!(pid); + } + "isolated" => { + assert!(!fs::exists("/self/endpoint").unwrap()); + } + "snapshot" => { + eprintln!("before snapshot"); + let mut file = fs::File::open("/self/log").unwrap(); + eprintln!("after snapshot"); + let mut contents = String::new(); + file.read_to_string(&mut contents).unwrap(); + assert!(contents.contains("start filesystem")); + assert!(!contents.contains("after snapshot")); + assert_eq!(file.read(&mut [0; 1]).unwrap(), 0); + } + "core" => { + assert!( + fs::read_to_string("/proc/core/PROTOCOL.md") + .unwrap() + .contains("## 4. The filesystem") + ); + for name in ["graph.json", "locks.json"] { + let _: Value = + serde_json::from_slice(&fs::read(format!("/proc/core/{name}")).unwrap()) + .unwrap(); + } + assert!( + fs::read_to_string("/proc/core/graph.dot") + .unwrap() + .contains("digraph") + ); + assert!(fs::read_dir("/proc/core/plugins").is_ok()); + assert!(fs::write("/proc/core/graph.json", "bad").is_err()); + } + _ => panic!("unknown filesystem topic"), + } + json!({}) +} +fn write_loop(request: &Value, lines: &mut Lines) { + assert_eq!( + lock(&json!({"path":"/state/loop"}), lines), + json!({"held":true}) + ); + std::fs::write("/state/loop", "old").unwrap(); + send( + json!({"jsonrpc":"2.0", "id":request["id"], "result":std::env::var("SAUSAGE_PID").unwrap()}), + ); + loop { + std::fs::write("/state/loop", "old").unwrap(); + } +} +fn locked_write(lines: &mut Lines) -> Value { + assert_eq!( + lock(&json!({"path":"/state/loop"}), lines), + json!({"held":true}) + ); + std::fs::write("/state/loop", "new").unwrap(); + std::thread::sleep(std::time::Duration::from_millis(100)); + assert_eq!(std::fs::read_to_string("/state/loop").unwrap(), "new"); + json!({}) +} fn dispatch(request: &Value, lines: &mut Lines) -> Option { let params = &request["params"]; let result = match request["method"].as_str().unwrap() { - "echo" | "rpc.discover" => echo(params), + "self_serve" => { + std::fs::write("/self/endpoint", "endpoint").unwrap(); + serve(request, lines); + return None; + } + "filesystem" => filesystem(params), + "write_loop" => { + write_loop(request, lines); + return None; + } + "locked_write" => locked_write(lines), + "echo" => echo(params), + "rpc.discover" => discover(params), "env" => environment(), "multicall" => multicall(lines), "notification" => notification(params), diff --git a/cli/tests/sdk.rs b/cli/tests/sdk.rs new file mode 100644 index 0000000..5d42baa --- /dev/null +++ b/cli/tests/sdk.rs @@ -0,0 +1,251 @@ +//! Exercise SDK plugins through the real host, without a fake core. +use sausage::{Host, world::World}; +use serde_json::{Value, json}; +use std::{path::PathBuf, sync::OnceLock, time::Duration}; + +fn fixtures() -> &'static PathBuf { + static TARGET: OnceLock = OnceLock::new(); + TARGET.get_or_init(|| { + let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .unwrap() + .to_owned(); + let target = root.join("target/sdk/real-guests"); + for package in [ + "examples/notes", + "examples/consumer", + "examples/www", + "examples/secrets", + "sdk/tests/plugins/probe", + "sdk/tests/plugins/consumer", + ] { + assert!( + std::process::Command::new("cargo") + .args([ + "build", + "--offline", + "--target", + "wasm32-wasip2", + "--manifest-path" + ]) + .arg(root.join(package).join("Cargo.toml")) + .env("CARGO_TARGET_DIR", &target) + .status() + .unwrap() + .success() + ); + } + target.join("wasm32-wasip2/debug") + }) +} +struct TestWorld(PathBuf); +impl TestWorld { + fn new() -> Self { + let path = std::env::temp_dir().join(format!("sausage-sdk-{}", uuid::Uuid::now_v7())); + let world = World::create(&path).unwrap(); + for (plugin, binary) in [ + ("notes", "notes"), + ("consumer", "consumer"), + ("www", "www"), + ("secrets", "secrets"), + ("probe", "sdk-probe"), + ("typed", "sdk-consumer"), + ] { + std::fs::create_dir_all(world.plugin_dir(plugin)).unwrap(); + std::fs::copy( + fixtures().join(format!("{binary}.wasm")), + world.plugin_dir(plugin).join("plugin.wasm"), + ) + .unwrap(); + } + Self(path) + } + async fn start(&self) -> Host { + Host::start_with_cache( + &self.0, + &PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../target/sdk/real-cache"), + ) + .await + .unwrap() + } +} +impl Drop for TestWorld { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} +async fn call(host: &Host, method: &str, params: Value) -> Value { + tokio::time::timeout(Duration::from_secs(60), host.call(method, params)) + .await + .unwrap() + .unwrap() +} + +/// Concurrent read-modify-write calls retain every note under the SDK lock. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn notes_lock_preserves_parallel_additions() { + let world = TestWorld::new(); + let host = world.start().await; + assert_eq!( + call(&host, "notes.add", json!({"text":"first"})).await["count"], + 1 + ); + let calls = (0..12) + .map(|n| { + let h = host.clone(); + tokio::spawn( + async move { call(&h, "notes.add", json!({"text":format!("note-{n}")})).await }, + ) + }) + .collect::>(); + for task in calls { + task.await.unwrap(); + } + let notes = call(&host, "notes.list", json!({})).await; + assert_eq!(notes.as_array().unwrap().len(), 13); + assert_eq!(notes[0], "first"); + for n in 0..12 { + assert!( + notes + .as_array() + .unwrap() + .contains(&json!(format!("note-{n}"))) + ); + } + host.stop().await; +} +/// The example consumer invokes notes through its generated import. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn generated_notes_import() { + let world = TestWorld::new(); + let host = world.start().await; + assert_eq!( + call(&host, "consumer.note", json!({"text":"typed"})).await, + 1 + ); + assert_eq!(call(&host, "notes.list", json!({})).await, json!(["typed"])); + host.stop().await; +} +/// Discovery retains the complete document and satisfies the consumer contract. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discovery_is_compatible() { + let world = TestWorld::new(); + let host = world.start().await; + let actual = call(&host, "notes.rpc.discover", json!({})).await; + let consumer = call(&host, "consumer.rpc.discover", json!({})).await; + let expected: Value = + serde_json::from_str(include_str!("../../examples/consumer/imports/notes.json")).unwrap(); + assert_eq!(actual, expected); + assert_eq!(actual["openrpc"], "1.2.6"); + assert!(actual["components"]["schemas"]["AddParams"].is_object()); + assert!( + actual["methods"] + .as_array() + .unwrap() + .iter() + .find(|m| m["name"] == "list") + .unwrap() + .get("x-sausage-params") + .is_none() + ); + assert!(actual["components"]["schemas"].get("Empty").is_none()); + assert!( + sausage_interface::compat::check(&consumer["x-sausage-imports"]["notes"], &actual) + .is_empty() + ); + host.stop().await; +} +/// Detached SDK calls receive progress and forward it through core.notify. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "core: stdout response can overtake a preceding core.notify"] +async fn detached_call_notifications() { + let world = TestWorld::new(); + let host = world.start().await; + let mut events = host.subscribe(); + assert_eq!(call(&host, "probe.relay", json!({})).await, 42); + let event = tokio::time::timeout(Duration::from_secs(5), events.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(event["params"]["name"], "step"); + host.stop().await; +} +/// Detached typed calls complete and detached services survive their original caller. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn detached_service_survives_parent() { + let world = TestWorld::new(); + let host = world.start().await; + assert_eq!(call(&host, "typed.run", json!({})).await, 5); + let child = call(&host, "probe.launch", json!({})).await; + tokio::time::timeout(Duration::from_secs(5), async { + while host.processes().values().any(|p| p.method == "launch") { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(host.processes().contains_key(child.as_str().unwrap())); + host.stop().await; +} +/// A core.stop request closes stdin and executes the SDK stop hook. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn stop_runs_hook() { + let world = TestWorld::new(); + let host = world.start().await; + let pid = call(&host, "probe.service", json!({})).await; + call(&host, "core.stop", json!({"pid":pid})).await; + assert_eq!(std::fs::read(world.0.join("stopped")).unwrap(), b"yes"); + 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() { + let world = TestWorld::new(); + let host = world.start().await; + let text = "z".repeat(192 * 1024); + assert_eq!(call(&host, "probe.echo", json!({"text":text})).await, text); + host.stop().await; +} +/// Normal SDK stdout never produces the discarded-output journal warning. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn no_discarded_stdout() { + let world = TestWorld::new(); + let host = world.start().await; + call(&host, "probe.echo", json!({"text":"framing"})).await; + host.stop().await; + assert_eq!( + call(&host, "core.log", json!({"grep":"discarded stdout"})).await, + json!([]) + ); +} +/// Two SDK processes exchange endpoint requests and progress at a world path. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn world_endpoint() { + let world = TestWorld::new(); + let host = world.start().await; + call(&host, "probe.service", json!({})).await; + assert_eq!(call(&host, "typed.world_endpoint", json!({})).await, 7); + host.stop().await; +} +/// The HTTP example exposes its self endpoint through the process namespace. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "needs /proc, /self, and /run (T6)"] +async fn www_control_through_proc() { + let world = TestWorld::new(); + let host = world.start().await; + assert_eq!( + call(&host, "consumer.demo", json!({"text":"web","port":0})).await, + 0 + ); + host.stop().await; +} +/// The vault example publishes a fixed endpoint in the shared run namespace. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "needs /proc, /self, and /run (T6)"] +async fn secrets_fixed_endpoint() { + let world = TestWorld::new(); + let host = world.start().await; + call(&host, "secrets.serve", json!({})).await; + assert_eq!(call(&host, "consumer.vault", json!({})).await, "hello"); + host.stop().await; +}