diff --git a/cli/src/cargo.rs b/cli/src/cargo.rs index e3f93ae..e25645b 100644 --- a/cli/src/cargo.rs +++ b/cli/src/cargo.rs @@ -11,7 +11,9 @@ use include_dir::{Dir, include_dir}; use crate::world::{self, World}; +/// The name reserved for the native Rust build plugin. pub const ID: &str = "cargo"; +/// The WASI component target used for Rust plugin builds. pub const TARGET: &str = "wasm32-wasip2"; /// The source of the `sausage-actor` SDK. @@ -19,23 +21,34 @@ static SDK_SRC: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk/src"); static SDK_MACROS: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk/macros/src"); #[derive(Clone, Default, Debug)] +/// The totals and output from a source tree's test run. pub struct TestResults { + /// The number of tests that passed. pub passed: u64, + /// The number of tests that failed. pub failed: u64, + /// The test output retained for the caller. pub output: String, } +/// The component artifact and diagnostics from a successful build. pub struct BuildOutput { /// Host path of the published `.wasm` file. pub wasm: PathBuf, + /// The elapsed build time in seconds. pub seconds: f64, + /// The compiler diagnostics without progress lines. pub warnings: String, + /// The totals and output of the test phase. pub tests: TestResults, } #[derive(Debug)] +/// The failed build or test phase and its diagnostics. pub struct BuildFailure { + /// The reason that the build could not complete. pub reason: String, + /// The totals and output of the test phase. pub tests: TestResults, } diff --git a/cli/src/components.rs b/cli/src/components.rs new file mode 100644 index 0000000..93a79f1 --- /dev/null +++ b/cli/src/components.rs @@ -0,0 +1,86 @@ +//! Component versions cached by file metadata, with independent compilation per hash. +use crate::{error, runtime::Runtime}; +use sha2::{Digest, Sha256}; +use std::{ + collections::HashMap, + path::PathBuf, + sync::{Arc, Mutex}, + time::SystemTime, +}; +use tokio::sync::{Mutex as AsyncMutex, OnceCell}; +use wasmtime::component::Component; + +struct Version { + size: u64, + modified: SystemTime, + hash: String, + component: Component, +} +type PathEntry = Arc>>; +type CompiledEntry = Arc>; + +#[derive(Default)] +pub(crate) struct Components { + paths: Mutex>, + compiled: Mutex>, +} +impl Components { + /// Stat each path, but read and hash its bytes only after its metadata changes. + pub(crate) async fn load( + &self, + path: PathBuf, + runtime: Arc, + ) -> Result<(Component, String), serde_json::Value> { + let entry = self + .paths + .lock() + .unwrap() + .entry(path.clone()) + .or_default() + .clone(); + let mut cached = entry.lock().await; + let metadata = tokio::fs::metadata(&path) + .await + .map_err(|_| error(-32001, "no such plugin"))?; + let modified = metadata + .modified() + .map_err(|e| error(-32603, &e.to_string()))?; + if let Some(version) = cached.as_ref() + && version.size == metadata.len() + && version.modified == modified + { + return Ok((version.component.clone(), version.hash.clone())); + } + let (bytes, hash) = tokio::task::spawn_blocking(move || -> std::io::Result<_> { + let bytes = std::fs::read(path)?; + let hash = format!("{:x}", Sha256::digest(&bytes)); + Ok((bytes, hash)) + }) + .await + .map_err(|e| error(-32603, &e.to_string()))? + .map_err(|e| error(-32603, &e.to_string()))?; + let compiled = self + .compiled + .lock() + .unwrap() + .entry(hash.clone()) + .or_default() + .clone(); + let component = compiled + .get_or_try_init(|| async move { + tokio::task::spawn_blocking(move || runtime.compile(&bytes)) + .await + .map_err(|e| error(-32603, &e.to_string()))? + .map_err(|e| error(-32603, &e.to_string())) + }) + .await? + .clone(); + *cached = Some(Version { + size: metadata.len(), + modified, + hash: hash.clone(), + component: component.clone(), + }); + Ok((component, hash)) + } +} diff --git a/cli/src/console.rs b/cli/src/console.rs new file mode 100644 index 0000000..cba57eb --- /dev/null +++ b/cli/src/console.rs @@ -0,0 +1,24 @@ +//! The minimal console reads explicit JSON-RPC calls and the quit command. +use crate::Host; +use tokio::io::{AsyncBufReadExt, BufReader}; + +/// Read console lines until EOF or quit, and print each call's result. +pub async fn run_input(host: Host) -> anyhow::Result<()> { + let mut lines = BufReader::new(tokio::io::stdin()).lines(); + while let Some(line) = lines.next_line().await? { + if line == "quit" { + break; + } + let (method, params) = line.split_once(' ').unwrap_or((&line, "{}")); + match serde_json::from_str(params) { + Ok(params) => { + let value = match host.call(method, params).await { + Ok(value) | Err(value) => value, + }; + println!("{value}"); + } + Err(error) => eprintln!("{error}"), + } + } + Ok(()) +} diff --git a/cli/src/core.rs b/cli/src/core.rs new file mode 100644 index 0000000..6539650 --- /dev/null +++ b/cli/src/core.rs @@ -0,0 +1,64 @@ +//! The native core methods and their JSON parameters. +use crate::{Host, RpcResult, error}; +use serde_json::{Value, json}; + +impl Host { + /// Execute a core method without starting a guest process. + pub(crate) async fn core( + &self, + caller: &str, + id: Option, + method: &str, + p: Value, + ) -> RpcResult { + match method { + "call" => { + self.route( + caller.into(), + id, + required_str(&p, "method")?.into(), + p.get("params").cloned().unwrap_or(json!({})), + p["detach"].as_bool().unwrap_or(false), + ) + .await + } + "notify" => { + self.notify( + caller, + required_str(&p, "name")?, + p.get("params").cloned().unwrap_or(json!({})), + ); + Ok(Value::Null) + } + "kill" => { + self.kill(required_str(&p, "pid")?); + Ok(json!({})) + } + "stop" => { + if let Some(mut n) = self.begin_stop(required_str(&p, "pid")?) { + let _ = n.wait_for(|ended| *ended).await; + } + Ok(json!({})) + } + "lock" => { + self.lock(caller, required_str(&p, "path")?, p["wait"] != false) + .await + } + "unlock" => { + self.unlock(caller, required_str(&p, "path")?); + Ok(Value::Null) + } + "log" => Ok(self.0.journal.query(p).await), + "exit" => { + self.0.exit.send_replace(true); + Ok(json!({})) + } + _ => Err(error(-32601, "method not found")), + } + } +} +fn required_str<'a>(params: &'a Value, name: &str) -> Result<&'a str, Value> { + params[name] + .as_str() + .ok_or_else(|| error(-32602, &format!("{name} must be a string"))) +} diff --git a/cli/src/execution.rs b/cli/src/execution.rs new file mode 100644 index 0000000..f5af1e0 --- /dev/null +++ b/cli/src/execution.rs @@ -0,0 +1,134 @@ +//! Execution of native futures and WASI processes, including transport drainage. +use crate::{ + Host, NativeContext, NativePlugin, + process::{ProcessGuard, ProcessInfo}, + runtime::{Instance, InstanceConfig}, +}; +use serde_json::{Value, json}; +use tokio::sync::mpsc; + +impl Host { + /// Run one process and retain its cleanup guard until all transport reads finish. + pub(crate) async fn run_process( + &self, + pid: String, + info: ProcessInfo, + component: Option, + native: Option, + incoming: mpsc::UnboundedReceiver, + has_id: bool, + ) { + let _guard = ProcessGuard { + host: self.clone(), + pid: pid.clone(), + }; + if let Some(native) = native { + let context = NativeContext { + host: self.clone(), + pid: pid.clone(), + }; + let result = native(context, info.method, info.params).await; + self.answer(&pid, result); + self.finish(&pid, "exit 0"); + return; + } + let result = self + .run_guest(&pid, &info, &component.unwrap(), incoming, has_id) + .await; + self.finish(&pid, &result.unwrap_or_else(|error| error.to_string())); + } + + /// Drain stdout before deciding whether the guest returned an answer. + async fn run_guest( + &self, + pid: &str, + info: &ProcessInfo, + component: &wasmtime::component::Component, + mut incoming: mpsc::UnboundedReceiver, + has_id: bool, + ) -> anyhow::Result { + let config = InstanceConfig { + world: self.0.world.root.clone(), + network: true, + env: vec![ + ("SAUSAGE_PID".into(), pid.into()), + ("SAUSAGE_PLUGIN".into(), info.plugin.clone()), + ( + "SAUSAGE_PARENT".into(), + info.parent.clone().unwrap_or_else(|| "core".into()), + ), + ], + }; + let mut instance = self.0.runtime.start(component, config).await?; + let mut initial = json!({ + "jsonrpc": "2.0", + "method": info.method, + "params": info.params, + }); + if has_id { + initial["id"] = json!(pid); + } + self.write_input(pid, &mut instance, initial).await?; + let mut stdin_open = true; + let mut stderr_open = true; + loop { + tokio::select! { + line = instance.stderr.recv(), if stderr_open => { + stderr_open = self.stderr_line(pid,&info.plugin,line); + } + line = instance.stdout.recv() => { + if !self.stdout_line(pid,line) { break; } + } + value = incoming.recv(), if stdin_open => { + stdin_open = self.forward_input(pid,&mut instance,value).await?; + } + } + } + while let Some(line) = instance.stderr.recv().await { + self.0.journal.write(pid, &info.plugin, "stderr", &line); + } + Ok(instance.wait().await.to_string()) + } + + /// Close stdin only after the queued responses reach the guest. + async fn forward_input( + &self, + pid: &str, + instance: &mut Instance, + value: Option, + ) -> std::io::Result { + if let Some(value) = value { + self.write_input(pid, instance, value).await?; + Ok(true) + } else { + instance.stdin.take(); + Ok(false) + } + } + async fn write_input( + &self, + pid: &str, + instance: &mut Instance, + value: Value, + ) -> std::io::Result<()> { + instance + .send_line(&value.to_string(), |line| self.output(pid, line)) + .await + } + fn stdout_line(&self, pid: &str, line: Option) -> bool { + if let Some(line) = line { + self.output(pid, line); + true + } else { + false + } + } + fn stderr_line(&self, pid: &str, plugin: &str, line: Option) -> bool { + if let Some(line) = line { + self.0.journal.write(pid, plugin, "stderr", &line); + true + } else { + false + } + } +} diff --git a/cli/src/host.rs b/cli/src/host.rs new file mode 100644 index 0000000..10edf61 --- /dev/null +++ b/cli/src/host.rs @@ -0,0 +1,131 @@ +//! Host startup, boot calls, and the public library API. +use crate::{RpcResult, components, journal, runtime::Runtime, world::World}; +use crate::{ + locks::Lock, + native::NativePlugin, + process::{Process, ProcessInfo}, +}; +use serde_json::{Value, json}; +use std::{ + collections::HashMap, + path::Path, + sync::{Arc, Mutex}, +}; +use tokio::sync::watch; +use uuid::Uuid; +pub(crate) struct State { + pub(crate) processes: HashMap, + pub(crate) locks: HashMap, + pub(crate) native: HashMap, +} +pub(crate) struct Inner { + pub(crate) world: World, + pub(crate) runtime: Arc, + pub(crate) state: Mutex, + pub(crate) journal: journal::Journal, + pub(crate) notifications: tokio::sync::broadcast::Sender, + pub(crate) exit: watch::Sender, + pub(crate) components: components::Components, +} +#[derive(Clone)] +/// A cloneable handle to a world and its running processes. +pub struct Host(pub(crate) Arc); + +impl Host { + /// Start a world with the default compile cache and its ordered boot calls. + pub async fn start(path: &Path) -> anyhow::Result { + let base = std::env::var_os("XDG_CACHE_HOME") + .map(std::path::PathBuf::from) + .unwrap_or_else(|| { + std::path::PathBuf::from(std::env::var_os("HOME").unwrap_or_default()) + .join(".cache") + }); + Self::start_with_cache(path, &base.join("sausage")).await + } + /// Start a world with an explicit compile cache folder. + pub async fn start_with_cache(path: &Path, cache: &Path) -> anyhow::Result { + let path = path.to_path_buf(); + let cache = cache.to_path_buf(); + let (world, runtime) = tokio::task::spawn_blocking(move || -> anyhow::Result<_> { + Ok((World::open(&path)?, Runtime::new(&cache)?)) + }) + .await??; + let journal = + journal::Journal::open(world.state_dir("core").join("journal.sqlite")).await?; + let host = Self(Arc::new(Inner { + world, + runtime, + journal, + state: Mutex::new(State { + processes: HashMap::new(), + locks: HashMap::new(), + native: HashMap::new(), + }), + notifications: tokio::sync::broadcast::channel(1024).0, + exit: watch::channel(false).0, + components: components::Components::default(), + })); + let boot: Vec = serde_json::from_slice( + &tokio::fs::read(host.0.world.root.join("config/boot.json")).await?, + )?; + for service in boot { + let result = host + .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 { + host.0.journal.write("core", "core", "core", &e.to_string()); + } + } + Ok(host) + } + /// Register a native plugin for subsequent calls. + pub fn register_native(&self, name: &str, plugin: NativePlugin) { + self.0 + .state + .lock() + .unwrap() + .native + .insert(name.into(), plugin); + } + /// Subscribe to notifications addressed to the console. + pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver { + self.0.notifications.subscribe() + } + /// Snapshot the metadata of each process that has not ended. + pub fn processes(&self) -> HashMap { + self.0 + .state + .lock() + .unwrap() + .processes + .iter() + .map(|(p, s)| (p.clone(), s.info.clone())) + .collect() + } + /// Make a console-side call and wait for its JSON result. + pub async fn call(&self, method: &str, params: Value) -> RpcResult { + self.route( + "console".into(), + Some(json!(Uuid::now_v7().to_string())), + method.into(), + params, + false, + ) + .await + } + /// Wait until a core.exit handler requests a restart. + pub async fn exit_requested(&self) { + let mut exit = self.0.exit.subscribe(); + let _ = exit.wait_for(|requested| *requested).await; + } + /// Return whether a core.exit handler requested a restart. + pub fn restart_requested(&self) -> bool { + *self.0.exit.borrow() + } +} diff --git a/cli/src/journal.rs b/cli/src/journal.rs index 3704d00..1223bfa 100644 --- a/cli/src/journal.rs +++ b/cli/src/journal.rs @@ -1,42 +1,158 @@ -//! SQLite journal entries and filters. +//! A dedicated SQLite writer batches entries and serves ordered journal queries. use rusqlite::{Connection, params}; use serde_json::{Value, json}; -use std::{collections::HashSet, path::Path, sync::Mutex}; -pub struct Journal { - db: Mutex, +use std::{ + collections::HashSet, + path::PathBuf, + sync::{Mutex, mpsc}, +}; +use tokio::sync::oneshot; + +struct Entry { + id: String, + pid: String, + plugin: String, + stream: String, + line: String, +} +enum Command { + Write(Entry), + Query(Value, oneshot::Sender), +} + +pub(crate) struct Journal { + sender: mpsc::Sender, warned: Mutex>, } impl Journal { - pub fn open(path: &Path) -> anyhow::Result { - std::fs::create_dir_all(path.parent().unwrap())?; - let db = Connection::open(path)?; - db.execute_batch("CREATE TABLE IF NOT EXISTS journal (id TEXT PRIMARY KEY, pid TEXT, plugin TEXT, stream TEXT, line TEXT)")?; + /// Open SQLite on its owning thread and wait for its schema to become ready. + pub(crate) async fn open(path: PathBuf) -> anyhow::Result { + let (sender, receiver) = mpsc::channel(); + let (ready, initialized) = oneshot::channel(); + std::thread::spawn(move || { + let opened = open_database(path); + match opened { + Ok(mut db) => { + let _ = ready.send(Ok(())); + while let Ok(first) = receiver.recv() { + let mut batch = vec![first]; + batch.extend(receiver.try_iter().take(255)); + write_batch(&mut db, batch); + } + } + Err(error) => { + let _ = ready.send(Err(error)); + } + } + }); + initialized.await??; Ok(Self { - db: Mutex::new(db), + sender, warned: Mutex::new(HashSet::new()), }) } - pub fn write(&self, pid: &str, plugin: &str, stream: &str, line: &str) { - self.db - .lock() - .unwrap() - .execute( - "INSERT INTO journal VALUES (?1,?2,?3,?4,?5)", - params![uuid::Uuid::now_v7().to_string(), pid, plugin, stream, line], - ) - .expect("write journal"); + /// Queue an entry without file operations or SQLite work on the caller's thread. + pub(crate) fn write(&self, pid: &str, plugin: &str, stream: &str, line: &str) { + let entry = Entry { + id: uuid::Uuid::now_v7().to_string(), + pid: pid.into(), + plugin: plugin.into(), + stream: stream.into(), + line: line.into(), + }; + self.sender + .send(Command::Write(entry)) + .expect("journal writer stopped"); } - pub fn warning(&self, pid: &str, plugin: &str) { + /// Queue one warning for the first non-blank discarded line of a process. + pub(crate) fn warning(&self, pid: &str, plugin: &str) { if self.warned.lock().unwrap().insert(pid.into()) { self.write(pid, plugin, "core", "warning: discarded stdout"); } } - pub fn query(&self, p: &Value) -> Value { - let db = self.db.lock().unwrap(); - let mut stmt=db.prepare("SELECT id,pid,plugin,stream,line FROM journal WHERE (?1 IS NULL OR plugin=?1) AND (?2 IS NULL OR pid=?2) AND (?3 IS NULL OR id>=?3) AND (?4 IS NULL OR id<=?4) AND (?5 IS NULL OR instr(line,?5)>0) ORDER BY id DESC LIMIT ?6").unwrap(); - let rows=stmt.query_map(params![p["plugin"].as_str(),p["pid"].as_str(),p["since"].as_str(),p["until"].as_str(),p["grep"].as_str(),p["limit"].as_i64().unwrap_or(100)],|r|Ok(json!({"id":r.get::<_,String>(0)?,"pid":r.get::<_,String>(1)?,"plugin":r.get::<_,String>(2)?,"stream":r.get::<_,String>(3)?,"line":r.get::<_,String>(4)?}))).unwrap(); - let mut entries: Vec<_> = rows.map(Result::unwrap).collect(); - entries.reverse(); - json!(entries) + /// Queue a query after all entries that the caller already submitted. + pub(crate) fn query_later(&self, filters: Value) -> oneshot::Receiver { + let (tx, rx) = oneshot::channel(); + self.sender + .send(Command::Query(filters, tx)) + .expect("journal writer stopped"); + rx + } + /// Wait asynchronously for the journal thread to execute a query. + pub(crate) async fn query(&self, filters: Value) -> Value { + self.query_later(filters) + .await + .expect("journal writer stopped") } } +fn open_database(path: PathBuf) -> anyhow::Result { + std::fs::create_dir_all(path.parent().unwrap())?; + let db = Connection::open(path)?; + db.execute_batch( + "PRAGMA journal_mode=WAL; + PRAGMA synchronous=NORMAL; + CREATE TABLE IF NOT EXISTS journal ( + id TEXT PRIMARY KEY, time TEXT, pid TEXT, plugin TEXT, stream TEXT, line TEXT + ); + CREATE INDEX IF NOT EXISTS journal_pid ON journal(pid); + CREATE INDEX IF NOT EXISTS journal_plugin ON journal(plugin);", + )?; + Ok(db) +} +/// Commit queued writes together before answering the batch's queries. +fn write_batch(db: &mut Connection, batch: Vec) { + let transaction = db.transaction().expect("journal transaction"); + for command in &batch { + if let Command::Write(entry) = command { + transaction + .execute( + "INSERT INTO journal (id,time,pid,plugin,stream,line) + VALUES (?1,strftime('%Y-%m-%dT%H:%M:%fZ','now'),?2,?3,?4,?5)", + params![entry.id, entry.pid, entry.plugin, entry.stream, entry.line], + ) + .expect("write journal"); + } + } + transaction.commit().expect("commit journal"); + for command in batch { + if let Command::Query(filters, response) = command { + let _ = response.send(query_database(db, &filters)); + } + } +} +fn query_database(db: &Connection, filters: &Value) -> Value { + let mut statement = db + .prepare( + "SELECT id,time,pid,plugin,stream,line FROM journal + WHERE (?1 IS NULL OR plugin=?1) AND (?2 IS NULL OR pid=?2) + AND (?3 IS NULL OR id>=?3) AND (?4 IS NULL OR id<=?4) + AND (?5 IS NULL OR instr(line,?5)>0) + ORDER BY id DESC LIMIT ?6", + ) + .expect("prepare journal query"); + let rows = statement + .query_map( + params![ + filters["plugin"].as_str(), + filters["pid"].as_str(), + filters["since"].as_str(), + filters["until"].as_str(), + filters["grep"].as_str(), + filters["limit"].as_i64().unwrap_or(100) + ], + |row| { + Ok(json!({ + "id":row.get::<_,String>(0)?, + "time":row.get::<_,String>(1)?, + "pid":row.get::<_,String>(2)?, + "plugin":row.get::<_,String>(3)?, + "stream":row.get::<_,String>(4)?, + "line":row.get::<_,String>(5)?, + })) + }, + ) + .expect("query journal"); + let mut entries: Vec<_> = rows.map(Result::unwrap).collect(); + entries.reverse(); + json!(entries) +} diff --git a/cli/src/lib.rs b/cli/src/lib.rs index fe82f87..4a4d4f5 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -1,636 +1,23 @@ -//! Process ownership, JSON-RPC routing, native plugins, and host lifetime. +//! A host for worlds of JSON-RPC plugins. pub mod cargo; +mod components; +pub mod console; +mod core; +mod execution; +mod host; mod journal; +mod locks; +mod native; +mod process; +mod routing; pub mod runtime; pub mod world; -use runtime::{InstanceConfig, Runtime}; +pub use host::Host; +pub use native::{NativeContext, NativeFuture, NativePlugin}; +pub use process::ProcessInfo; use serde_json::{Value, json}; -use sha2::{Digest, Sha256}; -use std::{ - collections::{HashMap, VecDeque}, - future::Future, - path::Path, - pin::Pin, - sync::{Arc, Mutex}, -}; -use tokio::sync::{mpsc, oneshot, watch}; -use uuid::Uuid; -use world::World; +/// A JSON result or a JSON-RPC error object. pub type RpcResult = Result; -pub type NativeFuture = Pin + Send>>; -pub type NativePlugin = Arc NativeFuture + Send + Sync>; -#[derive(Clone)] -pub struct NativeContext { - pub host: Host, - pub pid: String, -} -impl NativeContext { - pub async fn call(&self, method: &str, params: Value) -> RpcResult { - let exit = requests_exit(method, ¶ms); - let result = self - .host - .route( - self.pid.clone(), - Some(json!(Uuid::now_v7().to_string())), - method.into(), - params, - false, - ) - .await; - if exit && result.is_ok() { - self.host.0.exit.send_replace(true); - } - result - } - pub fn notify(&self, name: &str, params: Value) { - self.host.notify(&self.pid, name, params); - } -} -#[derive(Clone, serde::Serialize)] -pub struct ProcessInfo { - pub plugin: String, - pub version: String, - pub method: String, - pub params: Value, - pub parent: Option, - pub detached: bool, - pub start_time: u128, - pub state: String, -} -struct Process { - info: ProcessInfo, - input: Option>, - abort: Option, - caller: String, - id: Option, - answer: Option>, - ended: watch::Sender, -} -struct Lock { - owner: String, - queue: VecDeque<(String, oneshot::Sender)>, -} -struct State { - processes: HashMap, - locks: HashMap, - components: HashMap, - native: HashMap, -} -struct Inner { - world: World, - runtime: Arc, - state: Mutex, - journal: journal::Journal, - notifications: tokio::sync::broadcast::Sender, - exit: watch::Sender, - compilation: tokio::sync::Mutex<()>, -} -#[derive(Clone)] -pub struct Host(Arc); -struct ProcessGuard { - host: Host, - pid: String, -} -impl Drop for ProcessGuard { - fn drop(&mut self) { - self.host.finish(&self.pid, "process task ended"); - } -} fn error(code: i64, text: &str) -> Value { json!({"code":code,"message":text}) } -impl Host { - pub async fn start(path: &Path) -> anyhow::Result { - let base = std::env::var_os("XDG_CACHE_HOME") - .map(std::path::PathBuf::from) - .unwrap_or_else(|| { - std::path::PathBuf::from(std::env::var_os("HOME").unwrap_or_default()) - .join(".cache") - }); - Self::start_with_cache(path, &base.join("sausage")).await - } - pub async fn start_with_cache(path: &Path, cache: &Path) -> anyhow::Result { - let world = World::open(path)?; - let journal = journal::Journal::open(&world.state_dir("core").join("journal.sqlite"))?; - let host = Self(Arc::new(Inner { - world, - runtime: Runtime::new(cache)?, - journal, - state: Mutex::new(State { - processes: HashMap::new(), - locks: HashMap::new(), - components: HashMap::new(), - native: HashMap::new(), - }), - notifications: tokio::sync::broadcast::channel(1024).0, - exit: watch::channel(false).0, - compilation: tokio::sync::Mutex::new(()), - })); - let boot: Vec = - serde_json::from_slice(&std::fs::read(host.0.world.root.join("config/boot.json"))?)?; - for service in boot { - let result = host - .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 { - host.0.journal.write("core", "core", "core", &e.to_string()); - } - } - Ok(host) - } - pub fn register_native(&self, name: &str, plugin: NativePlugin) { - self.0 - .state - .lock() - .unwrap() - .native - .insert(name.into(), plugin); - } - pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver { - self.0.notifications.subscribe() - } - pub fn processes(&self) -> HashMap { - self.0 - .state - .lock() - .unwrap() - .processes - .iter() - .map(|(p, s)| (p.clone(), s.info.clone())) - .collect() - } - pub async fn call(&self, method: &str, params: Value) -> RpcResult { - let exit = requests_exit(method, ¶ms); - let result = self - .route( - "console".into(), - Some(json!(Uuid::now_v7().to_string())), - method.into(), - params, - false, - ) - .await; - if exit && result.is_ok() { - self.0.exit.send_replace(true); - } - result - } - pub async fn exit_requested(&self) { - let mut exit = self.0.exit.subscribe(); - let _ = exit.wait_for(|requested| *requested).await; - } - pub fn restart_requested(&self) -> bool { - *self.0.exit.borrow() - } - fn send(&self, pid: &str, value: Value) { - if pid == "console" { - let _ = self.0.notifications.send(value); - } else if let Some(tx) = self - .0 - .state - .lock() - .unwrap() - .processes - .get(pid) - .and_then(|p| p.input.as_ref()) - { - let _ = tx.send(value); - } - } - fn notify(&self, pid: &str, name: &str, params: Value) { - let target = self - .0 - .state - .lock() - .unwrap() - .processes - .get(pid) - .and_then(|p| p.answer.as_ref().map(|_| (p.caller.clone(), p.id.clone()))); - if let Some((caller, id)) = target { - self.send(&caller,json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":id,"name":name,"params":params}})); - } - } - fn route( - &self, - caller: String, - id: Option, - method: String, - params: Value, - detached: bool, - ) -> NativeFuture { - let host = self.clone(); - Box::pin(async move { - if !params.is_object() { - return Err(error(-32602, "params must be an object")); - } - if caller != "console" && caller != "core" { - let state = host.0.state.lock().unwrap(); - let Some(process) = state.processes.get(&caller) else { - return Err(error(-32600, "process ended")); - }; - if process.info.state != "running" && method != "core.notify" { - return Err(error(-32600, "process is stopping")); - } - } - let (plugin, method) = method - .split_once('.') - .ok_or_else(|| error(-32601, "method not found"))?; - if plugin == "core" { - return host.core(&caller, id, method, params).await; - } - if plugin.is_empty() - || !plugin - .bytes() - .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_') - { - return Err(error(-32601, "invalid plugin name")); - } - let native = host.0.state.lock().unwrap().native.get(plugin).cloned(); - let (component, version) = if native.is_some() { - (None, "native".into()) - } else { - let _compilation = host.0.compilation.lock().await; - let bytes = std::fs::read(host.0.world.plugin_dir(plugin).join("plugin.wasm")) - .map_err(|_| error(-32001, "no such plugin"))?; - let hash = format!("{:x}", Sha256::digest(&bytes)); - let cached = host.0.state.lock().unwrap().components.get(&hash).cloned(); - let component = match cached { - Some(c) => c, - None => { - let rt = host.0.runtime.clone(); - let c = tokio::task::spawn_blocking(move || rt.compile(&bytes)) - .await - .map_err(|e| error(-32603, &e.to_string()))? - .map_err(|e| error(-32603, &e.to_string()))?; - host.0 - .state - .lock() - .unwrap() - .components - .insert(hash.clone(), c.clone()); - c - } - }; - (Some(component), hash) - }; - let pid = format!("{plugin}-{}", Uuid::now_v7()); - let (tx, rx) = oneshot::channel(); - let (input, mut incoming) = mpsc::unbounded_channel::(); - let info = ProcessInfo { - plugin: plugin.into(), - version, - method: method.into(), - params: params.clone(), - parent: if detached { None } else { Some(caller.clone()) }, - detached, - start_time: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis(), - state: "running".into(), - }; - let parent_name = info.parent.clone().unwrap_or_else(|| "core".into()); - { - let mut state = host.0.state.lock().unwrap(); - if caller != "console" - && caller != "core" - && !state - .processes - .get(&caller) - .is_some_and(|p| p.info.state == "running") - { - return Err(error(-32600, "caller stopped or ended")); - } - host.0 - .journal - .write(&pid, plugin, "core", &format!("start {method} {params}")); - state.processes.insert( - pid.clone(), - Process { - info, - input: Some(input), - abort: None, - caller, - id: id.clone(), - answer: Some(tx), - ended: watch::channel(false).0, - }, - ); - } - let plugin_name = plugin.to_string(); - let h = host.clone(); - let p = pid.clone(); - let method = method.to_string(); - let has_id = id.is_some(); - let task = tokio::spawn(async move { - let _guard = ProcessGuard { - host: h.clone(), - pid: p.clone(), - }; - if let Some(native) = native { - let result = native( - NativeContext { - host: h.clone(), - pid: p.clone(), - }, - method, - params, - ) - .await; - h.answer(&p, result); - h.finish(&p, "exit 0"); - return; - } - let cfg = InstanceConfig { - world: h.0.world.root.clone(), - network: true, - env: vec![ - ("SAUSAGE_PID".into(), p.clone()), - ("SAUSAGE_PLUGIN".into(), plugin_name.clone()), - ("SAUSAGE_PARENT".into(), parent_name), - ], - }; - let result = async { - let mut instance = h.0.runtime.start(&component.unwrap(),cfg)?; - let initial = if has_id { json!({"jsonrpc":"2.0","id":p,"method":method,"params":params}) } else {json!({"jsonrpc":"2.0","method":method,"params":params})}; - instance.send_line(&initial.to_string(),|line|h.output(&p,line)).await?; - let mut stdin_open = true; - let mut stderr_open = true; - loop { tokio::select! { - line = instance.stderr.recv(), if stderr_open => match line { Some(line) => h.0.journal.write(&p,&plugin_name,"stderr",&line), None => stderr_open=false }, - line = instance.stdout.recv() => match line { Some(line) => h.output(&p,line), None => break }, - value = incoming.recv(), if stdin_open => match value { Some(v) => { instance.send_line(&v.to_string(),|line|h.output(&p,line)).await?; }, None => { instance.stdin.take(); stdin_open=false; } } - }} - while let Some(line) = instance.stderr.recv().await { h.0.journal.write(&p,&plugin_name,"stderr",&line); } - Ok::<_,anyhow::Error>(instance.wait().await.to_string()) - }.await; - h.finish(&p, &result.unwrap_or_else(|e| e.to_string())); - }); - if let Some(p) = host.0.state.lock().unwrap().processes.get_mut(&pid) { - p.abort = Some(task.abort_handle()); - } else { - task.abort(); - } - if id.is_none() { - return Ok(Value::Null); - } - rx.await - .unwrap_or_else(|_| Err(error(-32002, "process ended"))) - }) - } - fn answer(&self, pid: &str, result: RpcResult) { - if let Some(tx) = self - .0 - .state - .lock() - .unwrap() - .processes - .get_mut(pid) - .and_then(|p| p.answer.take()) - { - let _ = tx.send(result); - } - } - fn output(&self, pid: &str, line: String) { - let Ok(v) = serde_json::from_str::(&line) else { - self.discard(pid); - return; - }; - if !v.is_object() || v["jsonrpc"] != "2.0" { - self.discard(pid); - return; - } - if let Some(method) = v["method"].as_str() { - let h = self.clone(); - let caller = pid.to_string(); - let exit = requests_exit(method, &v["params"]); - let method = method.to_string(); - tokio::spawn(async move { - let id = v.get("id").cloned(); - let result = h - .route( - caller.clone(), - id.clone(), - method, - v.get("params").cloned().unwrap_or(json!({})), - false, - ) - .await; - let success = result.is_ok(); - if let Some(id) = id { - let value = match result { - Ok(r) => json!({"jsonrpc":"2.0","id":id,"result":r}), - Err(e) => json!({"jsonrpc":"2.0","id":id,"error":e}), - }; - h.send(&caller, value); - } - if exit && success { - h.0.exit.send_replace(true); - } - }); - } else if v["id"] == pid { - self.answer( - pid, - if let Some(e) = v.get("error") { - Err(e.clone()) - } else { - Ok(v["result"].clone()) - }, - ); - } - } - fn discard(&self, pid: &str) { - // The journal deduplicates this warning for each process. - if let Some(info) = self.processes().get(pid) { - self.0.journal.warning(pid, &info.plugin); - } - } - fn finish(&self, pid: &str, status: &str) { - let mut state = self.0.state.lock().unwrap(); - let process = state.processes.remove(pid); - if let Some(p) = process { - self.0.journal.write(pid, &p.info.plugin, "core", status); - if let Some(tx) = p.answer { - let _=tx.send(Err(json!({"code":-32002,"message":"process ended without an answer","data":{"status":status,"lines":self.0.journal.query(&json!({"pid":pid,"limit":20}))}}))); - } - Self::release_locks(&mut state, pid); - let _ = p.ended.send(true); - let children: Vec<_> = state - .processes - .iter() - .filter(|(_, p)| p.info.parent.as_deref() == Some(pid)) - .map(|(id, _)| id.clone()) - .collect(); - drop(state); - for child in children { - self.begin_stop(&child); - } - } - } - fn begin_stop(&self, pid: &str) -> Option> { - let mut state = self.0.state.lock().unwrap(); - let p = state.processes.get_mut(pid)?; - let ended = p.ended.subscribe(); - if p.info.state == "stopping" { - return Some(ended); - } - p.info.state = "stopping".into(); - p.input.take(); - self.0.journal.write(pid, &p.info.plugin, "core", "stop"); - let children: Vec<_> = state - .processes - .iter() - .filter(|(_, p)| p.info.parent.as_deref() == Some(pid)) - .map(|(id, _)| id.clone()) - .collect(); - drop(state); - for child in children { - self.begin_stop(&child); - } - let h = self.clone(); - let pid = pid.to_string(); - tokio::spawn(async move { - tokio::time::sleep(std::time::Duration::from_secs(5)).await; - h.kill(&pid); - }); - Some(ended) - } - fn kill(&self, pid: &str) { - if let Some(p) = self.0.state.lock().unwrap().processes.get(pid) { - self.0.journal.write(pid, &p.info.plugin, "core", "kill"); - if let Some(a) = &p.abort { - a.abort(); - } - } - self.finish(pid, "killed"); - } - pub async fn stop(&self) { - let ids: Vec<_> = self.processes().into_keys().collect(); - let mut waits = Vec::new(); - for pid in ids { - if let Some(n) = self.begin_stop(&pid) { - waits.push(n); - } - } - for mut n in waits { - let _ = n.wait_for(|ended| *ended).await; - } - } - fn release_locks(state: &mut State, pid: &str) { - for lock in state.locks.values_mut() { - lock.queue.retain(|(p, _)| p != pid); - } - let paths: Vec<_> = state - .locks - .iter() - .filter(|(_, l)| l.owner == pid) - .map(|(p, _)| p.clone()) - .collect(); - for path in paths { - grant(state, &path); - } - } - async fn core(&self, caller: &str, id: Option, method: &str, p: Value) -> RpcResult { - match method { - "call" => { - self.route( - caller.into(), - id, - required_str(&p, "method")?.into(), - p.get("params").cloned().unwrap_or(json!({})), - p["detach"].as_bool().unwrap_or(false), - ) - .await - } - "notify" => { - self.notify( - caller, - required_str(&p, "name")?, - p.get("params").cloned().unwrap_or(json!({})), - ); - Ok(Value::Null) - } - "kill" => { - self.kill(required_str(&p, "pid")?); - Ok(json!({})) - } - "stop" => { - if let Some(mut n) = self.begin_stop(required_str(&p, "pid")?) { - let _ = n.wait_for(|ended| *ended).await; - } - Ok(json!({})) - } - "lock" => { - let path = p["path"] - .as_str() - .ok_or_else(|| error(-32602, "missing path"))?; - let rx = { - let mut state = self.0.state.lock().unwrap(); - if let Some(lock) = state.locks.get_mut(path) { - if p["wait"] == false { - return Ok(json!({"held":false})); - } - let (tx, rx) = oneshot::channel(); - lock.queue.push_back((caller.into(), tx)); - Some(rx) - } else { - state.locks.insert( - path.into(), - Lock { - owner: caller.into(), - queue: VecDeque::new(), - }, - ); - None - } - }; - match rx { - Some(rx) => rx - .await - .unwrap_or_else(|_| Err(error(-32603, "lock canceled"))), - None => Ok(json!({"held":true})), - } - } - "unlock" => { - let mut state = self.0.state.lock().unwrap(); - let path = required_str(&p, "path")?; - if state.locks.get(path).is_some_and(|l| l.owner == caller) { - grant(&mut state, path); - } - Ok(Value::Null) - } - "log" => Ok(self.0.journal.query(&p)), - "exit" => Ok(json!({})), - _ => Err(error(-32601, "method not found")), - } - } -} -fn grant(state: &mut State, path: &str) { - if let Some(mut lock) = state.locks.remove(path) { - while let Some((owner, tx)) = lock.queue.pop_front() { - if tx.send(Ok(json!({"held":true}))).is_ok() { - lock.owner = owner; - state.locks.insert(path.into(), lock); - break; - } - } - } -} - -fn requests_exit(method: &str, params: &Value) -> bool { - method == "core.exit" - || (method == "core.call" - && requests_exit(params["method"].as_str().unwrap_or(""), ¶ms["params"])) -} - -fn required_str<'a>(params: &'a Value, name: &str) -> Result<&'a str, Value> { - params[name] - .as_str() - .ok_or_else(|| error(-32602, &format!("{name} must be a string"))) -} diff --git a/cli/src/locks.rs b/cli/src/locks.rs new file mode 100644 index 0000000..bae40ad --- /dev/null +++ b/cli/src/locks.rs @@ -0,0 +1,81 @@ +//! Advisory FIFO locks and cleanup of ended processes. +use crate::error; +use crate::{Host, RpcResult, host::State}; +use serde_json::json; +use std::collections::VecDeque; +use tokio::sync::oneshot; + +pub(crate) struct Lock { + pub(crate) owner: String, + pub(crate) queue: VecDeque<(String, oneshot::Sender)>, +} + +impl Host { + /// Acquire a FIFO queue position, or report whether the lock is free. + pub(crate) async fn lock(&self, caller: &str, path: &str, wait: bool) -> RpcResult { + let rx = { + let mut state = self.0.state.lock().unwrap(); + if let Some(lock) = state.locks.get_mut(path) { + if !wait { + return Ok(json!({"held":false})); + } + let (tx, rx) = oneshot::channel(); + lock.queue.push_back((caller.into(), tx)); + Some(rx) + } else { + state.locks.insert( + path.into(), + Lock { + owner: caller.into(), + queue: VecDeque::new(), + }, + ); + None + } + }; + match rx { + Some(rx) => rx + .await + .unwrap_or_else(|_| Err(error(-32603, "lock canceled"))), + None => Ok(json!({"held":true})), + } + } + /// Release a held lock only when this process owns it. + pub(crate) fn unlock(&self, caller: &str, path: &str) { + let mut state = self.0.state.lock().unwrap(); + if state + .locks + .get(path) + .is_some_and(|lock| lock.owner == caller) + { + grant(&mut state, path); + } + } + + /// Remove queued acquisitions and grant the next owner of each held lock. + pub(crate) fn release_locks(state: &mut State, pid: &str) { + for lock in state.locks.values_mut() { + lock.queue.retain(|(p, _)| p != pid); + } + let paths: Vec<_> = state + .locks + .iter() + .filter(|(_, l)| l.owner == pid) + .map(|(p, _)| p.clone()) + .collect(); + for path in paths { + grant(state, &path); + } + } +} +pub(crate) fn grant(state: &mut State, path: &str) { + if let Some(mut lock) = state.locks.remove(path) { + while let Some((owner, tx)) = lock.queue.pop_front() { + if tx.send(Ok(json!({"held":true}))).is_ok() { + lock.owner = owner; + state.locks.insert(path.into(), lock); + break; + } + } + } +} diff --git a/cli/src/main.rs b/cli/src/main.rs index b88d799..45c28fd 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -4,7 +4,6 @@ use clap::{Parser, Subcommand}; use sausage::{Host, world::World}; use serde_json::json; use std::{io::Write, path::PathBuf}; -use tokio::io::{AsyncBufReadExt, BufReader}; #[derive(Parser)] #[command(version)] struct Cli { @@ -71,24 +70,15 @@ async fn main() -> Result<()> { println!("{notification}"); } }); - let h = host.clone(); - let console = async move { - let mut lines = BufReader::new(tokio::io::stdin()).lines(); - loop { - tokio::select! { - line=lines.next_line()=>{let Some(line)=line? else {break;};if line=="quit" {break;} - let (method,params)=line.split_once(' ').unwrap_or((&line,"{}")); - match serde_json::from_str(params) {Ok(p)=>println!("{}",match h.call(method,p).await {Ok(v)=>v,Err(e)=>e}),Err(e)=>eprintln!("{e}")} - } - } - } - Ok::<_, anyhow::Error>(()) - }; + let console = sausage::console::run_input(host.clone()); let exit = tokio::select! { _=host.exit_requested()=>true, _=term.recv()=>false, _=tokio::signal::ctrl_c()=>false, - r=console,if !cli.headless=>{r?;false}, + result = console, if !cli.headless => { + result?; + false + }, }; let exit = exit || host.restart_requested(); host.stop().await; diff --git a/cli/src/native.rs b/cli/src/native.rs new file mode 100644 index 0000000..fe79854 --- /dev/null +++ b/cli/src/native.rs @@ -0,0 +1,35 @@ +//! Native plugin futures and their process context. +use crate::{Host, RpcResult}; +use serde_json::{Value, json}; +use std::{future::Future, pin::Pin, sync::Arc}; +use uuid::Uuid; +/// The cancellable future of a native process. +pub type NativeFuture = Pin + Send>>; +/// A handler for the methods of one native plugin. +pub type NativePlugin = Arc NativeFuture + Send + Sync>; +#[derive(Clone)] +/// The process identity and host access supplied to a native handler. +pub struct NativeContext { + /// The host that owns this native process. + pub host: Host, + /// The ID assigned to this native process. + pub pid: String, +} +impl NativeContext { + /// Make an outgoing call with this native process as its caller. + pub async fn call(&self, method: &str, params: Value) -> RpcResult { + self.host + .route( + self.pid.clone(), + Some(json!(Uuid::now_v7().to_string())), + method.into(), + params, + false, + ) + .await + } + /// Send progress to this process's caller while its call remains unanswered. + pub fn notify(&self, name: &str, params: Value) { + self.host.notify(&self.pid, name, params); + } +} diff --git a/cli/src/process.rs b/cli/src/process.rs new file mode 100644 index 0000000..ca08bbd --- /dev/null +++ b/cli/src/process.rs @@ -0,0 +1,131 @@ +//! Process metadata, process lifetime, and attached children. +use crate::{Host, RpcResult}; +use serde_json::{Value, json}; +use tokio::sync::{mpsc, oneshot, watch}; +#[derive(Clone, serde::Serialize)] +/// A snapshot of process metadata for inspection and the VFS. +pub struct ProcessInfo { + /// The plugin that supplied the process code. + pub plugin: String, + /// The SHA-256 of its component, or native for a native plugin. + pub version: String, + /// The method without its plugin prefix. + pub method: String, + /// The object supplied by the caller. + pub params: Value, + /// The attached parent, or no parent for a detached process. + pub parent: Option, + /// Whether this process survives the end of its caller. + pub detached: bool, + /// The start time in milliseconds since the Unix epoch. + pub start_time: u128, + /// The current lifecycle state. + pub state: String, +} +pub(crate) struct Process { + pub(crate) info: ProcessInfo, + pub(crate) input: Option>, + pub(crate) abort: Option, + pub(crate) caller: String, + pub(crate) id: Option, + 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) { + let mut state = self.0.state.lock().unwrap(); + let process = state.processes.remove(pid); + if let Some(p) = process { + self.0.journal.write(pid, &p.info.plugin, "core", status); + let logs = self.0.journal.query_later(json!({"pid":pid,"limit":20})); + Self::release_locks(&mut state, pid); + let status = status.to_string(); + tokio::spawn(async move { + let lines = logs.await.expect("journal writer stopped"); + if let Some(answer) = p.answer { + let failure = json!({ + "code": -32002, + "message": "process ended without an answer", + "data": { "status": status, "lines": lines } + }); + let _ = answer.send(Err(failure)); + } + p.ended.send_replace(true); + }); + let children: Vec<_> = state + .processes + .iter() + .filter(|(_, p)| p.info.parent.as_deref() == Some(pid)) + .map(|(id, _)| id.clone()) + .collect(); + drop(state); + for child in children { + self.begin_stop(&child); + } + } + } + /// Close stdin and start the same five-second deadline for each attached child. + pub(crate) fn begin_stop(&self, pid: &str) -> Option> { + let mut state = self.0.state.lock().unwrap(); + let p = state.processes.get_mut(pid)?; + let ended = p.ended.subscribe(); + if p.info.state == "stopping" { + return Some(ended); + } + p.info.state = "stopping".into(); + p.input.take(); + self.0.journal.write(pid, &p.info.plugin, "core", "stop"); + let children: Vec<_> = state + .processes + .iter() + .filter(|(_, p)| p.info.parent.as_deref() == Some(pid)) + .map(|(id, _)| id.clone()) + .collect(); + drop(state); + for child in children { + self.begin_stop(&child); + } + let h = self.clone(); + let pid = pid.to_string(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + h.kill(&pid); + }); + Some(ended) + } + /// 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) { + 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) { + let ids: Vec<_> = self.processes().into_keys().collect(); + let mut waits = Vec::new(); + for pid in ids { + if let Some(n) = self.begin_stop(&pid) { + waits.push(n); + } + } + for mut n in waits { + let _ = n.wait_for(|ended| *ended).await; + } + self.0.journal.query(json!({"limit":0})).await; + } +} diff --git a/cli/src/routing.rs b/cli/src/routing.rs new file mode 100644 index 0000000..6e995cc --- /dev/null +++ b/cli/src/routing.rs @@ -0,0 +1,233 @@ +//! JSON-RPC routing and responses between callers and processes. +use crate::{ + Host, NativeFuture, RpcResult, error, + process::{Process, ProcessInfo}, +}; +use serde_json::{Value, json}; +use tokio::sync::{mpsc, oneshot, watch}; +use uuid::Uuid; + +impl Host { + /// Queue an incoming response or notification for its destination. + pub(crate) fn send(&self, pid: &str, value: Value) { + if pid == "console" { + let _ = self.0.notifications.send(value); + } else if let Some(tx) = self + .0 + .state + .lock() + .unwrap() + .processes + .get(pid) + .and_then(|p| p.input.as_ref()) + { + let _ = tx.send(value); + } + } + /// Remap progress to the caller's original request ID. + pub(crate) fn notify(&self, pid: &str, name: &str, params: Value) { + let target = self + .0 + .state + .lock() + .unwrap() + .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); + } + } + /// Validate and route a request, then start its native or WASI process. + pub(crate) fn route( + &self, + caller: String, + id: Option, + method: String, + params: Value, + detached: bool, + ) -> NativeFuture { + let host = self.clone(); + Box::pin(async move { + if !params.is_object() { + return Err(error(-32602, "params must be an object")); + } + if caller != "console" && caller != "core" { + let state = host.0.state.lock().unwrap(); + let Some(process) = state.processes.get(&caller) else { + return Err(error(-32600, "process ended")); + }; + if process.info.state != "running" && method != "core.notify" { + return Err(error(-32600, "process is stopping")); + } + } + let (plugin, method) = method + .split_once('.') + .ok_or_else(|| error(-32601, "method not found"))?; + if plugin == "core" { + return host.core(&caller, id, method, params).await; + } + if plugin.is_empty() + || !plugin + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_') + { + return Err(error(-32601, "invalid plugin name")); + } + let native = host.0.state.lock().unwrap().native.get(plugin).cloned(); + let (component, version) = if native.is_some() { + (None, "native".into()) + } else { + let (component, hash) = host + .0 + .components + .load( + host.0.world.plugin_dir(plugin).join("plugin.wasm"), + host.0.runtime.clone(), + ) + .await?; + (Some(component), hash) + }; + let pid = format!("{plugin}-{}", Uuid::now_v7()); + let (tx, rx) = oneshot::channel(); + let (input, incoming) = mpsc::unbounded_channel::(); + let info = ProcessInfo { + plugin: plugin.into(), + version, + method: method.into(), + params: params.clone(), + parent: if detached { None } else { Some(caller.clone()) }, + detached, + start_time: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis(), + state: "running".into(), + }; + { + let mut state = host.0.state.lock().unwrap(); + if caller != "console" + && caller != "core" + && !state + .processes + .get(&caller) + .is_some_and(|p| p.info.state == "running") + { + return Err(error(-32600, "caller stopped or ended")); + } + host.0 + .journal + .write(&pid, plugin, "core", &format!("start {method} {params}")); + state.processes.insert( + pid.clone(), + Process { + info: info.clone(), + input: Some(input), + abort: None, + caller, + id: id.clone(), + answer: Some(tx), + ended: watch::channel(false).0, + }, + ); + } + let h = host.clone(); + let p = pid.clone(); + let has_id = id.is_some(); + let task = tokio::spawn(async move { + h.run_process(p, info, component, native, incoming, has_id) + .await; + }); + if let Some(p) = host.0.state.lock().unwrap().processes.get_mut(&pid) { + p.abort = Some(task.abort_handle()); + } else { + task.abort(); + } + if id.is_none() { + return Ok(Value::Null); + } + rx.await + .unwrap_or_else(|_| Err(error(-32002, "process ended"))) + }) + } + /// Complete the initial call once and discard late responses. + pub(crate) fn answer(&self, pid: &str, result: RpcResult) { + if let Some(tx) = self + .0 + .state + .lock() + .unwrap() + .processes + .get_mut(pid) + .and_then(|p| p.answer.take()) + { + let _ = tx.send(result); + } + } + /// Classify a stdout line and route requests or the initial response. + pub(crate) fn output(&self, pid: &str, line: String) { + if line.trim().is_empty() { + return; + } + let Ok(v) = serde_json::from_str::(&line) else { + self.discard(pid); + return; + }; + if !v.is_object() || v["jsonrpc"] != "2.0" { + self.discard(pid); + return; + } + if let Some(method) = v["method"].as_str() { + let h = self.clone(); + let caller = pid.to_string(); + let method = method.to_string(); + tokio::spawn(async move { + let id = v.get("id").cloned(); + let result = h + .route( + caller.clone(), + id.clone(), + method, + v.get("params").cloned().unwrap_or(json!({})), + false, + ) + .await; + if let Some(id) = id { + let value = match result { + Ok(result) => json!({ + "jsonrpc": "2.0", + "id": id, + "result": result + }), + Err(error) => json!({ + "jsonrpc": "2.0", + "id": id, + "error": error + }), + }; + h.send(&caller, value); + } + }); + } else if v["id"] == pid { + self.answer( + pid, + if let Some(e) = v.get("error") { + Err(e.clone()) + } else { + Ok(v["result"].clone()) + }, + ); + } + } + pub(crate) fn discard(&self, pid: &str) { + // The journal deduplicates this warning for each process. + if let Some(info) = self.processes().get(pid) { + self.0.journal.warning(pid, &info.plugin); + } + } +} diff --git a/cli/src/runtime.rs b/cli/src/runtime.rs index 40d23da..affe4a3 100644 --- a/cli/src/runtime.rs +++ b/cli/src/runtime.rs @@ -20,6 +20,7 @@ use wasmtime_wasi::p2::bindings::Command; use wasmtime_wasi::{FsPerms, I32Exit, WasiCtx, WasiCtxView, WasiView}; use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpHooks, WasiHttpView}; +/// The HTTP transport hooks supplied to WASI processes. pub struct HttpHooks { allow: bool, } @@ -54,6 +55,7 @@ impl WasiHttpHooks for HttpHooks { } } +/// The WASI and HTTP resources owned by one store. pub struct ProcessState { ctx: WasiCtx, table: ResourceTable, @@ -80,6 +82,7 @@ impl WasiHttpView for ProcessState { } } +/// A Wasmtime engine and linker shared by the world. pub struct Runtime { engine: Engine, linker: Linker, @@ -106,18 +109,23 @@ 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, /// Outgoing HTTP, and TCP, UDP, and DNS through `wasi:sockets`. pub network: bool, + /// The environment variables supplied to the guest. pub env: Vec<(String, String)>, } /// A running guest. Dropping it kills the guest. pub struct Instance { + /// The host end of stdin, removed when the process stops. pub stdin: Option, /// Lines that the guest writes to stdout. Closed when the guest ends. pub stdout: mpsc::Receiver, + /// The stderr lines that the process writes. pub stderr: mpsc::Receiver, task: JoinHandle, } @@ -178,6 +186,7 @@ fn short_trap(full: &str) -> String { } impl Runtime { + /// Create an engine with disk caching and a ten-millisecond epoch tick. pub fn new(cache: &Path) -> Result> { let mut config = Config::new(); config.epoch_interruption(true); @@ -206,11 +215,17 @@ impl Runtime { Ok(runtime) } + /// Compile component bytes through Wasmtime's disk cache. pub fn compile(&self, bytes: &[u8]) -> Result { Ok(Component::new(&self.engine, bytes)?) } - pub fn start(self: &Arc, component: &Component, cfg: InstanceConfig) -> Result { + /// Prepare a world preopen off the async workers, then start the WASI task. + pub async fn start( + self: &Arc, + component: &Component, + cfg: InstanceConfig, + ) -> Result { let (host_in, guest_in) = tokio::io::duplex(64 * 1024); let (guest_out, host_out) = tokio::io::duplex(64 * 1024); @@ -229,7 +244,12 @@ impl Runtime { b.env(k, v); } // Filesystem setup lives here. T6 replaces this preopen with the VFS. - b.preopened_dir(&cfg.world, "/", FsPerms::ReadWrite)?; + 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(), @@ -245,7 +265,9 @@ impl Runtime { if n == 0 { break; } - let line = String::from_utf8(std::mem::take(&mut bytes)).unwrap_or_default(); + // Preserve a non-blank invalid line so routing emits its warning. + let line = String::from_utf8(std::mem::take(&mut bytes)) + .unwrap_or_else(|_| "\u{fffd}".into()); if line_tx.send(line).await.is_err() { break; } diff --git a/cli/src/world.rs b/cli/src/world.rs index 997ccf7..992eea4 100644 --- a/cli/src/world.rs +++ b/cli/src/world.rs @@ -2,10 +2,13 @@ use anyhow::{Result, bail}; use std::path::{Path, PathBuf}; #[derive(Clone)] +/// The absolute host folder shared by all processes. pub struct World { + /// The absolute path of the world on the host. pub root: PathBuf, } impl World { + /// Validate reserved root names and resolve the world path. pub fn open(path: &Path) -> Result { for name in ["proc", "run", "self"] { if path.join(name).symlink_metadata().is_ok() { @@ -16,6 +19,7 @@ impl World { root: path.canonicalize()?, }) } + /// Create the folders and empty boot configuration of a new world. pub fn create(path: &Path) -> Result { if path.exists() { bail!("world already exists: {}", path.display()); @@ -26,13 +30,16 @@ impl World { std::fs::write(path.join("config/boot.json"), "[]\n")?; Self::open(path) } + /// Return the folder that contains a plugin component. pub fn plugin_dir(&self, name: &str) -> PathBuf { self.root.join("plugins").join(name) } + /// Return the persistent state folder of a plugin. pub fn state_dir(&self, name: &str) -> PathBuf { self.root.join("state").join(name) } } +/// Write a temporary sibling and rename it over the destination. pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> { std::fs::create_dir_all(path.parent().unwrap())?; let tmp = path.with_extension("tmp"); @@ -40,6 +47,7 @@ pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> { std::fs::rename(tmp, path)?; Ok(()) } +/// Replace a build copy with a source tree, without target folders. pub fn replace_source(from: &Path, to: &Path) -> Result<()> { if to.exists() { std::fs::remove_dir_all(to)?; diff --git a/cli/tests/core.rs b/cli/tests/core.rs index b709d00..0973a3b 100644 --- a/cli/tests/core.rs +++ b/cli/tests/core.rs @@ -52,15 +52,35 @@ async fn call(h: &Host, m: &str, p: Value) -> Value { .unwrap() .unwrap() } -#[tokio::test(flavor = "multi_thread")] -async fn routing_environment_files_and_errors() { + +/// PROTOCOL 3.1: json rpc lines carry results. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn json_rpc_lines_carry_results() { let w = TestWorld::new(); let h = w.start().await; assert_eq!(call(&h, "probe.echo", json!({"x":1})).await, json!({"x":1})); + + h.stop().await; +} + +/// PROTOCOL 3.2: method names split at the first dot. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn method_names_split_at_the_first_dot() { + let w = TestWorld::new(); + let h = w.start().await; assert_eq!( call(&h, "probe.rpc.discover", json!({"x":2})).await, json!({"x":2}) ); + + h.stop().await; +} + +/// PROTOCOL 3.3: processes can call other plugins. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn processes_can_call_other_plugins() { + let w = TestWorld::new(); + let h = w.start().await; assert_eq!( call( &h, @@ -70,10 +90,28 @@ async fn routing_environment_files_and_errors() { .await, json!({"nested":true}) ); + + h.stop().await; +} + +/// PROTOCOL 3.3: pending ids distinguish numbers from strings. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pending_ids_distinguish_numbers_from_strings() { + let w = TestWorld::new(); + let h = w.start().await; assert_eq!( call(&h, "probe.multicall", json!({})).await, json!({"count":2}) ); + + h.stop().await; +} + +/// PROTOCOL 3.3: notifications start attached processes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn notifications_start_attached_processes() { + let w = TestWorld::new(); + let h = w.start().await; call( &h, "probe.notification", @@ -84,14 +122,41 @@ async fn routing_environment_files_and_errors() { std::fs::read_to_string(w.0.join("state/notification")).unwrap(), "delivered" ); + + h.stop().await; +} + +/// PROTOCOL 3.3: unknown response ids are dropped. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unknown_response_ids_are_dropped() { + let w = TestWorld::new(); + let h = w.start().await; assert_eq!( h.call("probe.unknownid", json!({})).await.unwrap_err()["code"], -32002 ); + + h.stop().await; +} + +/// PROTOCOL 3.8: process environment identifies the caller. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn process_environment_identifies_the_caller() { + let w = TestWorld::new(); + let h = w.start().await; let env = call(&h, "probe.env", json!({})).await; assert_eq!(env["plugin"], "probe"); assert_eq!(env["parent"], "console"); assert!(env["pid"].as_str().unwrap().starts_with("probe-")); + + h.stop().await; +} + +/// PROTOCOL 3.6: processes read and write the world. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn processes_read_and_write_the_world() { + let w = TestWorld::new(); + let h = w.start().await; call( &h, "probe.write", @@ -102,18 +167,54 @@ async fn routing_environment_files_and_errors() { call(&h, "probe.read", json!({"path":"/state/test"})).await, "hello" ); + + h.stop().await; +} + +/// PROTOCOL 3.9: missing plugins return minus 32001. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn missing_plugins_return_minus_32001() { + let w = TestWorld::new(); + let h = w.start().await; assert_eq!( h.call("absent.run", json!({})).await.unwrap_err()["code"], -32001 ); + + h.stop().await; +} + +/// PROTOCOL 9: unimplemented core methods return method not found. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +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"], -32601 ); + + h.stop().await; +} + +/// PROTOCOL 3.1: params must be objects. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn params_must_be_objects() { + let w = TestWorld::new(); + let h = w.start().await; assert_eq!( h.call("probe.echo", json!([])).await.unwrap_err()["code"], -32602 ); + + h.stop().await; +} + +/// PROTOCOL 3.9: exit without answer reports status and last logs. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exit_without_answer_reports_status_and_last_logs() { + let w = TestWorld::new(); + let h = w.start().await; let e = h.call("probe.exit", json!({})).await.unwrap_err(); assert_eq!(e["code"], -32002); assert!( @@ -123,10 +224,13 @@ async fn routing_environment_files_and_errors() { .iter() .any(|l| l["line"] == "last line") ); + h.stop().await; } -#[tokio::test(flavor = "multi_thread")] -async fn journal_notifications_locks_and_kill() { + +/// PROTOCOL 3.5: progress notifications reach the caller. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn progress_notifications_reach_the_caller() { let w = TestWorld::new(); let h = w.start().await; let mut notifications = h.subscribe(); @@ -135,6 +239,15 @@ async fn journal_notifications_locks_and_kill() { notifications.recv().await.unwrap()["params"]["name"], "progress" ); + + h.stop().await; +} + +/// PROTOCOL 3.1: non message stdout warns once without its content. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn non_message_stdout_warns_once_without_its_content() { + let w = TestWorld::new(); + let h = w.start().await; call(&h, "probe.junk", json!({})).await; tokio::time::timeout(std::time::Duration::from_secs(5), async { while !h.processes().is_empty() { @@ -148,8 +261,28 @@ async fn journal_notifications_locks_and_kill() { let warnings = call(&h, "core.log", json!({"pid":pid,"grep":"discarded stdout"})).await; assert_eq!(warnings.as_array().unwrap().len(), 1); assert!(!warnings.to_string().contains("junk")); + + h.stop().await; +} + +/// PROTOCOL 3.1: stderr lines enter the journal. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn stderr_lines_enter_the_journal() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "probe.junk", json!({})).await; + h.stop().await; let log = call(&h, "core.log", json!({"grep":"stderr line"})).await; assert_eq!(log.as_array().unwrap().len(), 1); + + h.stop().await; +} + +/// PROTOCOL 6: try lock reports held without waiting. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn try_lock_reports_held_without_waiting() { + let w = TestWorld::new(); + let h = w.start().await; assert_eq!( call(&h, "core.lock", json!({"path":"x"})).await, json!({"held":true}) @@ -163,13 +296,25 @@ async fn journal_notifications_locks_and_kill() { call(&h, "probe.lock", json!({"path":"x"})).await, json!({"held":true}) ); + + h.stop().await; +} + +/// PROTOCOL 9: kill ends a process at once. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn kill_ends_a_process_at_once() { + let w = TestWorld::new(); + let h = w.start().await; let pid = call(&h, "probe.serve", json!({})).await; call(&h, "core.kill", json!({"pid":pid})).await; assert!(!h.processes().contains_key(pid.as_str().unwrap())); + h.stop().await; } -#[tokio::test(flavor = "multi_thread")] -async fn native_and_reserved_world_paths() { + +/// PROTOCOL 9: native calls have process ids and results. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn native_calls_have_process_ids_and_results() { let w = TestWorld::new(); let h = w.start().await; h.register_native( @@ -184,21 +329,30 @@ async fn native_and_reserved_world_paths() { let result = call(&h, "native.echo", json!({"x":1})).await; assert_eq!(result["method"], "echo"); assert!(result["pid"].as_str().unwrap().starts_with("native-")); + h.stop().await; +} + +/// PROTOCOL 2: reserved world roots prevent startup. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn reserved_world_roots_prevent_startup() { + let w = TestWorld::new(); + let h = w.start().await; for name in ["proc", "run", "self"] { std::fs::create_dir(w.0.join(name)).unwrap(); assert!(World::open(&w.0).is_err()); std::fs::remove_dir(w.0.join(name)).unwrap(); } + h.stop().await; } -#[tokio::test(flavor = "multi_thread")] -async fn tree_detach_stop_boot_and_scheduler() { +/// PROTOCOL 8: boot waits for each readiness result before the next service. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn boot_waits_for_readiness_in_order() { let w = TestWorld::new(); std::fs::write( w.0.join("config/boot.json"), serde_json::to_vec(&json!([ - {"method":"absent.start","params":{}}, {"method":"probe.write","params":{"path":"/state/boot","text":"ready"}}, {"method":"probe.read","params":{"path":"/state/boot"}} ])) @@ -210,6 +364,15 @@ async fn tree_detach_stop_boot_and_scheduler() { std::fs::read_to_string(w.0.join("state/boot")).unwrap(), "ready" ); + + h.stop().await; +} + +/// PROTOCOL 3.6: attached children stop when the parent ends. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn attached_children_stop_when_the_parent_ends() { + let w = TestWorld::new(); + let h = w.start().await; let attached = call( &h, "probe.call", @@ -223,6 +386,15 @@ async fn tree_detach_stop_boot_and_scheduler() { }) .await .unwrap(); + + h.stop().await; +} + +/// PROTOCOL 3.4: detached calls have no parent and outlive the caller. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn detached_calls_have_no_parent_and_outlive_the_caller() { + let w = TestWorld::new(); + let h = w.start().await; let detached = call( &h, "probe.call", @@ -231,6 +403,15 @@ async fn tree_detach_stop_boot_and_scheduler() { .await; assert!(h.processes()[detached.as_str().unwrap()].detached); call(&h, "core.stop", json!({"pid":detached})).await; + + h.stop().await; +} + +/// PROTOCOL 3.8: busy guests do not block other calls. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn busy_guests_do_not_block_other_calls() { + let w = TestWorld::new(); + let h = w.start().await; let hc = h.clone(); let busy = tokio::spawn(async move { hc.call("probe.busy", json!({})).await }); let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async { @@ -249,13 +430,16 @@ async fn tree_detach_stop_boot_and_scheduler() { ); call(&h, "core.kill", json!({"pid":pid})).await; assert_eq!(busy.await.unwrap().unwrap_err()["code"], -32002); + h.stop().await; } -#[tokio::test(flavor = "multi_thread")] +/// PROTOCOL 6: same process lock acquisitions queue. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn same_process_lock_acquisitions_queue() { let w = TestWorld::new(); let h = w.start().await; + h.register_native( "locks", std::sync::Arc::new(|ctx, _, _| { @@ -290,46 +474,16 @@ async fn same_process_lock_acquisitions_queue() { call(&h, "core.lock", json!({"path":"fifo","wait":false})).await, json!({"held":true}) ); -} - -#[tokio::test(flavor = "multi_thread")] -async fn stop_closes_stdin_deadline_and_exit() { - let w = TestWorld::new(); - let h = w.start().await; - let hc = h.clone(); - let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await }); - 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; - } - tokio::task::yield_now().await; - } - }) - .await - .unwrap(); - call(&h, "core.stop", json!({"pid":pid})).await; - assert_eq!( - std::fs::read_to_string(w.0.join("state/stopped")).unwrap(), - "yes" - ); - assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002); - assert_eq!(call(&h, "core.exit", json!({})).await, json!({})); - tokio::time::timeout(std::time::Duration::from_secs(1), h.exit_requested()) - .await - .unwrap(); h.stop().await; } -#[tokio::test(flavor = "multi_thread")] +/// PROTOCOL 3.8: tcp udp and dns. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn tcp_udp_and_dns() { - use tokio::io::AsyncWriteExt; let w = TestWorld::new(); let h = w.start().await; + + use tokio::io::AsyncWriteExt; let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let tcp = format!("localhost:{}", listener.local_addr().unwrap().port()); let udp = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(); @@ -346,11 +500,16 @@ async fn tcp_udp_and_dns() { udp.recv_from(&mut buf).await.unwrap(); assert_eq!(&buf, b"udp"); server.await.unwrap(); + h.stop().await; } -#[tokio::test(flavor = "multi_thread")] +/// PROTOCOL 6: native kill drops future and releases locks. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn native_kill_drops_future_and_releases_locks() { + let w = TestWorld::new(); + let h = w.start().await; + use std::sync::{ Arc, atomic::{AtomicBool, Ordering}, @@ -361,8 +520,6 @@ async fn native_kill_drops_future_and_releases_locks() { self.0.store(true, Ordering::SeqCst); } } - let w = TestWorld::new(); - let h = w.start().await; let dropped = Arc::new(AtomicBool::new(false)); let flag = dropped.clone(); h.register_native( @@ -405,13 +562,16 @@ async fn native_kill_drops_future_and_releases_locks() { json!({"held":true}) ); assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002); + h.stop().await; } -#[tokio::test(flavor = "multi_thread")] +/// PROTOCOL 3.6: shutdown stops busy processes in parallel after five seconds. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() { let w = TestWorld::new(); let h = w.start().await; + let mut tasks = Vec::new(); for _ in 0..2 { let hc = h.clone(); @@ -434,61 +594,16 @@ async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() { assert_eq!(task.await.unwrap().unwrap_err()["code"], -32002); } assert!(h.processes().is_empty()); + h.stop().await; } -#[tokio::test(flavor = "multi_thread")] -async fn journal_filters_persist_and_process_versions_are_hashes() { - use sha2::{Digest, Sha256}; - let w = TestWorld::new(); - let h = w.start().await; - let pid = call(&h, "probe.serve", json!({"version":1})).await; - let pid = pid.as_str().unwrap(); - let info = h.processes()[pid].clone(); - assert_eq!( - info.version, - format!("{:x}", Sha256::digest(std::fs::read(fixture()).unwrap())) - ); - assert_eq!(info.plugin, "probe"); - assert_eq!(info.method, "serve"); - assert_eq!(info.params, json!({"version":1})); - assert_eq!(info.parent.as_deref(), Some("console")); - assert_eq!(info.state, "running"); - assert!(!info.detached); - assert!(info.start_time > 0); - let uuid = uuid::Uuid::parse_str(pid.strip_prefix("probe-").unwrap()).unwrap(); - assert_eq!(uuid.get_version_num(), 7); - call(&h, "core.stop", json!({"pid":pid})).await; - let entries = call( - &h, - "core.log", - json!({"plugin":"probe","pid":pid,"limit":2}), - ) - .await; - assert_eq!(entries.as_array().unwrap().len(), 2); - let entry = &entries[0]; - let id = &entry["id"]; - let single = call(&h, "core.log", json!({"since":id,"until":id})).await; - assert_eq!(single, json!([entry])); - h.stop().await; - drop(h); - let h = w.start().await; - assert_eq!( - call(&h, "core.log", json!({"since":id,"until":id})).await, - single - ); - h.stop().await; - assert!( - std::fs::read(w.0.join("state/core/journal.sqlite")) - .unwrap() - .starts_with(b"SQLite format 3") - ); -} - -#[tokio::test(flavor = "multi_thread")] +/// PROTOCOL 3.6: stopping rejects calls but allows notifications. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn stopping_rejects_calls_but_allows_notifications() { - use std::sync::{Arc, Mutex}; let w = TestWorld::new(); let h = w.start().await; + + use std::sync::{Arc, Mutex}; let (tx, rx) = tokio::sync::oneshot::channel(); let tx = Arc::new(Mutex::new(Some(tx))); let gate = Arc::new(tokio::sync::Notify::new()); @@ -533,5 +648,465 @@ async fn stopping_rejects_calls_but_allows_notifications() { release.notify_one(); assert_eq!(pending.await.unwrap().unwrap(), json!({})); assert_eq!(stop.await.unwrap().unwrap(), json!({})); + + h.stop().await; +} + +/// PROTOCOL 9: stop closes stdin and waits for exit. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn stop_closes_stdin_and_waits_for_exit() { + let w = TestWorld::new(); + let h = w.start().await; + let hc = h.clone(); + let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await }); + 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; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + call(&h, "core.stop", json!({"pid":pid})).await; + assert_eq!( + std::fs::read_to_string(w.0.join("state/stopped")).unwrap(), + "yes" + ); + assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002); + + h.stop().await; +} + +/// PROTOCOL 9: core exit requests restart from its handler. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn core_exit_requests_restart_from_its_handler() { + let w = TestWorld::new(); + let h = w.start().await; + assert_eq!(call(&h, "core.exit", json!({})).await, json!({})); + tokio::time::timeout(std::time::Duration::from_secs(1), h.exit_requested()) + .await + .unwrap(); + + h.stop().await; +} + +/// PROTOCOL 3.3: process ids are uuidv7 and versions are sha256. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn process_ids_are_uuidv7_and_versions_are_sha256() { + let w = TestWorld::new(); + let h = w.start().await; + use sha2::{Digest, Sha256}; + let pid = call(&h, "probe.serve", json!({"version":1})).await; + let pid = pid.as_str().unwrap(); + let info = h.processes()[pid].clone(); + assert_eq!( + info.version, + format!("{:x}", Sha256::digest(std::fs::read(fixture()).unwrap())) + ); + assert_eq!(info.plugin, "probe"); + assert_eq!(info.method, "serve"); + assert_eq!(info.params, json!({"version":1})); + assert_eq!(info.parent.as_deref(), Some("console")); + assert_eq!(info.state, "running"); + assert!(!info.detached); + assert!(info.start_time > 0); + let uuid = uuid::Uuid::parse_str(pid.strip_prefix("probe-").unwrap()).unwrap(); + assert_eq!(uuid.get_version_num(), 7); + call(&h, "core.stop", json!({"pid":pid})).await; + + h.stop().await; +} + +/// PROTOCOL 7: journal filters and entries persist in sqlite. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn journal_filters_and_entries_persist_in_sqlite() { + let w = TestWorld::new(); + let h = w.start().await; + let pid = call(&h, "probe.serve", json!({})).await; + call(&h, "core.stop", json!({"pid":pid})).await; + let entries = call( + &h, + "core.log", + json!({"plugin":"probe","pid":pid,"limit":2}), + ) + .await; + assert_eq!(entries.as_array().unwrap().len(), 2); + let entry = &entries[0]; + let id = &entry["id"]; + let single = call(&h, "core.log", json!({"since":id,"until":id})).await; + assert_eq!(single, json!([entry])); + h.stop().await; + drop(h); + let h = w.start().await; + assert_eq!( + call(&h, "core.log", json!({"since":id,"until":id})).await, + single + ); + h.stop().await; + assert!( + std::fs::read(w.0.join("state/core/journal.sqlite")) + .unwrap() + .starts_with(b"SQLite format 3") + ); + h.stop().await; +} + +/// PROTOCOL 3.1: blank stdout lines do not warn. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn blank_stdout_lines_do_not_warn() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "probe.blank", json!({})).await; + h.stop().await; + let warnings = call(&h, "core.log", json!({"grep":"discarded stdout"})).await; + assert_eq!(warnings, json!([])); + h.stop().await; +} + +/// PROTOCOL 3.1: large requests and responses cross pipe capacity. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn large_requests_and_responses_cross_pipe_capacity() { + let w = TestWorld::new(); + let h = w.start().await; + let params = json!({"text":"x".repeat(192*1024)}); + assert_eq!(call(&h, "probe.echo", params.clone()).await, params); + let outgoing = json!({"method":"probe.echo","params":params}); + assert_eq!(call(&h, "probe.call", outgoing).await, params); + h.stop().await; +} + +/// PROTOCOL 3.8: two plugins execute calls at the same time. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn two_plugins_execute_calls_at_the_same_time() { + let w = TestWorld::new(); + let h = w.start().await; + let other = w.0.join("plugins/other"); + std::fs::create_dir_all(&other).unwrap(); + std::fs::copy(fixture(), other.join("plugin.wasm")).unwrap(); + let hc = h.clone(); + let slow = tokio::spawn(async move { hc.call("probe.sleep", json!({"ms":1500})).await }); + while !h.processes().values().any(|p| p.method == "sleep") { + tokio::task::yield_now().await; + } + assert_eq!( + call(&h, "other.echo", json!({"ready":true})).await, + json!({"ready":true}) + ); + assert!(!slow.is_finished()); + slow.await.unwrap().unwrap(); + h.stop().await; +} + +/// PROTOCOL 3.3: request id is the process id. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn request_id_is_the_process_id() { + let w = TestWorld::new(); + let h = w.start().await; + let pid = call(&h, "probe.request_id", json!({})).await; + let uuid = + uuid::Uuid::parse_str(pid.as_str().unwrap().strip_prefix("probe-").unwrap()).unwrap(); + assert_eq!(uuid.get_version_num(), 7); + h.stop().await; +} + +/// PROTOCOL 9: core call returns the target result. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn core_call_returns_the_target_result() { + let w = TestWorld::new(); + let h = w.start().await; + assert_eq!( + call( + &h, + "core.call", + json!({"method":"probe.echo","params":{"n":1}}) + ) + .await, + json!({"n":1}) + ); + h.stop().await; +} + +/// PROTOCOL 9: core unlock releases the named lock. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn core_unlock_releases_the_named_lock() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "core.lock", json!({"path":"unlock"})).await; + call(&h, "core.unlock", json!({"path":"unlock"})).await; + assert_eq!( + call(&h, "core.lock", json!({"path":"unlock","wait":false})).await, + json!({"held":true}) + ); + h.stop().await; +} + +/// PROTOCOL 6: locks do not block file writes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn locks_do_not_block_file_writes() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "core.lock", json!({"path":"/state/advisory"})).await; + call( + &h, + "probe.write", + json!({"path":"/state/advisory","text":"allowed"}), + ) + .await; + assert_eq!( + std::fs::read_to_string(w.0.join("state/advisory")).unwrap(), + "allowed" + ); + h.stop().await; +} + +/// PROTOCOL 6: crashes release all process locks. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn crashes_release_all_process_locks() { + let w = TestWorld::new(); + let h = w.start().await; + h.register_native( + "crash", + std::sync::Arc::new(|ctx, _, _| { + Box::pin(async move { + ctx.call("core.lock", json!({"path":"crash"})) + .await + .unwrap(); + panic!("test crash"); + }) + }), + ); + assert_eq!( + h.call("crash.run", json!({})).await.unwrap_err()["code"], + -32002 + ); + assert_eq!( + call(&h, "core.lock", json!({"path":"crash","wait":false})).await, + json!({"held":true}) + ); + h.stop().await; +} + +/// PROTOCOL 7: journal entries have uuid time identity stream and line. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn journal_entries_have_uuid_time_identity_stream_and_line() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "probe.junk", json!({})).await; + h.stop().await; + let entries = call(&h, "core.log", json!({"plugin":"probe"})).await; + for entry in entries.as_array().unwrap() { + assert_eq!( + uuid::Uuid::parse_str(entry["id"].as_str().unwrap()) + .unwrap() + .get_version_num(), + 7 + ); + assert!(entry["time"].as_str().unwrap().contains('T')); + assert!(entry["time"].as_str().unwrap().ends_with('Z')); + assert!(entry["pid"].as_str().unwrap().starts_with("probe-")); + assert!(["stderr", "core"].contains(&entry["stream"].as_str().unwrap())); + assert!(entry["line"].is_string()); + } + h.stop().await; +} + +/// PROTOCOL 7: journal has wal and identity indexes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn journal_has_wal_and_identity_indexes() { + let w = TestWorld::new(); + let h = w.start().await; + h.stop().await; + let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap(); + assert_eq!( + db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0)) + .unwrap(), + "wal" + ); + for name in ["journal_pid", "journal_plugin"] { + assert_eq!( + db.query_row( + "SELECT count(*) FROM sqlite_master WHERE type='index' AND name=?1", + [name], + |r| r.get::<_, i64>(0) + ) + .unwrap(), + 1 + ); + } + h.stop().await; +} + +/// PROTOCOL 9: core exit works through core call. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn core_exit_works_through_core_call() { + let w = TestWorld::new(); + let h = w.start().await; + assert_eq!( + call(&h, "core.call", json!({"method":"core.exit","params":{}})).await, + json!({}) + ); + assert!(h.restart_requested()); + h.stop().await; +} + +/// PROTOCOL 6: acquisitions from different processes keep their FIFO positions. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn lock_acquisitions_from_different_processes_are_fifo() { + let w = TestWorld::new(); + let h = w.start().await; + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let gate = std::sync::Arc::new(tokio::sync::Semaphore::new(0)); + let release = gate.clone(); + h.register_native( + "queued", + std::sync::Arc::new(move |ctx, _, _| { + let tx = tx.clone(); + let gate = gate.clone(); + Box::pin(async move { + tx.send(ctx).unwrap(); + gate.acquire().await.unwrap().forget(); + Ok(json!({})) + }) + }), + ); + let first_host = h.clone(); + let first = tokio::spawn(async move { first_host.call("queued.wait", json!({})).await }); + let second_host = h.clone(); + let second = tokio::spawn(async move { second_host.call("queued.wait", json!({})).await }); + let first_ctx = rx.recv().await.unwrap(); + let second_ctx = rx.recv().await.unwrap(); + call(&h, "core.lock", json!({"path":"ordered"})).await; + let mut first_lock = Box::pin(first_ctx.call("core.lock", json!({"path":"ordered"}))); + let mut second_lock = Box::pin(second_ctx.call("core.lock", json!({"path":"ordered"}))); + std::future::poll_fn(|cx| { + assert!(first_lock.as_mut().poll(cx).is_pending()); + assert!(second_lock.as_mut().poll(cx).is_pending()); + std::task::Poll::Ready(()) + }) + .await; + call(&h, "core.unlock", json!({"path":"ordered"})).await; + assert_eq!(first_lock.await.unwrap(), json!({"held":true})); + std::future::poll_fn(|cx| { + assert!(second_lock.as_mut().poll(cx).is_pending()); + std::task::Poll::Ready(()) + }) + .await; + first_ctx + .call("core.unlock", json!({"path":"ordered"})) + .await + .unwrap(); + assert_eq!(second_lock.await.unwrap(), json!({"held":true})); + release.add_permits(2); + first.await.unwrap().unwrap(); + second.await.unwrap().unwrap(); + h.stop().await; +} + +/// PROTOCOL 3.5: notifications after the result do not reach the caller. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn notifications_after_the_result_are_dropped() { + let w = TestWorld::new(); + let h = w.start().await; + let (tx, rx) = tokio::sync::oneshot::channel(); + let tx = std::sync::Mutex::new(Some(tx)); + h.register_native( + "native", + std::sync::Arc::new(move |ctx, _, _| { + tx.lock().unwrap().take().unwrap().send(ctx).ok(); + Box::pin(async { Ok(json!({})) }) + }), + ); + let mut notifications = h.subscribe(); + call(&h, "native.answer", json!({})).await; + let ctx = rx.await.unwrap(); + ctx.notify("late", json!({})); + assert!(matches!( + notifications.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + h.stop().await; +} + +/// PROTOCOL 3.1: non-UTF-8 stdout is discarded, not treated as blank framing. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn non_utf8_stdout_is_not_blank_framing() { + let w = TestWorld::new(); + let h = w.start().await; + call(&h, "probe.invalid_utf8", json!({})).await; + h.stop().await; + let warnings = call(&h, "core.log", json!({"grep":"discarded stdout"})).await; + assert_eq!(warnings.as_array().unwrap().len(), 1); +} + +/// PROTOCOL 8: a failed service enters the journal and does not prevent later services. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_boot_services_are_logged_and_boot_continues() { + let w = TestWorld::new(); + let services = json!([ + {"method":"absent.start","params":{}}, + {"method":"probe.write","params":{"path":"/state/continued","text":"yes"}} + ]); + std::fs::write( + w.0.join("config/boot.json"), + serde_json::to_vec(&services).unwrap(), + ) + .unwrap(); + let h = w.start().await; + assert_eq!( + std::fs::read_to_string(w.0.join("state/continued")).unwrap(), + "yes" + ); + let errors = call( + &h, + "core.log", + json!({"plugin":"core","grep":"no such plugin"}), + ) + .await; + assert_eq!(errors.as_array().unwrap().len(), 1); + h.stop().await; +} + +/// PROTOCOL 8: boot services are detached and remain alive after their readiness result. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn boot_services_are_detached() { + let w = TestWorld::new(); + std::fs::write( + w.0.join("config/boot.json"), + br#"[{"method":"probe.serve","params":{}}]"#, + ) + .unwrap(); + let h = w.start().await; + let processes = h.processes(); + assert_eq!(processes.len(), 1); + assert!(processes.values().next().unwrap().detached); + h.stop().await; +} + +/// PROTOCOL 7: the core records start parameters, stop, kill, and exit status. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn core_lifetime_events_enter_the_journal() { + let w = TestWorld::new(); + let h = w.start().await; + let stopped = call(&h, "probe.serve", json!({"n":1})).await; + call(&h, "core.stop", json!({"pid":stopped})).await; + let killed = call(&h, "probe.serve", json!({"n":2})).await; + call(&h, "core.kill", json!({"pid":killed})).await; + let entries = call(&h, "core.log", json!({"plugin":"probe","limit":100})).await; + let lines: Vec<_> = entries + .as_array() + .unwrap() + .iter() + .map(|entry| entry["line"].as_str().unwrap()) + .collect(); + assert!(lines.contains(&"start serve {\"n\":1}")); + for line in ["stop", "kill", "exit 0", "killed"] { + assert!(lines.contains(&line), "missing {line}"); + } h.stop().await; } diff --git a/cli/tests/plugins/probe/src/main.rs b/cli/tests/plugins/probe/src/main.rs index 1b0620f..5d7cf89 100644 --- a/cli/tests/plugins/probe/src/main.rs +++ b/cli/tests/plugins/probe/src/main.rs @@ -1,47 +1,219 @@ -//! A raw JSON-RPC guest for host integration tests. -use std::io::{self,BufRead,Write}; -use serde_json::{Value,json}; -fn send(v:Value) { let mut out=io::stdout().lock();writeln!(out,"\n{v}").unwrap();out.flush().unwrap(); } +//! Raw JSON-RPC methods used by the host's protocol tests. +use serde_json::{Value, json}; +use std::io::{self, BufRead, Read, Write}; +type Lines = io::Lines>; + +fn send(value: Value) { + let mut output = io::stdout().lock(); + writeln!(output, "\n{value}").unwrap(); + output.flush().unwrap(); +} +fn receive(lines: &mut Lines) -> Value { + serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap() +} +fn echo(params: &Value) -> Value { + params.clone() +} +fn environment() -> Value { + json!({ + "pid": std::env::var("SAUSAGE_PID").unwrap(), + "plugin": std::env::var("SAUSAGE_PLUGIN").unwrap(), + "parent": std::env::var("SAUSAGE_PARENT").unwrap(), + }) +} +fn multicall(lines: &mut Lines) -> Value { + send(json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "probe.echo", + "params": {"n": 1} + })); + send(json!({ + "jsonrpc": "2.0", + "id": "1", + "method": "probe.echo", + "params": {"n": 2} + })); + for _ in 0..2 { + let response = receive(lines); + if response["id"] == 1 { + assert_eq!(response["result"]["n"], 1); + } else { + assert_eq!(response["id"], "1"); + assert_eq!(response["result"]["n"], 2); + } + } + json!({"count":2}) +} +fn notification(params: &Value) -> Value { + send(json!({ + "jsonrpc": "2.0", + "method": "probe.write", + "params": params + })); + for _ in 0..200 { + let contents = std::fs::read_to_string(params["path"].as_str().unwrap()).ok(); + if contents.as_deref() == params["text"].as_str() { + break; + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + json!({}) +} +fn unknown_id() { + send(json!({ + "jsonrpc": "2.0", + "id": "late", + "result": 7 + })); +} +fn wait_eof(lines: &mut Lines) { + for _ in lines {} + std::fs::write("/state/stopped", "yes").unwrap(); +} +fn busy() -> Value { + loop { + std::hint::spin_loop(); + } +} +fn sleep(params: &Value) -> Value { + let duration = std::time::Duration::from_millis(params["ms"].as_u64().unwrap_or(100)); + std::thread::sleep(duration); + json!({}) +} +fn exit_without_answer() { + eprintln!("last line"); +} +fn junk() -> Value { + io::stdout().write_all(&[255, 10]).unwrap(); + println!("junk\n{{}}\n[]"); + eprintln!("stderr line"); + json!({}) +} +fn invalid_utf8() -> Value { + io::stdout().write_all(&[255, 10]).unwrap(); + json!({}) +} +fn blank() -> Value { + println!("\n \t\n"); + json!({}) +} +fn notify() -> Value { + send(json!({ + "jsonrpc": "2.0", + "method": "core.notify", + "params": { + "name": "progress", + "params": {"n": 1} + } + })); + std::thread::sleep(std::time::Duration::from_millis(30)); + json!({}) +} +fn serve(request: &Value, lines: &mut Lines) { + send(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": std::env::var("SAUSAGE_PID").unwrap() + })); + for _ in lines {} +} +fn network(params: &Value) -> Value { + let mut socket = std::net::TcpStream::connect(params["tcp"].as_str().unwrap()).unwrap(); + let mut text = String::new(); + socket.read_to_string(&mut text).unwrap(); + let udp = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + udp.send_to(b"udp", params["udp"].as_str().unwrap()) + .unwrap(); + json!(text) +} +fn read_file(params: &Value) -> Value { + json!(std::fs::read_to_string(params["path"].as_str().unwrap()).unwrap()) +} +fn write_file(params: &Value) -> Value { + std::fs::write( + params["path"].as_str().unwrap(), + params["text"].as_str().unwrap(), + ) + .unwrap(); + json!({}) +} +fn outgoing_call(params: &Value, lines: &mut Lines) -> Value { + send(json!({ + "jsonrpc": "2.0", + "id": {"nested": 7}, + "method": params["method"], + "params": params["params"] + })); + loop { + let response = receive(lines); + if response["id"] == json!({"nested":7}) { + return response["result"].clone(); + } + } +} +fn lock(params: &Value, lines: &mut Lines) -> Value { + outgoing_call(&json!({"method":"core.lock", "params":params}), lines) +} +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), + "env" => environment(), + "multicall" => multicall(lines), + "notification" => notification(params), + "unknownid" => { + unknown_id(); + return None; + } + "request_id" => request["id"].clone(), + "wait_eof" => { + wait_eof(lines); + return None; + } + "busy" => busy(), + "sleep" => sleep(params), + "exit" => { + exit_without_answer(); + return None; + } + "junk" => junk(), + "blank" => blank(), + "invalid_utf8" => invalid_utf8(), + "notify" => notify(), + "serve" => { + serve(request, lines); + return None; + } + "network" => network(params), + "read" => read_file(params), + "write" => write_file(params), + "call" => outgoing_call(params, lines), + "lock" => lock(params, lines), + _ => { + send(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "error": { + "code": -32601, + "message": "method not found" + } + })); + return None; + } + }; + Some(result) +} fn main() { - let mut lines=io::stdin().lock().lines(); - let request:Value=serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap(); - let id=request.get("id").cloned();let p=&request["params"]; - let result=match request["method"].as_str().unwrap() { - "echo"|"rpc.discover"=>p.clone(), - "env"=>json!({"pid":std::env::var("SAUSAGE_PID").unwrap(),"plugin":std::env::var("SAUSAGE_PLUGIN").unwrap(),"parent":std::env::var("SAUSAGE_PARENT").unwrap()}), - "multicall"=>{ - send(json!({"jsonrpc":"2.0","id":1,"method":"probe.echo","params":{"n":1}})); - send(json!({"jsonrpc":"2.0","id":"1","method":"probe.echo","params":{"n":2}})); - let mut seen=Vec::new();while seen.len()<2 {let r:Value=serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap();if r["id"]==1 {assert_eq!(r["result"]["n"],1);} else {assert_eq!(r["id"],"1");assert_eq!(r["result"]["n"],2);}seen.push(r["id"].clone());}json!({"count":seen.len()}) - }, - "notification"=>{ - send(json!({"jsonrpc":"2.0","method":"probe.write","params":p})); - for _ in 0..200 {if std::fs::exists(p["path"].as_str().unwrap()).unwrap() {break;}std::thread::sleep(std::time::Duration::from_millis(10));}json!({}) - }, - "unknownid"=>{send(json!({"jsonrpc":"2.0","id":"late","result":7}));return;}, - "request_id"=>request["id"].clone(), - "wait_eof"=>{for _ in lines {} std::fs::write("/state/stopped","yes").unwrap(); return;}, - "busy"=>loop {std::hint::spin_loop();}, - "sleep"=>{std::thread::sleep(std::time::Duration::from_millis(p["ms"].as_u64().unwrap_or(100)));json!({})}, - "exit"=>{eprintln!("last line");return;}, - "junk"=>{io::stdout().write_all(&[255,10]).unwrap();println!("junk\n{{}}\n[]");eprintln!("stderr line");json!({})}, - "notify"=>{send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"name":"progress","params":{"n":1}}}));std::thread::sleep(std::time::Duration::from_millis(30));json!({})}, - "serve"=>{send(json!({"jsonrpc":"2.0","id":id,"result":std::env::var("SAUSAGE_PID").unwrap()}));for _ in lines {} return;}, - "network"=>{ - use std::io::Read; - let mut socket=std::net::TcpStream::connect(p["tcp"].as_str().unwrap()).unwrap(); - let mut text=String::new();socket.read_to_string(&mut text).unwrap(); - let udp=std::net::UdpSocket::bind("127.0.0.1:0").unwrap();udp.send_to(b"udp",p["udp"].as_str().unwrap()).unwrap();json!(text) - }, - "read"=>json!(std::fs::read_to_string(p["path"].as_str().unwrap()).unwrap()), - "write"=>{std::fs::write(p["path"].as_str().unwrap(),p["text"].as_str().unwrap()).unwrap();json!({})}, - "call"|"lock"=>{ - let method=if request["method"]=="lock" {"core.lock"} else {p["method"].as_str().unwrap()}; - let params=if request["method"]=="lock" {p.clone()} else {p["params"].clone()}; - send(json!({"jsonrpc":"2.0","id":{"nested":7},"method":method,"params":params})); - loop {let response:Value=serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap();if response["id"]==json!({"nested":7}) {break response["result"].clone();}} - }, - _=>{send(json!({"jsonrpc":"2.0","id":id,"error":{"code":-32601,"message":"method not found"}}));return;} - }; - if let Some(id)=id {send(json!({"jsonrpc":"2.0","id":id,"result":result}));} + let mut lines = io::BufReader::new(io::stdin()).lines(); + let request = receive(&mut lines); + if let Some(result) = dispatch(&request, &mut lines) + && let Some(id) = request.get("id") + { + send(json!({ + "jsonrpc": "2.0", + "id": id, + "result": result + })); + } }