From 8d3c163992d9461c6f7133e1886764eca5cc70e0 Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 23 Oct 2026 00:35:45 +0100 Subject: [PATCH] sausage-plugin part 2: import! with typed clients (sausage_interface::client_rs), TCP endpoints with rpc.endpoint, examples Part 1 fixes: complete stdout writes, buffered stdin, Rust type names in schemas, enum variant docs, -32603 for serde errors in handlers, crate docs and modules. End-to-end check of a generated consumer through a fake core. --- Cargo.lock | 13 + examples/consumer/Cargo.toml | 7 + examples/consumer/imports/notes.json | 92 +++ examples/consumer/imports/secrets.json | 135 ++++ examples/consumer/imports/www.json | 104 +++ examples/consumer/src/main.rs | 47 ++ examples/notes-v2/Cargo.lock | 115 --- examples/notes-v2/Cargo.toml | 10 - examples/notes-v2/src/main.rs | 31 - examples/notes-v2/tests/basic.json | 9 - examples/notes/Cargo.lock | 568 +++++++++++++- examples/notes/Cargo.toml | 7 +- examples/notes/src/main.rs | 67 +- examples/notes/tests/basic.json | 7 - examples/secrets/Cargo.toml | 7 + examples/secrets/src/main.rs | 46 ++ examples/www/Cargo.toml | 8 + examples/www/src/main.rs | 75 ++ interface/Cargo.toml | 10 + interface/src/client.rs | 160 ++++ interface/src/lib.rs | 4 + interface/tests/client.rs | 16 + sdk/Cargo.toml | 4 +- sdk/macros/Cargo.toml | 2 + sdk/macros/src/lib.rs | 75 +- sdk/src/calls.rs | 78 ++ sdk/src/discovery.rs | 172 +++++ sdk/src/endpoints.rs | 247 +++++++ sdk/src/error.rs | 42 ++ sdk/src/helpers.rs | 19 + sdk/src/imports.rs | 46 ++ sdk/src/lib.rs | 696 ++---------------- sdk/src/lines.rs | 29 + sdk/src/locks.rs | 65 ++ sdk/src/transport.rs | 379 ++++++++++ sdk/tests/plugins/consumer/Cargo.toml | 7 + sdk/tests/plugins/consumer/imports/probe.json | 268 +++++++ sdk/tests/plugins/consumer/src/main.rs | 25 + sdk/tests/plugins/probe/src/main.rs | 34 +- sdk/tests/sdk.rs | 38 +- sdk/tests/ui/params.stderr | 8 +- sdk/tests/wasi.rs | 104 ++- 42 files changed, 3020 insertions(+), 856 deletions(-) create mode 100644 examples/consumer/Cargo.toml create mode 100644 examples/consumer/imports/notes.json create mode 100644 examples/consumer/imports/secrets.json create mode 100644 examples/consumer/imports/www.json create mode 100644 examples/consumer/src/main.rs delete mode 100644 examples/notes-v2/Cargo.lock delete mode 100644 examples/notes-v2/Cargo.toml delete mode 100644 examples/notes-v2/src/main.rs delete mode 100644 examples/notes-v2/tests/basic.json delete mode 100644 examples/notes/tests/basic.json create mode 100644 examples/secrets/Cargo.toml create mode 100644 examples/secrets/src/main.rs create mode 100644 examples/www/Cargo.toml create mode 100644 examples/www/src/main.rs create mode 100644 interface/src/client.rs create mode 100644 interface/tests/client.rs create mode 100644 sdk/src/calls.rs create mode 100644 sdk/src/discovery.rs create mode 100644 sdk/src/endpoints.rs create mode 100644 sdk/src/error.rs create mode 100644 sdk/src/helpers.rs create mode 100644 sdk/src/imports.rs create mode 100644 sdk/src/lines.rs create mode 100644 sdk/src/locks.rs create mode 100644 sdk/src/transport.rs create mode 100644 sdk/tests/plugins/consumer/Cargo.toml create mode 100644 sdk/tests/plugins/consumer/imports/probe.json create mode 100644 sdk/tests/plugins/consumer/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 2e9c8b2..cb5c54b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -193,6 +193,18 @@ version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "cap-fs-ext" +version = "4.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56ff379b70af8e08307a8f65e7040c7301cb4a572538ade16b4984f0da77847f" +dependencies = [ + "cap-primitives", + "cap-std", + "io-lifetimes 3.0.1", + "windows-sys 0.61.2", +] + [[package]] name = "cap-primitives" version = "4.0.3" @@ -1884,6 +1896,7 @@ version = "0.1.0" dependencies = [ "async-trait", "bytes", + "cap-fs-ext", "cap-std", "proptest", "rusqlite", diff --git a/examples/consumer/Cargo.toml b/examples/consumer/Cargo.toml new file mode 100644 index 0000000..53dd5e9 --- /dev/null +++ b/examples/consumer/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "consumer" +version = "0.1.0" +edition = "2024" +[workspace] +[dependencies] +sausage-plugin = { path = "../../sdk" } diff --git a/examples/consumer/imports/notes.json b/examples/consumer/imports/notes.json new file mode 100644 index 0000000..3a8d6fd --- /dev/null +++ b/examples/consumer/imports/notes.json @@ -0,0 +1,92 @@ +{ + "components": { + "schemas": { + "AddParams": { + "description": "Parameters for a new note.", + "properties": { + "text": { + "description": "The complete note text, including any newlines.", + "type": "string" + } + }, + "required": [ + "text" + ], + "type": "object" + }, + "Added": { + "description": "The result after a note enters the list.", + "properties": { + "count": { + "description": "The number of notes, including the new note.", + "format": "uint", + "minimum": 0, + "type": "integer" + } + }, + "required": [ + "count" + ], + "type": "object" + }, + "Empty": { + "description": "Empty parameters for methods without arguments.", + "type": "object" + } + } + }, + "info": { + "title": "notes", + "version": "0.1.0" + }, + "methods": [ + { + "description": "Add a note and return the new count.", + "name": "add", + "paramStructure": "by-name", + "params": [ + { + "description": "The complete note text, including any newlines.", + "name": "text", + "required": true, + "schema": { + "description": "The complete note text, including any newlines.", + "type": "string" + } + } + ], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/Added" + } + }, + "summary": "Add a note and return the new count.", + "x-sausage-params": { + "$ref": "#/components/schemas/AddParams" + } + }, + { + "description": "List all notes, oldest first.", + "name": "list", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + "summary": "List all notes, oldest first.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + } + ], + "openrpc": "1.2.6", + "x-sausage-endpoints": {}, + "x-sausage-imports": {} +} diff --git a/examples/consumer/imports/secrets.json b/examples/consumer/imports/secrets.json new file mode 100644 index 0000000..72c9814 --- /dev/null +++ b/examples/consumer/imports/secrets.json @@ -0,0 +1,135 @@ +{ + "components": { + "schemas": { + "Empty": { + "description": "Empty parameters for methods without arguments.", + "type": "object" + }, + "GetParams": { + "description": "The name of a secret to read.", + "properties": { + "name": { + "description": "The unique name within this vault.", + "type": "string" + } + }, + "required": [ + "name" + ], + "type": "object" + }, + "PutParams": { + "description": "The name and value of a secret.", + "properties": { + "name": { + "description": "The unique name within this vault.", + "type": "string" + }, + "value": { + "description": "The text to store until process exit.", + "type": "string" + } + }, + "required": [ + "name", + "value" + ], + "type": "object" + } + } + }, + "info": { + "title": "secrets", + "version": "0.1.0" + }, + "methods": [ + { + "description": "Open the in-memory vault and publish its endpoint.", + "name": "serve", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "null" + } + }, + "summary": "Open the in-memory vault and publish its endpoint.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + } + ], + "openrpc": "1.2.6", + "x-sausage-endpoints": { + "vault": { + "methods": [ + { + "description": "Store a secret and report its name to the caller.", + "name": "put", + "paramStructure": "by-name", + "params": [ + { + "description": "The unique name within this vault.", + "name": "name", + "required": true, + "schema": { + "description": "The unique name within this vault.", + "type": "string" + } + }, + { + "description": "The text to store until process exit.", + "name": "value", + "required": true, + "schema": { + "description": "The text to store until process exit.", + "type": "string" + } + } + ], + "result": { + "name": "result", + "schema": { + "type": "null" + } + }, + "summary": "Store a secret and report its name to the caller.", + "x-sausage-params": { + "$ref": "#/components/schemas/PutParams" + } + }, + { + "description": "Read a secret, or return null when its name is absent.", + "name": "get", + "paramStructure": "by-name", + "params": [ + { + "description": "The unique name within this vault.", + "name": "name", + "required": true, + "schema": { + "description": "The unique name within this vault.", + "type": "string" + } + } + ], + "result": { + "name": "result", + "schema": { + "type": [ + "string", + "null" + ] + } + }, + "summary": "Read a secret, or return null when its name is absent.", + "x-sausage-params": { + "$ref": "#/components/schemas/GetParams" + } + } + ] + } + }, + "x-sausage-imports": {} +} diff --git a/examples/consumer/imports/www.json b/examples/consumer/imports/www.json new file mode 100644 index 0000000..3a50913 --- /dev/null +++ b/examples/consumer/imports/www.json @@ -0,0 +1,104 @@ +{ + "components": { + "schemas": { + "Empty": { + "description": "Empty parameters for methods without arguments.", + "type": "object" + }, + "StartParams": { + "description": "The server configuration.", + "properties": { + "port": { + "description": "The HTTP port on all network interfaces.", + "format": "uint16", + "maximum": 65535, + "minimum": 0, + "type": "integer" + } + }, + "required": [ + "port" + ], + "type": "object" + } + } + }, + "info": { + "title": "www", + "version": "0.1.0" + }, + "methods": [ + { + "description": "Start the HTTP server and return its process ID.", + "name": "start", + "paramStructure": "by-name", + "params": [ + { + "description": "The HTTP port on all network interfaces.", + "name": "port", + "required": true, + "schema": { + "description": "The HTTP port on all network interfaces.", + "format": "uint16", + "maximum": 65535, + "minimum": 0, + "type": "integer" + } + } + ], + "result": { + "name": "result", + "schema": { + "type": "string" + } + }, + "summary": "Start the HTTP server and return its process ID.", + "x-sausage-params": { + "$ref": "#/components/schemas/StartParams" + } + } + ], + "openrpc": "1.2.6", + "x-sausage-endpoints": { + "control": { + "methods": [ + { + "description": "Count the HTTP requests this server answered.", + "name": "stats", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "format": "uint64", + "minimum": 0, + "type": "integer" + } + }, + "summary": "Count the HTTP requests this server answered.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + }, + { + "description": "End this process after the response reaches the client.", + "name": "stop", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "null" + } + }, + "summary": "End this process after the response reaches the client.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + } + ] + } + }, + "x-sausage-grace": 5, + "x-sausage-imports": {} +} diff --git a/examples/consumer/src/main.rs b/examples/consumer/src/main.rs new file mode 100644 index 0000000..f0f1401 --- /dev/null +++ b/examples/consumer/src/main.rs @@ -0,0 +1,47 @@ +//! Use typed calls from discovered provider documents in `imports/`. +//! Copy fresh documents from `/proc/core/plugins//discover.json` after an interface changes. +use sausage_plugin::{Result, rpc, schema}; +sausage_plugin::import!("notes"); +sausage_plugin::import!("www"); +sausage_plugin::import!("secrets"); +struct Consumer; +/// The text and HTTP port for this demonstration. +#[schema] +struct DemoParams { + /// The note text to append. + text: String, + /// The port for the detached HTTP server. + port: u16, +} +#[rpc] +impl Consumer { + /// Add a note, start a server, and inspect its control endpoint. + async fn demo(&self, p: DemoParams) -> Result { + let added = notes::add(notes::AddParams { text: p.text }).await?; + eprintln!("{} notes", added.count); + let pid = www::start(www::StartParams { port: p.port }) + .detach() + .await?; + let control = www::control::connect(format!("/proc/{pid}/control")).await?; + control.stats().await + } + /// Store and read a value through the boot service's fixed endpoint. + async fn vault(&self) -> Result> { + let vault = secrets::vault::connect("/run/secrets").await?; + vault + .put(secrets::PutParams { + name: "example".into(), + value: "hello".into(), + }) + .on_notify(|name, params| eprintln!("{name}: {params}")) + .await?; + vault + .get(secrets::GetParams { + name: "example".into(), + }) + .await + } +} +fn main() { + sausage_plugin::run(Consumer) +} diff --git a/examples/notes-v2/Cargo.lock b/examples/notes-v2/Cargo.lock deleted file mode 100644 index c5d02cc..0000000 --- a/examples/notes-v2/Cargo.lock +++ /dev/null @@ -1,115 +0,0 @@ -# This file is automatically @generated by Cargo. -# It is not intended for manual editing. -version = 4 - -[[package]] -name = "itoa" -version = "1.0.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" - -[[package]] -name = "memchr" -version = "2.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" - -[[package]] -name = "notes" -version = "0.1.0" -dependencies = [ - "sausage-actor", - "serde_json", -] - -[[package]] -name = "proc-macro2" -version = "1.0.107" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" -dependencies = [ - "unicode-ident", -] - -[[package]] -name = "quote" -version = "1.0.47" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" -dependencies = [ - "proc-macro2", -] - -[[package]] -name = "sausage-actor" -version = "0.1.0" -dependencies = [ - "serde", - "serde_json", -] - -[[package]] -name = "serde" -version = "1.0.229" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" -dependencies = [ - "serde_core", - "serde_derive", -] - -[[package]] -name = "serde_core" -version = "1.0.229" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" -dependencies = [ - "serde_derive", -] - -[[package]] -name = "serde_derive" -version = "1.0.229" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "serde_json" -version = "1.0.154" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6" -dependencies = [ - "itoa", - "memchr", - "serde", - "serde_core", - "zmij", -] - -[[package]] -name = "syn" -version = "3.0.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b" -dependencies = [ - "proc-macro2", - "quote", - "unicode-ident", -] - -[[package]] -name = "unicode-ident" -version = "1.0.27" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" - -[[package]] -name = "zmij" -version = "1.0.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/examples/notes-v2/Cargo.toml b/examples/notes-v2/Cargo.toml deleted file mode 100644 index 3ca0202..0000000 --- a/examples/notes-v2/Cargo.toml +++ /dev/null @@ -1,10 +0,0 @@ -[package] -name = "notes" -version = "0.1.0" -edition = "2024" - -[dependencies] -sausage-actor = { version = "0.1", path = "../../sdk" } -serde_json = "1" - -[workspace] diff --git a/examples/notes-v2/src/main.rs b/examples/notes-v2/src/main.rs deleted file mode 100644 index 3952b6e..0000000 --- a/examples/notes-v2/src/main.rs +++ /dev/null @@ -1,31 +0,0 @@ -//! Version 2 adds a note count and fixes the crash command. -use sausage_actor::{json, messages, reply}; - -const DATA: &str = "/data/notes.json"; - -fn save(notes: &[String]) { - std::fs::write(DATA, serde_json::to_string(notes).unwrap()).unwrap(); - std::fs::write("/public/notes.txt", notes.join("\n") + "\n").unwrap(); -} - -fn main() { - let mut notes: Vec = std::fs::read_to_string(DATA) - .ok() - .and_then(|text| serde_json::from_str(&text).ok()) - .unwrap_or_default(); - for message in messages() { - let body = &message.body; - let response = match message.kind() { - "add" => { - notes.push(body["text"].as_str().unwrap_or_default().to_string()); - save(¬es); - json!({"type": "added", "count": notes.len()}) - } - "list" => json!({"type": "notes", "items": notes}), - "count" => json!({"type": "count", "count": notes.len(), "version": 2}), - "crash" => json!({"type": "not_crashing", "version": 2}), - _ => json!({"type": "error", "reason": "unknown message"}), - }; - let _ = reply(&message, response); - } -} diff --git a/examples/notes-v2/tests/basic.json b/examples/notes-v2/tests/basic.json deleted file mode 100644 index 6900f67..0000000 --- a/examples/notes-v2/tests/basic.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "description": "version 2 counts notes and no longer crashes", - "steps": [ - {"send": {"type": "add", "text": "test note"}, "expect": {"type": "added"}}, - {"send": {"type": "list"}, "expect": {"type": "notes"}}, - {"send": {"type": "count"}, "expect": {"type": "count", "version": 2}}, - {"send": {"type": "crash"}, "expect": {"type": "not_crashing"}} - ] -} diff --git a/examples/notes/Cargo.lock b/examples/notes/Cargo.lock index c5d02cc..1bb114b 100644 --- a/examples/notes/Cargo.lock +++ b/examples/notes/Cargo.lock @@ -2,12 +2,252 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + +[[package]] +name = "async-task" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" + +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cfg-if" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" + +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + +[[package]] +name = "equivalent" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826" + +[[package]] +name = "fastrand" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be" +dependencies = [ + "instant", +] + +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-lite" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49a9d51ce47660b1e808d3c990b4709f2f415d928835a17dfd16991515c46bce" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "memchr", + "parking", + "pin-project-lite", + "waker-fn", +] + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project-lite", + "slab", +] + +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "instant" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "inventory" +version = "0.3.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6928282826c822ad91bf1c9a1cb90a30ba1c26770749929b4656cd6be829cd7c" +dependencies = [ + "rustversion", +] + [[package]] name = "itoa" version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + [[package]] name = "memchr" version = "2.8.3" @@ -18,8 +258,29 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" name = "notes" version = "0.1.0" dependencies = [ - "sausage-actor", - "serde_json", + "sausage-plugin", +] + +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn 2.0.119", ] [[package]] @@ -41,11 +302,136 @@ dependencies = [ ] [[package]] -name = "sausage-actor" +name = "ref-cast" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "regress" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "158a764437582235e3501f683b93a0a6f8d825d04a789dbe5ed30b8799b8908a" +dependencies = [ + "hashbrown", + "memchr", +] + +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + +[[package]] +name = "sausage-interface" version = "0.1.0" dependencies = [ + "prettyplease", + "proc-macro2", + "quote", + "schemars 0.8.22", + "serde_json", + "syn 2.0.119", + "typify", +] + +[[package]] +name = "sausage-plugin" +version = "0.1.0" +dependencies = [ + "futures", + "inventory", + "sausage-interface", + "sausage-plugin-macros", + "schemars 1.2.3", "serde", "serde_json", + "wstd", +] + +[[package]] +name = "sausage-plugin-macros" +version = "0.1.0" +dependencies = [ + "proc-macro2", + "quote", + "sausage-interface", + "serde_json", + "syn 2.0.119", +] + +[[package]] +name = "schemars" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3fbf2ae1b8bc8e02df939598064d22402220cd5bbcca1c76f7d6a310974d5615" +dependencies = [ + "dyn-clone", + "schemars_derive 0.8.22", + "serde", + "serde_json", +] + +[[package]] +name = "schemars" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab508826f74a77ca9d5aba6ff19b522583ee3eaf28a19384ff3d0e5835fadf6e" +dependencies = [ + "dyn-clone", + "ref-cast", + "schemars_derive 1.2.3", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e265784ad618884abaea0600a9adf15393368d840e0222d101a072f3f7534d" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals 0.29.1", + "syn 2.0.119", +] + +[[package]] +name = "schemars_derive" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1c3a92094fa7d61aa124645844facb6b554dfc797136d0f5fd1f890e2bffc69" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals 0.30.0", + "syn 3.0.9", +] + +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" +dependencies = [ + "serde", + "serde_core", ] [[package]] @@ -75,7 +461,29 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.9", +] + +[[package]] +name = "serde_derive_internals" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "serde_derive_internals" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", ] [[package]] @@ -91,6 +499,35 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_tokenstream" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7c49585c52c01f13c5c2ebb333f14f6885d76daa768d8a037d28017ec538c69" +dependencies = [ + "proc-macro2", + "quote", + "serde", + "syn 2.0.119", +] + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "syn" version = "3.0.9" @@ -102,12 +539,135 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "thiserror" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "typify" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc0b89f47309feaeb23c4509c15c9a04234f7deccef6f96c3bfe95319819a304" +dependencies = [ + "typify-impl", + "typify-macro", +] + +[[package]] +name = "typify-impl" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa7b026f540b148b81043c720889dbb942b08659aa8a43f624ac4f04dbfc1861" +dependencies = [ + "heck", + "log", + "proc-macro2", + "quote", + "regress", + "schemars 0.8.22", + "semver", + "serde", + "serde_json", + "syn 2.0.119", + "thiserror", + "unicode-ident", +] + +[[package]] +name = "typify-macro" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39ed96c57f06ae0839416b986921a98f18b220da63bbb243a8570a00c8492183" +dependencies = [ + "proc-macro2", + "quote", + "schemars 0.8.22", + "semver", + "serde", + "serde_json", + "serde_tokenstream", + "syn 2.0.119", + "typify-impl", +] + [[package]] name = "unicode-ident" version = "1.0.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" +[[package]] +name = "waker-fn" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7" + +[[package]] +name = "wasip2" +version = "1.0.4+wasi-0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" +dependencies = [ + "wit-bindgen", +] + +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +dependencies = [ + "bitflags", +] + +[[package]] +name = "wstd" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29b52936db10a79bb724dadd1c2c3aac958e8229dcb1f1c7f2b7044ca9fc6a3a" +dependencies = [ + "anyhow", + "async-task", + "bytes", + "futures-lite", + "http", + "http-body", + "http-body-util", + "itoa", + "pin-project-lite", + "serde", + "serde_json", + "slab", + "wasip2", + "wstd-macro", +] + +[[package]] +name = "wstd-macro" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "153db9b65508bf6c2efe26617169840c03671ba5719a35868db7682a5261f7d4" +dependencies = [ + "quote", + "syn 2.0.119", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/examples/notes/Cargo.toml b/examples/notes/Cargo.toml index 3ca0202..741be46 100644 --- a/examples/notes/Cargo.toml +++ b/examples/notes/Cargo.toml @@ -2,9 +2,6 @@ name = "notes" version = "0.1.0" edition = "2024" - -[dependencies] -sausage-actor = { version = "0.1", path = "../../sdk" } -serde_json = "1" - [workspace] +[dependencies] +sausage-plugin = { path = "../../sdk" } diff --git a/examples/notes/src/main.rs b/examples/notes/src/main.rs index a695596..83afd78 100644 --- a/examples/notes/src/main.rs +++ b/examples/notes/src/main.rs @@ -1,30 +1,43 @@ -//! Keep notes in `/data/notes.json` and publish them at `/public/notes.txt`. -use sausage_actor::{json, messages, reply}; - -const DATA: &str = "/data/notes.json"; - -fn save(notes: &[String]) { - std::fs::write(DATA, serde_json::to_string(notes).unwrap()).unwrap(); - std::fs::write("/public/notes.txt", notes.join("\n") + "\n").unwrap(); +//! Persistent notes. Each request reads the shared file instead of process memory. +use sausage_plugin::serde_json; +use sausage_plugin::{Result, lock, rpc, schema, state_dir, write_atomic}; +struct Notes; +/// Parameters for a new note. +#[schema] +struct AddParams { + /// The complete note text, including any newlines. + text: String, } - -fn main() { - let mut notes: Vec = std::fs::read_to_string(DATA) - .ok() - .and_then(|text| serde_json::from_str(&text).ok()) - .unwrap_or_default(); - for message in messages() { - let body = &message.body; - let response = match message.kind() { - "add" => { - notes.push(body["text"].as_str().unwrap_or_default().to_string()); - save(¬es); - json!({"type": "added", "count": notes.len()}) - } - "list" => json!({"type": "notes", "items": notes}), - "crash" => panic!("asked to crash"), - _ => json!({"type": "error", "reason": "unknown message"}), - }; - let _ = reply(&message, response); +/// The result after a note enters the list. +#[schema] +struct Added { + /// The number of notes, including the new note. + count: usize, +} +#[rpc] +impl Notes { + /// Add a note and return the new count. + async fn add(&self, AddParams { text }: AddParams) -> Result { + let path = state_dir().join("notes.json"); + let _guard = lock(&path).await?; + let mut notes = load()?; + notes.push(text); + std::fs::create_dir_all(state_dir())?; + write_atomic(path, serde_json::to_vec(¬es)?)?; + Ok(Added { count: notes.len() }) + } + /// List all notes, oldest first. + fn list(&self) -> Result> { + load() } } +fn load() -> Result> { + match std::fs::read(state_dir().join("notes.json")) { + Ok(bytes) => Ok(serde_json::from_slice(&bytes)?), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(vec![]), + Err(e) => Err(e.into()), + } +} +fn main() { + sausage_plugin::run(Notes) +} diff --git a/examples/notes/tests/basic.json b/examples/notes/tests/basic.json deleted file mode 100644 index 1f7a84b..0000000 --- a/examples/notes/tests/basic.json +++ /dev/null @@ -1,7 +0,0 @@ -{ - "description": "add a note, then list the notes", - "steps": [ - {"send": {"type": "add", "text": "test note"}, "expect": {"type": "added"}}, - {"send": {"type": "list"}, "expect": {"type": "notes"}} - ] -} diff --git a/examples/secrets/Cargo.toml b/examples/secrets/Cargo.toml new file mode 100644 index 0000000..d43901b --- /dev/null +++ b/examples/secrets/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "secrets" +version = "0.1.0" +edition = "2024" +[workspace] +[dependencies] +sausage-plugin = { path = "../../sdk" } diff --git a/examples/secrets/src/main.rs b/examples/secrets/src/main.rs new file mode 100644 index 0000000..8324df9 --- /dev/null +++ b/examples/secrets/src/main.rs @@ -0,0 +1,46 @@ +//! Keep a process-local vault behind a fixed endpoint at `/run/secrets`. +//! Start `secrets.serve` as a detached boot service. Its memory lasts until process exit. +use sausage_plugin::{Ctx, Result, listen, rpc, schema}; +use std::{cell::RefCell, collections::BTreeMap, rc::Rc}; +struct Secrets; +#[rpc(endpoints(vault=Vault))] +impl Secrets { + /// Open the in-memory vault and publish its endpoint. + async fn serve(&self) -> Result<()> { + listen("/run/secrets", Rc::new(Vault::default())).await + } +} +#[derive(Default)] +struct Vault { + entries: RefCell>, +} +/// The name and value of a secret. +#[schema] +struct PutParams { + /// The unique name within this vault. + name: String, + /// The text to store until process exit. + value: String, +} +/// The name of a secret to read. +#[schema] +struct GetParams { + /// The unique name within this vault. + name: String, +} +#[rpc] +impl Vault { + /// Store a secret and report its name to the caller. + fn put(&self, ctx: &Ctx, p: PutParams) -> Result<()> { + ctx.notify("stored", sausage_plugin::serde_json::json!({"name":p.name}))?; + self.entries.borrow_mut().insert(p.name, p.value); + Ok(()) + } + /// Read a secret, or return null when its name is absent. + fn get(&self, p: GetParams) -> Result> { + Ok(self.entries.borrow().get(&p.name).cloned()) + } +} +fn main() { + sausage_plugin::run(Secrets) +} diff --git a/examples/www/Cargo.toml b/examples/www/Cargo.toml new file mode 100644 index 0000000..aafafd1 --- /dev/null +++ b/examples/www/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "www" +version = "0.1.0" +edition = "2024" +[workspace] +[dependencies] +sausage-plugin = { path = "../../sdk" } +wstd = "0.6" diff --git a/examples/www/src/main.rs b/examples/www/src/main.rs new file mode 100644 index 0000000..fdff6f8 --- /dev/null +++ b/examples/www/src/main.rs @@ -0,0 +1,75 @@ +//! Serve files from `/state/www` and expose a control endpoint at `/self/control`. +use sausage_plugin::{Ctx, Result, background, listen, pid, rpc, schema, state_dir}; +use std::{cell::Cell, rc::Rc}; +use wstd::{ + io::{AsyncRead, AsyncWrite}, + iter::AsyncIterator, + net::TcpListener, +}; +struct Www; +/// The server configuration. +#[schema] +struct StartParams { + /// The HTTP port on all network interfaces. + port: u16, +} +#[rpc(endpoints(control=Control),grace_seconds=5)] +impl Www { + /// Start the HTTP server and return its process ID. + async fn start(&self, p: StartParams) -> Result { + let tcp = TcpListener::bind(&format!("0.0.0.0:{}", p.port)).await?; + let control = Rc::new(Control { hits: Cell::new(0) }); + listen("/self/control", control.clone()).await?; + background(serve(tcp, control)); + Ok(pid()) + } +} +struct Control { + hits: Cell, +} +#[rpc] +impl Control { + /// Count the HTTP requests this server answered. + fn stats(&self) -> Result { + Ok(self.hits.get()) + } + /// End this process after the response reaches the client. + fn stop(&self, ctx: &Ctx) -> Result<()> { + ctx.exit_after_response(); + Ok(()) + } +} +async fn serve(tcp: TcpListener, control: Rc) { + let mut incoming = tcp.incoming(); + while let Some(Ok(mut stream)) = incoming.next().await { + let mut request = [0; 8192]; + let Ok(size) = stream.read(&mut request).await else { + continue; + }; + let line = String::from_utf8_lossy(&request[..size]); + let path = line + .split_whitespace() + .nth(1) + .unwrap_or("/") + .trim_start_matches('/'); + let (status, body) = match std::fs::read(state_dir().join(if path.is_empty() { + "index.html" + } else { + path + })) { + Ok(body) => ("200 OK", body), + Err(_) => ("404 Not Found", b"Not found".to_vec()), + }; + control.hits.set(control.hits.get() + 1); + let header = format!( + "HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(header.as_bytes()).await; + let _ = stream.write_all(&body).await; + let _ = stream.flush().await; + } +} +fn main() { + sausage_plugin::run(Www) +} diff --git a/interface/Cargo.toml b/interface/Cargo.toml index 56b93af..3e080a6 100644 --- a/interface/Cargo.toml +++ b/interface/Cargo.toml @@ -4,8 +4,18 @@ version = "0.1.0" edition = "2024" description = "Sausage plugin interfaces: OpenRPC documents, the compatibility check, and client code generation." +[features] +default = ["codegen"] +codegen = ["dep:typify", "dep:schemars08", "dep:syn", "dep:quote", "dep:prettyplease", "dep:proc-macro2"] + [dependencies] serde_json = "1" +typify = { version = "0.6", optional = true } +schemars08 = { package = "schemars", version = "0.8", optional = true } +syn = { version = "2", features = ["full", "visit-mut"], optional = true } +quote = { version = "1", optional = true } +prettyplease = { version = "0.2", optional = true } +proc-macro2 = { version = "1", optional = true } [dev-dependencies] proptest = "1" diff --git a/interface/src/client.rs b/interface/src/client.rs new file mode 100644 index 0000000..3e12e0d --- /dev/null +++ b/interface/src/client.rs @@ -0,0 +1,160 @@ +//! Generate Rust clients from OpenRPC documents with typify's Draft 7 converter. +use quote::{format_ident, quote}; +use serde_json::{Value, json}; + +/// Generate a formatted private client module for a provider's document. +pub fn client_rs(name: &str, doc: &Value) -> Result { + fn local_refs(value: &mut Value) { + match value { + Value::Object(map) => { + if let Some(Value::String(r)) = map.get_mut("$ref") { + *r = r.replace("#/components/schemas/", "#/definitions/"); + } + for v in map.values_mut() { + local_refs(v); + } + } + Value::Array(a) => { + for v in a { + local_refs(v); + } + } + _ => {} + } + } + let module = + syn::parse_str::(&name.replace('-', "_")).map_err(|e| e.to_string())?; + let mut space = typify::TypeSpace::default(); + let mut defs = doc["components"]["schemas"] + .as_object() + .cloned() + .unwrap_or_default(); + for schema in defs.values_mut() { + local_refs(schema); + } + let defs: Vec<(String, schemars08::schema::Schema)> = defs + .into_iter() + .map(|(n, v)| serde_json::from_value(v).map(|v| (n, v))) + .collect::>() + .map_err(|e| e.to_string())?; + space.add_ref_types(defs).map_err(|e| e.to_string())?; + fn methods( + space: &mut typify::TypeSpace, + provider: &str, + methods: &Value, + endpoint: bool, + ) -> Result, String> { + let mut result = vec![]; + for method in methods.as_array().into_iter().flatten() { + let wire = method["name"].as_str().ok_or("method name missing")?; + let ident = syn::parse_str::(wire).map_err(|e| e.to_string())?; + let mut ps=method.get("x-sausage-params").cloned().unwrap_or_else(|| { + let mut properties=serde_json::Map::new();let mut required=vec![]; + for p in method["params"].as_array().into_iter().flatten(){let n=p["name"].as_str().unwrap();properties.insert(n.into(),p["schema"].clone());if p["required"]==true{required.push(n);}} + json!({"title":format!("{}Params",wire.split('_').map(|s|{let mut c=s.chars();c.next().unwrap().to_uppercase().to_string()+c.as_str()}).collect::()),"type":"object","properties":properties,"required":required}) + }); + local_refs(&mut ps); + let mut rs = method["result"]["schema"].clone(); + local_refs(&mut rs); + let p = space + .add_type(&serde_json::from_value(ps).map_err(|e| e.to_string())?) + .map_err(|e| e.to_string())?; + let r = space + .add_type(&serde_json::from_value(rs).map_err(|e| e.to_string())?) + .map_err(|e| e.to_string())?; + let p = space.get_type(&p).map_err(|e| e.to_string())?.ident(); + let r = space.get_type(&r).map_err(|e| e.to_string())?.ident(); + let docs = method + .get("description") + .or_else(|| method.get("summary")) + .and_then(Value::as_str) + .unwrap_or(""); + let empty = method["params"].as_array().is_none_or(Vec::is_empty); + let arg = if empty { quote!() } else { quote!(params:#p) }; + let params = if empty { + quote!(::sausage_plugin::serde_json::json!({})) + } else { + quote!(params) + }; + let target = format!("{provider}.{wire}"); + if endpoint { + let arg = if empty { quote!() } else { quote!(, #arg) }; + result.push(quote!(#[doc=#docs] #[warn(dead_code)] pub fn #ident(&self #arg)->::sausage_plugin::EndpointCall<#r>{self.0.call(#wire,#params)})); + } else { + result.push(quote!(#[doc=#docs] #[warn(dead_code)] pub fn #ident(#arg)->::sausage_plugin::TypedCall<#r>{::sausage_plugin::TypedCall::new(::sausage_plugin::call(#target,#params))})); + } + } + Ok(result) + } + let functions = methods(&mut space, name, &doc["methods"], false)?; + let mut endpoints = vec![]; + for (name, description) in doc["x-sausage-endpoints"].as_object().into_iter().flatten() { + let ident = format_ident!("{}", name); + let functions = methods(&mut space, "", &description["methods"], true)?; + endpoints.push(quote!{pub mod #ident {use super::*;pub struct Client(::sausage_plugin::Endpoint);pub async fn connect(path:impl AsRef<::std::path::Path>)->::sausage_plugin::Result{let endpoint=::sausage_plugin::connect(path).await?;endpoint.require_interface(#name)?;Ok(Client(endpoint))} impl Client {#(#functions)*}}}); + } + let types = space.to_stream().to_string(); + let mut file = syn::parse_file(&types).map_err(|e| e.to_string())?; + fn serde_attrs(items: &mut [syn::Item]) { + for item in items { + match item { + syn::Item::Struct(s) => s + .attrs + .push(syn::parse_quote!(#[serde(crate="::sausage_plugin::serde")])), + syn::Item::Enum(e) => e + .attrs + .push(syn::parse_quote!(#[serde(crate="::sausage_plugin::serde")])), + syn::Item::Mod(m) => { + if let Some((_, items)) = &mut m.content { + serde_attrs(items) + } + } + _ => {} + } + } + } + // Only data types have serde derives. ConversionError is a support type. + for item in &mut file.items { + if let syn::Item::Mod(m) = item + && m.ident == "error" + { + continue; + } + serde_attrs(std::slice::from_mut(item)); + } + let types = &file.items; + let mut file = + syn::parse2::(quote!(mod #module {#(#types)* #(#functions)* #(#endpoints)*})) + .map_err(|e| e.to_string())?; + struct Reexports; + impl syn::visit_mut::VisitMut for Reexports { + fn visit_path_mut(&mut self, path: &mut syn::Path) { + syn::visit_mut::visit_path_mut(self, path); + if path.leading_colon.is_some() + && path + .segments + .first() + .is_some_and(|s| s.ident == "serde" || s.ident == "serde_json") + { + path.segments.insert(0, syn::parse_quote!(sausage_plugin)); + } + } + fn visit_attribute_mut(&mut self, attr: &mut syn::Attribute) { + if attr.path().is_ident("derive") { + let mut paths = attr + .parse_args_with( + syn::punctuated::Punctuated::::parse_terminated, + ) + .expect("derive paths"); + for path in &mut paths { + self.visit_path_mut(path); + } + *attr = syn::parse_quote!(#[derive(#paths)]); + } else { + syn::visit_mut::visit_attribute_mut(self, attr); + } + } + } + syn::visit_mut::VisitMut::visit_file_mut(&mut Reexports, &mut file); + Ok(prettyplease::unparse(&file)) +} diff --git a/interface/src/lib.rs b/interface/src/lib.rs index 0b93d27..97d2aa1 100644 --- a/interface/src/lib.rs +++ b/interface/src/lib.rs @@ -4,3 +4,7 @@ pub mod compat; pub mod document; pub use document::Document; +#[cfg(feature="codegen")] +mod client; +#[cfg(feature="codegen")] +pub use client::client_rs; diff --git a/interface/tests/client.rs b/interface/tests/client.rs new file mode 100644 index 0000000..73ba768 --- /dev/null +++ b/interface/tests/client.rs @@ -0,0 +1,16 @@ +//! Check the shared generator against a real example's discovered document. +#[test] +fn notes_client_preserves_names_and_docs() { + let document = + serde_json::from_str(include_str!("../../examples/consumer/imports/notes.json")).unwrap(); + let code = sausage_interface::client_rs("notes", &document).unwrap(); + assert!(code.contains("pub struct AddParams")); + assert!(code.contains("pub struct Added")); + assert!(!code.contains("AddedResult")); + assert!(code.contains("The complete note text")); + assert!(code.contains("pub fn add(params: AddParams)")); + syn::parse_file(&code).unwrap(); + let target = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../target/sdk"); + std::fs::create_dir_all(&target).unwrap(); + std::fs::write(target.join("notes-client.rs"), code).unwrap(); +} diff --git a/sdk/Cargo.toml b/sdk/Cargo.toml index 630895e..e81d29b 100644 --- a/sdk/Cargo.toml +++ b/sdk/Cargo.toml @@ -6,14 +6,16 @@ description = "SDK for Sausage plugins." [dependencies] sausage-plugin-macros = { version = "0.1", path = "macros" } -sausage-interface = { path = "../interface" } +sausage-interface = { path = "../interface", default-features = false } serde = { version = "1", features = ["derive"] } serde_json = "1" schemars = "1" futures = "0.3" +inventory = "0.3" [target.'cfg(target_os = "wasi")'.dependencies] wstd = "0.6" [dev-dependencies] +sausage-interface = { path = "../interface", features = ["codegen"] } trybuild = "1" diff --git a/sdk/macros/Cargo.toml b/sdk/macros/Cargo.toml index 629383e..41b0b34 100644 --- a/sdk/macros/Cargo.toml +++ b/sdk/macros/Cargo.toml @@ -8,6 +8,8 @@ description = "Macros of the Sausage plugin SDK. Use them through `sausage-plugi proc-macro = true [dependencies] +sausage-interface = { path = "../../interface" } +serde_json = "1" syn = { version = "2", features = ["full"] } quote = "1" proc-macro2 = "1" diff --git a/sdk/macros/src/lib.rs b/sdk/macros/src/lib.rs index c8e03f4..0fd1236 100644 --- a/sdk/macros/src/lib.rs +++ b/sdk/macros/src/lib.rs @@ -47,6 +47,7 @@ fn serde_subset(attrs: &[Attribute]) -> syn::Result<()> { } Ok(()) } +/// Derive serde and Draft 7 schemas for a documented type in the supported subset. #[proc_macro_attribute] pub fn schema(_: TokenStream, input: TokenStream) -> TokenStream { let item = parse_macro_input!(input as Item); @@ -110,12 +111,14 @@ pub fn schema(_: TokenStream, input: TokenStream) -> TokenStream { })(); result.unwrap_or_else(syn::Error::into_compile_error).into() } +/// Generate interface dispatch and discovery from documented methods. #[proc_macro_attribute] pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream { let mut item = parse_macro_input!(input as ItemImpl); let args = proc_macro2::TokenStream::from(args); let result = (|| -> syn::Result { let mut grace = None; + let mut endpoints: Vec<(syn::Ident, syn::Type)> = vec![]; let parser = syn::meta::parser(|m| { if m.path.is_ident("grace_seconds") { grace = Some(m.value()?.parse::()?); @@ -123,7 +126,15 @@ pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream { } else if m.path.is_ident("endpoints") { let content; syn::parenthesized!(content in m.input); - let _ = content.parse::()?; + while !content.is_empty() { + let name = content.parse()?; + let _: syn::Token![=] = content.parse()?; + let ty = content.parse()?; + endpoints.push((name, ty)); + if !content.is_empty() { + let _: syn::Token![,] = content.parse()?; + } + } Ok(()) } else { Err(m.error("unknown RPC option")) @@ -179,7 +190,10 @@ pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream { )); } param = Some(p.ty.as_ref()); - args.push(quote!(::sausage_plugin::serde_json::from_value(params)?)); + args.push(quote!( + ::sausage_plugin::serde_json::from_value(params) + .map_err(|e| ::sausage_plugin::Error::new(-32602, e.to_string()))? + )); } } let await_ = f.sig.asyncness.map(|_| quote!(.await)); @@ -225,22 +239,77 @@ pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream { } let grace = grace.map(|n| quote!(doc["x-sausage-grace"]=::sausage_plugin::serde_json::json!(#n);)); + let names: Vec<_> = endpoints.iter().map(|(n, _)| n.to_string()).collect(); + let types: Vec<_> = endpoints.iter().map(|(_, t)| t).collect(); Ok(quote! {#item impl ::sausage_plugin::Interface for #ty { - fn discover(&self)->::sausage_plugin::serde_json::Value {let mut doc=::sausage_plugin::document(env!("CARGO_PKG_NAME"),env!("CARGO_PKG_VERSION")); #(#entries)* #grace doc} + fn endpoint_types(&self)->Vec<(::std::any::TypeId,String)>{vec![#((::std::any::TypeId::of::<#types>(),#names.into())),*]} + fn methods_document()->::sausage_plugin::serde_json::Value {let mut doc=::sausage_plugin::document(env!("CARGO_PKG_NAME"),env!("CARGO_PKG_VERSION")); #(#entries)* #grace ::sausage_plugin::finish_document(doc)} + fn describe()->::sausage_plugin::serde_json::Value {let mut doc=::methods_document(); #(let endpoint=<#types as ::sausage_plugin::Interface>::methods_document();doc["x-sausage-endpoints"][#names]=::sausage_plugin::serde_json::json!({"methods":endpoint["methods"]});doc["components"]["schemas"].as_object_mut().unwrap().extend(endpoint["components"]["schemas"].as_object().unwrap().clone());)* doc} + fn discover(&self)->::sausage_plugin::serde_json::Value {::describe()} fn dispatch<'a>(&'a self, method:&'a str, params: ::sausage_plugin::serde_json::Value,ctx:&'a ::sausage_plugin::Ctx)->::std::pin::Pin>+'a>> {Box::pin(async move {if !params.is_object() {return Err(::sausage_plugin::Error::new(-32602,"params must be an object"));} match method {#(#arms,)* _=>Err(::sausage_plugin::Error::new(-32601,"method not found"))}})} #stop }}) })(); result.unwrap_or_else(syn::Error::into_compile_error).into() } +/// Generate a private client from `imports/.json`. +#[proc_macro] +pub fn import(input: TokenStream) -> TokenStream { + let name = parse_macro_input!(input as syn::LitStr); + let result = (|| -> Result { + let path = std::path::PathBuf::from( + std::env::var("CARGO_MANIFEST_DIR").map_err(|e| e.to_string())?, + ) + .join("imports") + .join(format!("{}.json", name.value())); + let contents = + std::fs::read_to_string(&path).map_err(|e| format!("{}: {e}", path.display()))?; + let doc = serde_json::from_str(&contents).map_err(|e| e.to_string())?; + let code = sausage_interface::client_rs(&name.value(), &doc)?; + let generated: proc_macro2::TokenStream = code + .parse() + .map_err(|e: proc_macro2::LexError| e.to_string())?; + fn source_spans( + tokens: proc_macro2::TokenStream, + span: proc_macro2::Span, + ) -> proc_macro2::TokenStream { + tokens + .into_iter() + .map(|mut token| { + if let proc_macro2::TokenTree::Group(group) = &token { + let mut replacement = proc_macro2::Group::new( + group.delimiter(), + source_spans(group.stream(), span), + ); + replacement.set_span(span); + token = proc_macro2::TokenTree::Group(replacement); + } + token.set_span(span); + token + }) + .collect() + } + let generated = source_spans(generated, name.span()); + let path = path.to_str().ok_or("non-UTF8 import path")?; + Ok( + quote! {#generated ::sausage_plugin::inventory::submit! {::sausage_plugin::Import {name:#name,document:include_str!(#path)}}}, + ) + })(); + result + .unwrap_or_else(|e| syn::Error::new(name.span(), e).into_compile_error()) + .into() +} +/// Mark the method that handles `rpc.retired` after an upgrade. #[proc_macro_attribute] pub fn on_retired(_: TokenStream, input: TokenStream) -> TokenStream { input } +/// Mark the hook that runs at stdin EOF without a live call transport. #[proc_macro_attribute] pub fn on_stop(_: TokenStream, input: TokenStream) -> TokenStream { input } +/// Run an async test with the SDK runtime and the available core transport. #[proc_macro_attribute] pub fn test(_: TokenStream, input: TokenStream) -> TokenStream { let mut f = parse_macro_input!(input as syn::ItemFn); diff --git a/sdk/src/calls.rs b/sdk/src/calls.rs new file mode 100644 index 0000000..ec55bda --- /dev/null +++ b/sdk/src/calls.rs @@ -0,0 +1,78 @@ +//! Outgoing core calls and notification callbacks. +use super::*; +/// An outgoing request builder. +pub struct Call { + method: String, + params: Value, + detach: bool, + notify: Option, + pub(super) unlock: Option>, +} +/// Build an outgoing call with object parameters. +pub fn call(method: impl Into, params: impl serde::Serialize) -> Call { + Call { + method: method.into(), + params: serde_json::to_value(params).expect("serializable params"), + detach: false, + notify: None, + unlock: None, + } +} +impl Call { + /// Detach the child process from its caller. + pub fn detach(mut self) -> Self { + self.detach = true; + self + } + /// Receive progress notifications before the result arrives. + pub fn on_notify(mut self, f: impl FnMut(&str, Value) + 'static) -> Self { + self.notify = Some(Box::new(f)); + self + } +} +impl IntoFuture for Call { + type Output = Result; + type IntoFuture = LocalBoxFuture<'static, Self::Output>; + fn into_future(self) -> Self::IntoFuture { + Box::pin(async move { + if !self.params.is_object() { + return Err(Error::new(-32602, "params must be an object")); + } + let (tx, rx) = oneshot::channel(); + let id = STATE.with(|s| { + let mut s = s.borrow_mut(); + if !s.active || s.stopping { + return Err(Error::new( + -32002, + "no core transport (or stdin reached EOF)", + )); + } + s.next += 1; + let id = s.next; + s.pending.insert( + id, + Pending { + tx, + notify: self.notify, + unlock: self.unlock, + }, + ); + Ok(id) + })?; + let (method, params) = if self.detach { + ( + "core.call".to_string(), + json!({"method":self.method,"params":self.params,"detach":true}), + ) + } else { + (self.method, self.params) + }; + if let Err(e) = send(json!({"jsonrpc":"2.0","id":id,"method":method,"params":params})) { + STATE.with(|s| s.borrow_mut().pending.remove(&id)); + return Err(e); + } + rx.await + .unwrap_or_else(|_| Err(Error::new(-32002, "core transport closed"))) + }) + } +} diff --git a/sdk/src/discovery.rs b/sdk/src/discovery.rs new file mode 100644 index 0000000..d95dfe9 --- /dev/null +++ b/sdk/src/discovery.rs @@ -0,0 +1,172 @@ +//! Schema contracts and OpenRPC document generation. +use crate::*; +use serde_json::{Value, json}; + +/// Types within the supported schema subset. +#[diagnostic::on_unimplemented(message = "this type is outside the Sausage schema subset")] +pub trait SchemaType: schemars::JsonSchema {} +/// Named parameter objects. +#[diagnostic::on_unimplemented(message = "RPC params must be a #[schema] struct with named fields")] +pub trait Params: SchemaType {} +macro_rules! primitive { ($($t:ty),*) => { $(impl SchemaType for $t {})* }; } +primitive!( + String, + bool, + u8, + u16, + u32, + u64, + usize, + i8, + i16, + i32, + i64, + isize, + f32, + f64, + () +); +impl SchemaType for Vec {} +impl SchemaType for Option {} + +/// Create an OpenRPC document builder for generated interfaces. +#[doc(hidden)] +pub fn document(title: &str, version: &str) -> Value { + let mut doc = sausage_interface::Document::new(title, version).into_value(); + doc["x-sdk-input"] = json!({}); + doc["x-sdk-output"] = json!({}); + doc +} + +/// Add a method with independent deserialize and serialize contracts. +#[doc(hidden)] +pub fn method(doc: &mut Value, name: &str, docs: &str) { + fn generator(serialize: bool) -> schemars::SchemaGenerator { + let mut settings = schemars::generate::SchemaSettings::draft07(); + settings.definitions_path = if serialize { + "/sdk-output" + } else { + "/sdk-input" + } + .into(); + if serialize { + settings.for_serialize() + } else { + settings.for_deserialize() + } + .into_generator() + } + let mut input = generator(false); + let params_ref = serde_json::to_value(input.subschema_for::

()).unwrap(); + let definitions = input.take_definitions(true); + let schema = &definitions[P::schema_name().as_ref()]; + let params:Vec<_>=schema["properties"].as_object().into_iter().flatten().map(|(name,field)| { + json!({ + "name": name, + "description": field.get("description").cloned().unwrap_or(Value::Null), + "required": schema["required"].as_array().is_some_and(|required|required.contains(&json!(name))), + "schema": field, + }) + }).collect(); + doc["x-sdk-input"] + .as_object_mut() + .unwrap() + .extend(definitions); + let mut output = generator(true); + let result_schema = serde_json::to_value(output.subschema_for::()).unwrap(); + doc["x-sdk-output"] + .as_object_mut() + .unwrap() + .extend(output.take_definitions(true)); + doc["methods"].as_array_mut().unwrap().push(json!({ + "name":name, + "summary":docs.split("\n\n").next().unwrap_or_default(), + "description":docs, + "paramStructure":"by-name", + "params":params, + "result":{"name":"result","schema":result_schema}, + "x-sausage-params":params_ref, + })); +} + +/// Resolve shared schema names after all methods contribute their contracts. +#[doc(hidden)] +pub fn finish_document(mut doc: Value) -> Value { + let input = doc.as_object_mut().unwrap().remove("x-sdk-input").unwrap(); + let output = doc.as_object_mut().unwrap().remove("x-sdk-output").unwrap(); + fn refs(value: &mut Value, names: &std::collections::HashMap) { + match value { + Value::Object(map) => { + if let Some(Value::String(reference)) = map.get_mut("$ref") { + if let Some(name) = reference.strip_prefix("#/sdk-input/") { + *reference = format!("#/components/schemas/{name}"); + } else if let Some(name) = reference.strip_prefix("#/sdk-output/") { + *reference = format!( + "#/components/schemas/{}", + names.get(name).map(String::as_str).unwrap_or(name) + ); + } + } + for child in map.values_mut() { + refs(child, names); + } + } + Value::Array(values) => { + for child in values { + refs(child, names); + } + } + _ => {} + } + } + let mut names = std::collections::HashMap::new(); + for (name, schema) in output.as_object().unwrap() { + if let Some(other) = input.get(name) { + let mut a = schema.clone(); + let mut b = other.clone(); + refs(&mut a, &names); + refs(&mut b, &names); + if a != b { + names.insert(name.clone(), format!("{name}Result")); + } + } + } + // Propagate differences through references to nested contracts. + loop { + let before = names.len(); + for (name, schema) in output.as_object().unwrap() { + if let Some(other) = input.get(name) { + let mut a = schema.clone(); + let mut b = other.clone(); + refs(&mut a, &names); + refs(&mut b, &names); + if a != b { + names.insert(name.clone(), format!("{name}Result")); + } + } + } + if names.len() == before { + break; + } + } + doc["components"]["schemas"] = input; + for (name, schema) in output.as_object().unwrap() { + let name = names.get(name).unwrap_or(name); + doc["components"]["schemas"] + .as_object_mut() + .unwrap() + .insert(name.clone(), schema.clone()); + } + refs(&mut doc, &names); + for import in inventory::iter:: { + doc["x-sausage-imports"][import.name] = + serde_json::from_str(import.document).expect("validated import document"); + } + doc +} + +/// Empty parameters for methods without arguments. +#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)] +pub struct Empty {} +impl SchemaType for Empty {} +impl Params for Empty {} diff --git a/sdk/src/endpoints.rs b/sdk/src/endpoints.rs new file mode 100644 index 0000000..cacb1b9 --- /dev/null +++ b/sdk/src/endpoints.rs @@ -0,0 +1,247 @@ +//! TCP endpoint publication, verified connections, and per-request notifications. +use crate::*; +type Notify = Box; +use serde_json::Value; +#[cfg(target_os = "wasi")] +use serde_json::json; +use std::{future::IntoFuture, marker::PhantomData, path::Path, rc::Rc}; + +/// A verified endpoint address and identity. +#[derive(Clone)] +#[cfg_attr(not(target_os = "wasi"), allow(dead_code))] +pub struct Endpoint { + address: String, + owner: String, + interface: String, +} +impl Endpoint { + /// Require the interface expected by a generated client. + pub fn require_interface(&self, name: &str) -> Result<()> { + if self.interface == name { + Ok(()) + } else { + Err(Error::new(-32002, "endpoint interface mismatch")) + } + } + /// Build a typed endpoint request. + pub fn call(&self, method: &str, params: impl serde::Serialize) -> EndpointCall { + EndpointCall { + endpoint: self.clone(), + method: method.into(), + params: serde_json::to_value(params).expect("serializable params"), + notify: None, + result: PhantomData, + } + } +} +/// A lazy endpoint request with optional progress callbacks. +pub struct EndpointCall { + endpoint: Endpoint, + method: String, + params: Value, + notify: Option, + result: PhantomData, +} +impl EndpointCall { + /// Receive progress from this endpoint handler. + pub fn on_notify(mut self, f: impl FnMut(&str, Value) + 'static) -> Self { + self.notify = Some(Box::new(f)); + self + } +} +impl IntoFuture for EndpointCall { + type Output = Result; + type IntoFuture = futures::future::LocalBoxFuture<'static, Result>; + fn into_future(self) -> Self::IntoFuture { + Box::pin(async move { + Ok(serde_json::from_value( + endpoint_request(self.endpoint, &self.method, self.params, self.notify).await?, + )?) + }) + } +} + +/// Publish a declared interface and keep its listener alive until stdin EOF. +#[cfg(target_os = "wasi")] +pub async fn listen(path: impl AsRef, value: Rc) -> Result<()> { + use wstd::iter::AsyncIterator; + let name = crate::transport::endpoint_name::()?; + let listener = wstd::net::TcpListener::bind("127.0.0.1:0").await?; + write_atomic( + path, + format!("tcp:{} {} {}\n", listener.local_addr()?, pid(), name), + )?; + crate::transport::endpoint_task(async move { + let mut incoming = listener.incoming(); + while let Some(Ok(stream)) = incoming.next().await { + let value = value.clone(); + let name = name.clone(); + crate::transport::endpoint_task(async move { + let _ = serve_connection(stream, value, name).await; + }); + } + }); + Ok(()) +} +/// Publish an endpoint. Native programs require the WASI runtime. +#[cfg(not(target_os = "wasi"))] +pub async fn listen(_path: impl AsRef, _value: Rc) -> Result<()> { + crate::transport::endpoint_name::()?; + Err(Error::new(-32002, "endpoints require WASI")) +} + +/// Read an endpoint file and verify its owner before returning a client. +#[cfg(target_os = "wasi")] +pub async fn connect(path: impl AsRef) -> Result { + let mut attempts = 0; + let contents = loop { + match std::fs::read_to_string(path.as_ref()) { + Ok(s) => break s, + Err(e) if e.kind() == std::io::ErrorKind::NotFound && attempts < 20 => { + attempts += 1; + wstd::task::sleep(wstd::time::Duration::from_millis(25)).await; + } + Err(e) => return Err(e.into()), + } + }; + let words: Vec<_> = contents.split_whitespace().collect(); + if words.len() != 3 { + return Err(Error::new(-32602, "invalid endpoint file")); + } + let endpoint = Endpoint { + address: words[0] + .strip_prefix("tcp:") + .ok_or_else(|| Error::new(-32602, "invalid endpoint address"))? + .into(), + owner: words[1].into(), + interface: words[2].into(), + }; + let _ = handshake(&endpoint).await?; + Ok(endpoint) +} +/// Connect to an endpoint. Native programs require the WASI runtime. +#[cfg(not(target_os = "wasi"))] +pub async fn connect(_path: impl AsRef) -> Result { + Err(Error::new(-32002, "endpoints require WASI")) +} + +#[cfg(target_os = "wasi")] +async fn send(stream: &mut impl wstd::io::AsyncWrite, value: Value) -> Result<()> { + stream.write_all(format!("{value}\n").as_bytes()).await?; + stream.flush().await?; + Ok(()) +} +#[cfg(target_os = "wasi")] +async fn handshake(endpoint: &Endpoint) -> Result { + let mut stream = wstd::net::TcpStream::connect(endpoint.address.as_str()).await?; + send( + &mut stream, + json!({"jsonrpc":"2.0","id":0,"method":"rpc.endpoint","params":{}}), + ) + .await?; + let bytes = crate::lines::Lines::default() + .next(&mut stream) + .await? + .ok_or_else(|| Error::new(-32002, "endpoint closed"))?; + let reply: Value = serde_json::from_slice(&bytes)?; + let result = &reply["result"]; + let plugin = endpoint + .owner + .len() + .checked_sub(37) + .and_then(|end| endpoint.owner.get(..end)); + if result["owner"] != endpoint.owner + || result["interface"] != endpoint.interface + || result["plugin"].as_str() != plugin + { + return Err(Error::new(-32002, "endpoint identity mismatch")); + } + Ok(stream) +} +#[cfg(target_os = "wasi")] +async fn endpoint_request( + endpoint: Endpoint, + method: &str, + params: Value, + mut notify: Option, +) -> Result { + let mut stream = handshake(&endpoint).await?; + let mut lines = crate::lines::Lines::default(); + send( + &mut stream, + json!({"jsonrpc":"2.0","id":1,"method":method,"params":params}), + ) + .await?; + while let Some(bytes) = lines.next(&mut stream).await? { + let value: Value = serde_json::from_slice(&bytes)?; + if value["method"] == "core.notify" { + if let Some(f) = &mut notify { + f( + value["params"]["name"].as_str().unwrap_or_default(), + value["params"]["params"].clone(), + ); + } + } else if value["id"] == 1 { + if let Some(error) = value.get("error") { + return Err(serde_json::from_value(error.clone())?); + } + return Ok(value["result"].clone()); + } + } + Err(Error::new(-32002, "endpoint closed before response")) +} +#[cfg(not(target_os = "wasi"))] +async fn endpoint_request( + _endpoint: Endpoint, + _method: &str, + _params: Value, + _notify: Option, +) -> Result { + Err(Error::new(-32002, "endpoints require WASI")) +} + +#[cfg(target_os = "wasi")] +async fn serve_connection( + stream: wstd::net::TcpStream, + value: Rc, + name: String, +) -> Result<()> { + use futures::{FutureExt, StreamExt}; + let (mut reader, mut writer) = stream.split(); + let mut lines = crate::lines::Lines::default(); + while let Some(bytes) = lines.next(&mut reader).await? { + let request: Value = serde_json::from_slice(&bytes)?; + if request["jsonrpc"] != "2.0" { + continue; + } + let method = request["method"].as_str().unwrap_or_default(); + let (tx, mut rx) = futures::channel::mpsc::unbounded(); + let ctx = Ctx { + sink: Some(tx), + ..Ctx::default() + }; + let handler=async { + if method=="rpc.endpoint" {Ok(json!({"owner":pid(),"plugin":std::env::var("SAUSAGE_PLUGIN").unwrap_or_default(),"interface":name}))} + else {value.dispatch(method,request.get("params").cloned().unwrap_or(json!({})),&ctx).await} + }.fuse(); + futures::pin_mut!(handler); + let result = loop { + futures::select! {result=handler=>break result,notification=rx.next().fuse()=>{if let Some(notification)=notification {send(&mut writer,notification).await?;}}} + }; + while let Ok(notification) = rx.try_recv() { + send(&mut writer, notification).await?; + } + if let Some(id) = request.get("id") { + let response = match result { + Ok(result) => json!({"jsonrpc":"2.0","id":id,"result":result}), + Err(error) => json!({"jsonrpc":"2.0","id":id,"error":error}), + }; + send(&mut writer, response).await?; + } + if ctx.exit.get() { + crate::transport::exit(); + break; + } + } + Ok(()) +} diff --git a/sdk/src/error.rs b/sdk/src/error.rs new file mode 100644 index 0000000..7f3fa31 --- /dev/null +++ b/sdk/src/error.rs @@ -0,0 +1,42 @@ +//! JSON-RPC failures and conversions from local errors. +use serde_json::Value; +/// An SDK operation that can fail. +pub type Result = std::result::Result; + +/// A JSON-RPC failure. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct Error { + /// The JSON-RPC error code. + pub code: i64, + /// The diagnostic message. + pub message: String, + #[serde(skip_serializing_if = "Option::is_none")] + /// Optional structured error details. + pub data: Option, +} +impl Error { + /// Construct an error without structured details. + pub fn new(code: i64, message: impl Into) -> Self { + Self { + code, + message: message.into(), + data: None, + } + } +} +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.message) + } +} +impl std::error::Error for Error {} +impl From for Error { + fn from(e: std::io::Error) -> Self { + Self::new(-32603, e.to_string()) + } +} +impl From for Error { + fn from(e: serde_json::Error) -> Self { + Self::new(-32603, e.to_string()) + } +} diff --git a/sdk/src/helpers.rs b/sdk/src/helpers.rs new file mode 100644 index 0000000..ff2dea9 --- /dev/null +++ b/sdk/src/helpers.rs @@ -0,0 +1,19 @@ +//! Process identity and atomic state-file helpers. +use crate::Result; +use std::path::{Path, PathBuf}; +/// Return the process ID from `SAUSAGE_PID`. +pub fn pid() -> String { + std::env::var("SAUSAGE_PID").unwrap_or_default() +} +/// Return the persistent state directory of this plugin. +pub fn state_dir() -> PathBuf { + PathBuf::from("/state").join(std::env::var("SAUSAGE_PLUGIN").unwrap_or_default()) +} +/// Replace a file atomically after a complete temporary write. +pub fn write_atomic(path: impl AsRef, bytes: impl AsRef<[u8]>) -> Result<()> { + let path = path.as_ref(); + let temp = path.with_extension(format!("tmp-{}", pid())); + std::fs::write(&temp, bytes)?; + std::fs::rename(temp, path)?; + Ok(()) +} diff --git a/sdk/src/imports.rs b/sdk/src/imports.rs new file mode 100644 index 0000000..8dee1e4 --- /dev/null +++ b/sdk/src/imports.rs @@ -0,0 +1,46 @@ +//! Typed outgoing calls and statically registered imported interfaces. +use crate::{Call, Result}; +use serde_json::Value; +use std::{future::IntoFuture, marker::PhantomData}; + +/// An imported provider document registered by `import!`. +#[doc(hidden)] +pub struct Import { + /// The provider name. + pub name: &'static str, + /// Its complete OpenRPC document. + pub document: &'static str, +} +inventory::collect!(Import); + +/// An outgoing call whose result decodes to a generated Rust type. +pub struct TypedCall { + call: Call, + result: PhantomData, +} +impl TypedCall { + /// Wrap an untyped call without sending it yet. + pub fn new(call: Call) -> Self { + Self { + call, + result: PhantomData, + } + } + /// Detach the child process from its caller. + pub fn detach(mut self) -> Self { + self.call = self.call.detach(); + self + } + /// Receive progress notifications while the call runs. + pub fn on_notify(mut self, f: impl FnMut(&str, Value) + 'static) -> Self { + self.call = self.call.on_notify(f); + self + } +} +impl IntoFuture for TypedCall { + type Output = Result; + type IntoFuture = futures::future::LocalBoxFuture<'static, Result>; + fn into_future(self) -> Self::IntoFuture { + Box::pin(async move { Ok(serde_json::from_value(self.call.await?)?) }) + } +} diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index 10cf8bd..9bfd430 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -1,633 +1,65 @@ -//! JSON-RPC transport, lifetime, and helpers for single-threaded Sausage plugins. -use futures::{channel::oneshot, future::LocalBoxFuture}; -pub use sausage_plugin_macros::{on_retired, on_stop, rpc, schema, test}; -use serde_json::{Value, json}; -use std::{ - cell::RefCell, - collections::HashMap, - future::{Future, IntoFuture}, - io::Write, - path::{Path, PathBuf}, -}; -pub use {schemars, serde, serde_json}; -pub type Result = std::result::Result; - -/// A JSON-RPC failure. -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct Error { - pub code: i64, - pub message: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub data: Option, -} -impl Error { - pub fn new(code: i64, message: impl Into) -> Self { - Self { - code, - message: message.into(), - data: None, - } - } -} -impl std::fmt::Display for Error { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str(&self.message) - } -} -impl std::error::Error for Error {} -impl From for Error { - fn from(e: std::io::Error) -> Self { - Self::new(-32603, e.to_string()) - } -} -impl From for Error { - fn from(e: serde_json::Error) -> Self { - Self::new(-32602, e.to_string()) - } -} - -/// Types within the supported schema subset. -#[diagnostic::on_unimplemented(message = "this type is outside the Sausage schema subset")] -pub trait SchemaType: schemars::JsonSchema {} -/// Named parameter objects. -#[diagnostic::on_unimplemented(message = "RPC params must be a #[schema] struct with named fields")] -pub trait Params: SchemaType {} -macro_rules! primitive { ($($t:ty),*) => { $(impl SchemaType for $t {})* }; } -primitive!( - String, - bool, - u8, - u16, - u32, - u64, - usize, - i8, - i16, - i32, - i64, - isize, - f32, - f64, - () -); -impl SchemaType for Vec {} -impl SchemaType for Option {} - -/// Generated main-interface dispatch. -pub trait Interface: 'static { - fn discover(&self) -> Value; - fn dispatch<'a>( - &'a self, - method: &'a str, - params: Value, - ctx: &'a Ctx, - ) -> LocalBoxFuture<'a, Result>; - fn stop(&self) -> LocalBoxFuture<'_, ()> { - Box::pin(async {}) - } -} - -/// Context of an incoming call. -#[derive(Default)] -pub struct Ctx { - exit: std::cell::Cell, -} -impl Ctx { - pub fn notify(&self, name: &str, params: impl serde::Serialize) -> Result<()> { - notify("core.notify", json!({"name":name,"params":params})) - } - pub fn exit_after_response(&self) { - self.exit.set(true); - } -} - -type Notify = Box; -struct Pending { - tx: oneshot::Sender>, - notify: Option, - unlock: Option>, -} -struct Lease { - path: String, - held: std::cell::Cell, -} -impl Drop for Lease { - fn drop(&mut self) { - if self.held.get() { - let _ = notify("core.unlock", json!({"path":self.path})); - } - } -} -#[derive(Default)] -struct State { - active: bool, - stopping: bool, - next: u64, - pending: HashMap, - #[cfg(target_os = "wasi")] - tasks: usize, - #[cfg(target_os = "wasi")] - wake: Option, - #[cfg(target_os = "wasi")] - exit: bool, -} -thread_local! { static STATE: RefCell = RefCell::new(State::default()); } -fn send(value: Value) -> Result<()> { - let bytes = format!("\n{value}\n").into_bytes(); - let written = std::io::stdout().write(&bytes)?; - if written != bytes.len() { - return Err(Error::new(-32603, "short protocol write")); - } - Ok(()) -} -fn notify(method: &str, params: Value) -> Result<()> { - if !STATE.with(|s| s.borrow().active) { - return Err(Error::new(-32002, "no core transport")); - } - send(json!({"jsonrpc":"2.0","method":method,"params":params})) -} - -/// An outgoing request builder. -pub struct Call { - method: String, - params: Value, - detach: bool, - notify: Option, - unlock: Option>, -} -pub fn call(method: impl Into, params: impl serde::Serialize) -> Call { - Call { - method: method.into(), - params: serde_json::to_value(params).expect("serializable params"), - detach: false, - notify: None, - unlock: None, - } -} -impl Call { - pub fn detach(mut self) -> Self { - self.detach = true; - self - } - pub fn on_notify(mut self, f: impl FnMut(&str, Value) + 'static) -> Self { - self.notify = Some(Box::new(f)); - self - } -} -impl IntoFuture for Call { - type Output = Result; - type IntoFuture = LocalBoxFuture<'static, Self::Output>; - fn into_future(self) -> Self::IntoFuture { - Box::pin(async move { - if !self.params.is_object() { - return Err(Error::new(-32602, "params must be an object")); - } - let (tx, rx) = oneshot::channel(); - let id = STATE.with(|s| { - let mut s = s.borrow_mut(); - if !s.active || s.stopping { - return Err(Error::new( - -32002, - "no core transport (or stdin reached EOF)", - )); - } - s.next += 1; - let id = s.next; - s.pending.insert( - id, - Pending { - tx, - notify: self.notify, - unlock: self.unlock, - }, - ); - Ok(id) - })?; - let (method, params) = if self.detach { - ( - "core.call".to_string(), - json!({"method":self.method,"params":self.params,"detach":true}), - ) - } else { - (self.method, self.params) - }; - if let Err(e) = send(json!({"jsonrpc":"2.0","id":id,"method":method,"params":params})) { - STATE.with(|s| s.borrow_mut().pending.remove(&id)); - return Err(e); - } - rx.await - .unwrap_or_else(|_| Err(Error::new(-32002, "core transport closed"))) - }) - } -} - -/// An advisory lock acquisition. -pub struct Lock { - path: String, - wait: bool, -} -pub fn lock(path: impl AsRef) -> Lock { - Lock { - path: path.as_ref().to_string_lossy().into_owned(), - wait: true, - } -} -impl Lock { - pub fn try_now(mut self) -> Self { - self.wait = false; - self - } -} -/// Releases its lock on drop, unless kept until process exit. -pub struct Guard { - path: Option, -} -impl Guard { - pub fn keep(mut self) { - self.path.take(); - } - pub fn held(&self) -> bool { - self.path.is_some() - } -} -impl Drop for Guard { - fn drop(&mut self) { - if let Some(path) = self.path.take() { - let _ = notify("core.unlock", json!({"path":path})); - } - } -} -impl IntoFuture for Lock { - type Output = Result; - type IntoFuture = LocalBoxFuture<'static, Self::Output>; - fn into_future(self) -> Self::IntoFuture { - Box::pin(async move { - let lease = std::rc::Rc::new(Lease { - path: self.path.clone(), - held: std::cell::Cell::new(false), - }); - let mut c = call("core.lock", json!({"path":self.path,"wait":self.wait})); - c.unlock = Some(lease.clone()); - let result = c.await?; - lease.held.set(false); - Ok(Guard { - path: if result["held"] == true { - Some(self.path) - } else { - None - }, - }) - }) - } -} - -pub fn pid() -> String { - std::env::var("SAUSAGE_PID").unwrap_or_default() -} -pub fn state_dir() -> PathBuf { - PathBuf::from("/state").join(std::env::var("SAUSAGE_PLUGIN").unwrap_or_default()) -} -pub fn write_atomic(path: impl AsRef, bytes: impl AsRef<[u8]>) -> Result<()> { - let path = path.as_ref(); - let temp = path.with_extension(format!("tmp-{}", pid())); - std::fs::write(&temp, bytes)?; - std::fs::rename(temp, path)?; - Ok(()) -} - -#[cfg(target_os = "wasi")] -pub fn block_on(f: F) -> F::Output { - wstd::runtime::block_on(f) -} -#[cfg(not(target_os = "wasi"))] -pub fn block_on(f: F) -> F::Output { - futures::executor::block_on(f) -} - -/// Run an SDK test with a core transport when the WASI runner supplies one. -#[cfg(target_os = "wasi")] -pub fn test_run(f: F) -> F::Output { - block_on(async { - use wstd::io::AsyncRead; - STATE.with(|s| s.borrow_mut().active = true); - let reader = wstd::runtime::spawn(async { - let mut input = wstd::io::stdin(); - let mut line = Vec::new(); - loop { - let mut byte = [0]; - if !matches!(input.read(&mut byte).await, Ok(1)) { - break; - } - if byte[0] == b'\n' { - if let Ok(v) = serde_json::from_slice::(&line) { - if v["jsonrpc"] == "2.0" { - receive(v); - } - } - line.clear(); - } else { - line.push(byte[0]); - } - } - STATE.with(|s| { - let mut s = s.borrow_mut(); - s.stopping = true; - s.pending.clear(); - }); - }); - let result = f.await; - drop(reader); - STATE.with(|s| { - let mut s = s.borrow_mut(); - s.active = false; - s.pending.clear(); - }); - result - }) -} -#[cfg(not(target_os = "wasi"))] -pub fn test_run(f: F) -> F::Output { - block_on(f) -} - -/// Register a task that keeps the process alive. -#[cfg(target_os = "wasi")] -pub fn background(f: impl Future + 'static) { - STATE.with(|s| s.borrow_mut().tasks += 1); - wstd::runtime::spawn(async move { - f.await; - STATE.with(|s| { - let mut s = s.borrow_mut(); - s.tasks -= 1; - if let Some(w) = s.wake.take() { - w.wake(); - } - }); - }) - .detach(); -} -#[cfg(not(target_os = "wasi"))] -pub fn background(_f: impl Future + 'static) { - panic!("background requires the WASI runtime"); -} - +//! Write single-threaded WASI plugins for Sausage. +//! +//! Each core call starts a process. Keep its initial value cheap to construct. +//! Declare documented parameter and result types with [`macro@schema`]. +//! Declare methods with [`macro@rpc`], then serve them with [`run`]. +//! +//! ```no_run +//! use sausage_plugin::{rpc, schema, Result}; +//! struct Calculator; +//! /// The numbers to add. +//! #[schema] +//! struct AddParams { +//! /// The first number. +//! a: i32, +//! /// The second number. +//! b: i32, +//! } +//! #[rpc] +//! impl Calculator { +//! /// Add two numbers. +//! fn add(&self, p: AddParams) -> Result { Ok(p.a + p.b) } +//! } +//! fn main() { sausage_plugin::run(Calculator); } +//! ``` +//! +//! Use [`call`] for outgoing calls. Await its builder for the JSON result. +//! Use `detach()` to detach the child and `on_notify(f)` to receive progress. +//! Use [`Ctx::notify`] to report progress to your caller. +//! Use [`Ctx::exit_after_response`] to end the process after the response. +//! +//! [`background`] keeps the process alive while a task runs. +//! At stdin EOF, pending calls fail and the optional [`macro@on_stop`] hook runs. +//! The stop hook can write files and send notifications, but cannot make calls. +//! [`macro@on_retired`] handles `rpc.retired` after an upgrade. +//! +//! Use [`lock`] before shared state changes. A guard releases its lock on drop. +//! `try_now()` returns a guard whose `held()` method reports success. +//! `keep()` retains a lock until process exit. +//! [`state_dir`], [`write_atomic`], and [`pid`] provide filesystem and identity helpers. +//! Use [`macro@test`] for async tests. Native tests fail clearly without a core transport. +//! [`macro@import`] reads `imports/.json` and generates a private typed client. +//! Its functions return [`TypedCall`] builders with progress and detach options. +//! Declare endpoint types with `#[rpc(endpoints(control = Control))]`. +//! [`listen`] publishes a declared interface at an endpoint file. +//! [`connect`] verifies its owner before it returns an [`Endpoint`]. +//! Endpoint calls report progress through [`EndpointCall::on_notify`]. +//! Log with `eprintln!`. Stdout belongs to JSON-RPC. +#![warn(missing_docs)] +mod discovery; +mod endpoints; +mod error; +mod helpers; +mod imports; +mod lines; +mod transport; +pub use discovery::*; +pub use endpoints::*; +pub use error::*; +pub use helpers::*; +pub use imports::*; #[doc(hidden)] -pub fn receive(value: Value) { - if value["method"] == "core.notify" { - if let Some(id) = value["params"]["call"].as_u64() { - let callback = STATE.with(|s| { - s.borrow_mut() - .pending - .get_mut(&id) - .and_then(|p| p.notify.take()) - }); - if let Some(mut f) = callback { - f( - value["params"]["name"].as_str().unwrap_or_default(), - value["params"]["params"].clone(), - ); - STATE.with(|s| { - if let Some(p) = s.borrow_mut().pending.get_mut(&id) { - p.notify = Some(f); - } - }); - } - } - return; - } - let pending = value["id"] - .as_u64() - .and_then(|id| STATE.with(|s| s.borrow_mut().pending.remove(&id))); - if let Some(p) = pending { - let result = if value.get("error").is_some() { - serde_json::from_value(value["error"].clone()).map_or_else(|e| Err(e.into()), Err) - } else { - Ok(value["result"].clone()) - }; - if let Some(lease) = &p.unlock { - lease - .held - .set(result.as_ref().is_ok_and(|v| v["held"] == true)); - } - let _ = p.tx.send(result); - } else { - eprintln!("note: response with unknown ID"); - } -} - -/// Serve the main interface until its work ends or stdin reaches EOF. -#[cfg(target_os = "wasi")] -pub fn run(value: T) { - block_on(serve(value)); -} -#[cfg(not(target_os = "wasi"))] -pub fn run(_value: T) { - panic!("run requires WASI stdin/stdout"); -} - -#[cfg(target_os = "wasi")] -async fn serve(value: T) { - use std::{rc::Rc, task::Poll}; - use wstd::io::AsyncRead; - STATE.with(|s| { - *s.borrow_mut() = State { - active: true, - ..State::default() - } - }); - let value = Rc::new(value); - let mut input = wstd::io::stdin(); - let mut line = Vec::new(); - let mut started = false; - loop { - let read = async { - loop { - let mut byte = [0]; - if input.read(&mut byte).await? == 0 { - return Ok::<_, std::io::Error>(None); - } - if byte[0] == b'\n' { - return Ok(Some(std::mem::take(&mut line))); - } - line.push(byte[0]); - } - }; - let idle = futures::future::poll_fn(|cx| { - STATE.with(|s| { - let mut s = s.borrow_mut(); - s.wake = Some(cx.waker().clone()); - if started && (s.tasks == 0 || s.exit) { - Poll::Ready(()) - } else { - Poll::Pending - } - }) - }); - futures::pin_mut!(read, idle); - let bytes = match futures::future::select(read, idle).await { - futures::future::Either::Left((Ok(Some(b)), _)) => b, - futures::future::Either::Left((_, _)) => { - STATE.with(|s| { - let mut s = s.borrow_mut(); - s.stopping = true; - s.pending.clear(); - }); - value.stop().await; - break; - } - futures::future::Either::Right(_) => break, - }; - let Ok(msg) = serde_json::from_slice::(&bytes) else { - continue; - }; - if msg["jsonrpc"] != "2.0" { - continue; - } - if let Some(method) = msg["method"].as_str() { - if method == "core.notify" { - receive(msg); - continue; - } - started = true; - let value = value.clone(); - let method = method.to_owned(); - background(async move { - let ctx = Ctx::default(); - let result = if method == "rpc.discover" { - Ok(value.discover()) - } else { - value - .dispatch( - &method, - msg.get("params").cloned().unwrap_or(json!({})), - &ctx, - ) - .await - }; - if let Some(id) = msg.get("id") { - let response = match result { - Ok(v) => json!({"jsonrpc":"2.0","id":id,"result":v}), - Err(e) => json!({"jsonrpc":"2.0","id":id,"error":e}), - }; - let _ = send(response); - } - if ctx.exit.get() { - STATE.with(|s| s.borrow_mut().exit = true); - } - }); - } else { - receive(msg); - } - } - STATE.with(|s| { - let mut s = s.borrow_mut(); - s.active = false; - s.pending.clear(); - }); -} - -/// Create a shared OpenRPC document. -pub fn document(title: &str, version: &str) -> Value { - sausage_interface::Document::new(title, version).into_value() -} - -/// Build OpenRPC entries with separate deserialize and serialize contracts. -pub fn method(doc: &mut Value, name: &str, docs: &str) { - fn generator(serialize: bool) -> schemars::SchemaGenerator { - let mut s = schemars::generate::SchemaSettings::draft07(); - s.definitions_path = "/components/schemas".into(); - if serialize { - s.for_serialize() - } else { - s.for_deserialize() - } - .into_generator() - } - let mut p = generator(false); - let pref = serde_json::to_value(p.subschema_for::

()).unwrap(); - let mut defs = p.take_definitions(true); - for schema in defs.values_mut() { - normalize_enums(schema); - } - let ps = &defs[P::schema_name().as_ref()]; - let params:Vec<_>=ps["properties"].as_object().into_iter().flatten().map(|(name,s)| json!({"name":name,"description":s.get("description").cloned().unwrap_or(Value::Null),"required":ps["required"].as_array().is_some_and(|a|a.contains(&json!(name))),"schema":s})).collect(); - doc["components"]["schemas"] - .as_object_mut() - .unwrap() - .extend(defs); - let mut r = generator(true); - let mut rs = serde_json::to_value(r.subschema_for::()).unwrap(); - fn result_refs(value: &mut Value) { - match value { - Value::Object(map) => { - if let Some(Value::String(reference)) = map.get_mut("$ref") - && reference.starts_with("#/components/schemas/") - { - reference.push_str("Result"); - } - for child in map.values_mut() { - result_refs(child); - } - } - Value::Array(values) => { - for child in values { - result_refs(child); - } - } - _ => {} - } - } - result_refs(&mut rs); - for (name, mut schema) in r.take_definitions(true) { - normalize_enums(&mut schema); - result_refs(&mut schema); - doc["components"]["schemas"] - .as_object_mut() - .unwrap() - .insert(format!("{name}Result"), schema); - } - doc["methods"].as_array_mut().unwrap().push(json!({"name":name,"summary":docs.split("\n\n").next().unwrap_or_default(),"description":docs,"paramStructure":"by-name","params":params,"result":{"name":"result","schema":rs},"x-sausage-params":pref})); -} -// Schemars represents documented unit variants as oneOf/const. Export string enums instead. -fn normalize_enums(value: &mut Value) { - match value { - Value::Object(map) => { - if let Some(variants) = map.get("oneOf").and_then(Value::as_array) { - let values: Option> = variants - .iter() - .map(|v| { - if let Some(s) = v.get("const").filter(|v| v.is_string()) { - Some(vec![s.clone()]) - } else { - v.get("enum") - .and_then(Value::as_array) - .filter(|a| a.iter().all(Value::is_string)) - .cloned() - } - }) - .collect::>>>() - .map(|v| v.into_iter().flatten().collect()); - if let Some(values) = values { - map.remove("oneOf"); - map.insert("type".into(), json!("string")); - map.insert("enum".into(), json!(values)); - } - } - for child in map.values_mut() { - normalize_enums(child); - } - } - Value::Array(values) => { - for child in values { - normalize_enums(child); - } - } - _ => {} - } -} - -/// Empty parameters for methods without arguments. -#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)] -pub struct Empty {} -impl SchemaType for Empty {} -impl Params for Empty {} +pub use inventory; +pub use sausage_plugin_macros::{import, on_retired, on_stop, rpc, schema, test}; +pub use transport::*; +pub use {schemars, serde, serde_json}; diff --git a/sdk/src/lines.rs b/sdk/src/lines.rs new file mode 100644 index 0000000..8236b90 --- /dev/null +++ b/sdk/src/lines.rs @@ -0,0 +1,29 @@ +//! Buffered newline framing shared by stdin and endpoint connections. +#[cfg(target_os = "wasi")] +#[derive(Default)] +pub(crate) struct Lines { + bytes: Vec, +} +#[cfg(target_os = "wasi")] +impl Lines { + pub(crate) async fn next( + &mut self, + input: &mut impl wstd::io::AsyncRead, + ) -> std::io::Result>> { + loop { + if let Some(end) = self.bytes.iter().position(|b| *b == b'\n') { + return Ok(Some(self.bytes.drain(..=end).collect())); + } + let mut buffer = [0; 8192]; + let count = input.read(&mut buffer).await?; + if count == 0 { + return Ok(if self.bytes.is_empty() { + None + } else { + Some(std::mem::take(&mut self.bytes)) + }); + } + self.bytes.extend_from_slice(&buffer[..count]); + } + } +} diff --git a/sdk/src/locks.rs b/sdk/src/locks.rs new file mode 100644 index 0000000..72c3f83 --- /dev/null +++ b/sdk/src/locks.rs @@ -0,0 +1,65 @@ +//! Advisory lock acquisition and guard lifetime. +use super::*; +/// An advisory lock acquisition. +pub struct Lock { + path: String, + wait: bool, +} +/// Acquire an advisory lock on a path. +pub fn lock(path: impl AsRef) -> Lock { + Lock { + path: path.as_ref().to_string_lossy().into_owned(), + wait: true, + } +} +impl Lock { + /// Return immediately, without waiting for a held lock. + pub fn try_now(mut self) -> Self { + self.wait = false; + self + } +} +/// Releases its lock on drop, unless kept until process exit. +pub struct Guard { + path: Option, +} +impl Guard { + /// Hold this lock until process exit. + pub fn keep(mut self) { + self.path.take(); + } + /// Report whether this guard holds the lock. + pub fn held(&self) -> bool { + self.path.is_some() + } +} +impl Drop for Guard { + fn drop(&mut self) { + if let Some(path) = self.path.take() { + let _ = notify("core.unlock", json!({"path":path})); + } + } +} +impl IntoFuture for Lock { + type Output = Result; + type IntoFuture = LocalBoxFuture<'static, Self::Output>; + fn into_future(self) -> Self::IntoFuture { + Box::pin(async move { + let lease = std::rc::Rc::new(Lease { + path: self.path.clone(), + held: std::cell::Cell::new(false), + }); + let mut c = call("core.lock", json!({"path":self.path,"wait":self.wait})); + c.unlock = Some(lease.clone()); + let result = c.await?; + lease.held.set(false); + Ok(Guard { + path: if result["held"] == true { + Some(self.path) + } else { + None + }, + }) + }) + } +} diff --git a/sdk/src/transport.rs b/sdk/src/transport.rs new file mode 100644 index 0000000..b8c75dd --- /dev/null +++ b/sdk/src/transport.rs @@ -0,0 +1,379 @@ +//! Stdin routing, stdout framing, and process lifetime. +#[cfg(target_os = "wasi")] +use crate::lines::Lines; +use crate::*; +use futures::{channel::oneshot, future::LocalBoxFuture}; +use serde_json::{Value, json}; +use std::{ + cell::RefCell, + collections::HashMap, + future::{Future, IntoFuture}, + io::Write, + path::Path, +}; +#[path = "calls.rs"] +mod calls; +#[path = "locks.rs"] +mod locks; +pub use calls::*; +pub use locks::*; +/// Generated main-interface dispatch. +pub trait Interface: 'static { + /// Describe this interface without constructing its value. + fn describe() -> Value + where + Self: Sized; + /// Describe only this type's methods, without endpoint declarations. + fn methods_document() -> Value + where + Self: Sized; + /// List the declared endpoint types and their names. + fn endpoint_types(&self) -> Vec<(std::any::TypeId, String)> { + vec![] + } + /// Describe this interface as OpenRPC. + fn discover(&self) -> Value; + /// Dispatch a request to a generated handler. + fn dispatch<'a>( + &'a self, + method: &'a str, + params: Value, + ctx: &'a Ctx, + ) -> LocalBoxFuture<'a, Result>; + /// Run the EOF hook. + fn stop(&self) -> LocalBoxFuture<'_, ()> { + Box::pin(async {}) + } +} + +/// Context of an incoming call. +#[derive(Default)] +pub struct Ctx { + pub(crate) exit: std::cell::Cell, + pub(crate) sink: Option>, +} +impl Ctx { + /// Send progress to the caller of this request. + pub fn notify(&self, name: &str, params: impl serde::Serialize) -> Result<()> { + let params = json!({"name":name,"params":params}); + if let Some(sink) = &self.sink { + sink.unbounded_send(json!({"jsonrpc":"2.0","method":"core.notify","params":params})) + .map_err(|_| Error::new(-32002, "endpoint closed")) + } else { + notify("core.notify", params) + } + } + /// End the process after this request receives its response. + pub fn exit_after_response(&self) { + self.exit.set(true); + } +} + +type Notify = Box; +struct Pending { + tx: oneshot::Sender>, + notify: Option, + unlock: Option>, +} +struct Lease { + path: String, + held: std::cell::Cell, +} +impl Drop for Lease { + fn drop(&mut self) { + if self.held.get() { + let _ = notify("core.unlock", json!({"path":self.path})); + } + } +} +#[derive(Default)] +struct State { + active: bool, + endpoints: HashMap, + stopping: bool, + next: u64, + pending: HashMap, + #[cfg(target_os = "wasi")] + tasks: usize, + #[cfg(target_os = "wasi")] + wake: Option, + #[cfg(target_os = "wasi")] + exit: bool, + #[cfg(target_os = "wasi")] + listeners: HashMap, +} +thread_local! { static STATE: RefCell = RefCell::new(State::default()); } +pub(crate) fn endpoint_name() -> Result { + STATE + .with(|s| { + s.borrow() + .endpoints + .get(&std::any::TypeId::of::()) + .cloned() + }) + .ok_or_else(|| { + Error::new( + -32602, + "endpoint type is not declared by the main interface", + ) + }) +} +#[cfg(target_os = "wasi")] +pub(crate) fn endpoint_task(future: impl Future + 'static) { + let (handle, registration) = futures::future::AbortHandle::new_pair(); + let id = STATE.with(|s| { + let mut s = s.borrow_mut(); + s.next += 1; + let id = s.next; + s.listeners.insert(id, handle); + id + }); + background(async move { + let _ = futures::future::Abortable::new(future, registration).await; + STATE.with(|s| s.borrow_mut().listeners.remove(&id)); + }); +} +#[cfg(target_os = "wasi")] +pub(crate) fn exit() { + STATE.with(|s| { + let mut s = s.borrow_mut(); + s.exit = true; + if let Some(w) = s.wake.take() { + w.wake(); + } + }); +} +fn send(value: Value) -> Result<()> { + let bytes = format!("\n{value}\n").into_bytes(); + let mut stdout = std::io::stdout().lock(); + stdout.write_all(&bytes)?; + stdout.flush()?; + Ok(()) +} +fn notify(method: &str, params: Value) -> Result<()> { + if !STATE.with(|s| s.borrow().active) { + return Err(Error::new(-32002, "no core transport")); + } + send(json!({"jsonrpc":"2.0","method":method,"params":params})) +} + +#[cfg(target_os = "wasi")] +/// Execute a future with the SDK runtime. +pub fn block_on(f: F) -> F::Output { + wstd::runtime::block_on(f) +} +#[cfg(not(target_os = "wasi"))] +/// Execute a future with the SDK runtime. +pub fn block_on(f: F) -> F::Output { + futures::executor::block_on(f) +} + +/// Run an SDK test with a core transport when the WASI runner supplies one. +#[cfg(target_os = "wasi")] +/// Execute an async test with the available core transport. +pub fn test_run(f: F) -> F::Output { + block_on(async { + STATE.with(|s| s.borrow_mut().active = true); + let reader = wstd::runtime::spawn(async { + let mut input = wstd::io::stdin(); + let mut lines = Lines::default(); + while let Ok(Some(line)) = lines.next(&mut input).await { + if let Ok(v) = serde_json::from_slice::(&line) + && v["jsonrpc"] == "2.0" + { + receive(v); + } + } + STATE.with(|s| { + let mut s = s.borrow_mut(); + s.stopping = true; + s.pending.clear(); + }); + }); + let result = f.await; + drop(reader); + STATE.with(|s| { + let mut s = s.borrow_mut(); + s.active = false; + s.pending.clear(); + }); + result + }) +} +#[cfg(not(target_os = "wasi"))] +/// Execute an async test with the available core transport. +pub fn test_run(f: F) -> F::Output { + block_on(f) +} + +/// Register a task that keeps the process alive. +#[cfg(target_os = "wasi")] +/// Keep this process alive while a future runs. +pub fn background(f: impl Future + 'static) { + STATE.with(|s| s.borrow_mut().tasks += 1); + wstd::runtime::spawn(async move { + f.await; + STATE.with(|s| { + let mut s = s.borrow_mut(); + s.tasks -= 1; + if let Some(w) = s.wake.take() { + w.wake(); + } + }); + }) + .detach(); +} +#[cfg(not(target_os = "wasi"))] +/// Keep this process alive while a future runs. +pub fn background(_f: impl Future + 'static) { + panic!("background requires the WASI runtime"); +} + +#[doc(hidden)] +pub fn receive(value: Value) { + if value["method"] == "core.notify" { + if let Some(id) = value["params"]["call"].as_u64() { + let callback = STATE.with(|s| { + s.borrow_mut() + .pending + .get_mut(&id) + .and_then(|p| p.notify.take()) + }); + if let Some(mut f) = callback { + f( + value["params"]["name"].as_str().unwrap_or_default(), + value["params"]["params"].clone(), + ); + STATE.with(|s| { + if let Some(p) = s.borrow_mut().pending.get_mut(&id) { + p.notify = Some(f); + } + }); + } + } + return; + } + let pending = value["id"] + .as_u64() + .and_then(|id| STATE.with(|s| s.borrow_mut().pending.remove(&id))); + if let Some(p) = pending { + let result = if value.get("error").is_some() { + serde_json::from_value(value["error"].clone()).map_or_else(|e| Err(e.into()), Err) + } else { + Ok(value["result"].clone()) + }; + if let Some(lease) = &p.unlock { + lease + .held + .set(result.as_ref().is_ok_and(|v| v["held"] == true)); + } + let _ = p.tx.send(result); + } else { + eprintln!("note: response with unknown ID {}", value["id"]); + } +} + +/// Serve the main interface until its work ends or stdin reaches EOF. +#[cfg(target_os = "wasi")] +/// Serve the main interface over stdin and stdout. +pub fn run(value: T) { + block_on(serve(value)); +} +#[cfg(not(target_os = "wasi"))] +/// Serve the main interface over stdin and stdout. +pub fn run(_value: T) { + panic!("run requires WASI stdin/stdout"); +} + +#[cfg(target_os = "wasi")] +async fn serve(value: T) { + use std::{rc::Rc, task::Poll}; + STATE.with(|s| { + *s.borrow_mut() = State { + active: true, + endpoints: value.endpoint_types().into_iter().collect(), + ..State::default() + } + }); + let value = Rc::new(value); + let mut input = wstd::io::stdin(); + let mut lines = Lines::default(); + let mut started = false; + loop { + let read = lines.next(&mut input); + let idle = futures::future::poll_fn(|cx| { + STATE.with(|s| { + let mut s = s.borrow_mut(); + s.wake = Some(cx.waker().clone()); + if started && (s.tasks == 0 || s.exit) { + Poll::Ready(()) + } else { + Poll::Pending + } + }) + }); + futures::pin_mut!(read, idle); + let bytes = match futures::future::select(read, idle).await { + futures::future::Either::Left((Ok(Some(b)), _)) => b, + futures::future::Either::Left((_, _)) => { + STATE.with(|s| { + let mut s = s.borrow_mut(); + s.stopping = true; + s.pending.clear(); + for (_, listener) in s.listeners.drain() { + listener.abort(); + } + }); + value.stop().await; + break; + } + futures::future::Either::Right(_) => break, + }; + let Ok(msg) = serde_json::from_slice::(&bytes) else { + continue; + }; + if msg["jsonrpc"] != "2.0" { + continue; + } + if let Some(method) = msg["method"].as_str() { + if method == "core.notify" { + receive(msg); + continue; + } + started = true; + let value = value.clone(); + let method = method.to_owned(); + background(async move { + let ctx = Ctx::default(); + let result = if method == "rpc.discover" { + Ok(value.discover()) + } else { + value + .dispatch( + &method, + msg.get("params").cloned().unwrap_or(json!({})), + &ctx, + ) + .await + }; + if let Some(id) = msg.get("id") { + let response = match result { + Ok(v) => json!({"jsonrpc":"2.0","id":id,"result":v}), + Err(e) => json!({"jsonrpc":"2.0","id":id,"error":e}), + }; + let _ = send(response); + } + if ctx.exit.get() { + STATE.with(|s| s.borrow_mut().exit = true); + } + }); + } else { + receive(msg); + } + } + STATE.with(|s| { + let mut s = s.borrow_mut(); + s.active = false; + s.pending.clear(); + }); +} diff --git a/sdk/tests/plugins/consumer/Cargo.toml b/sdk/tests/plugins/consumer/Cargo.toml new file mode 100644 index 0000000..89b5efe --- /dev/null +++ b/sdk/tests/plugins/consumer/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "sdk-consumer" +version = "0.1.0" +edition = "2024" +[workspace] +[dependencies] +sausage-plugin = { path = "../../.." } diff --git a/sdk/tests/plugins/consumer/imports/probe.json b/sdk/tests/plugins/consumer/imports/probe.json new file mode 100644 index 0000000..d920257 --- /dev/null +++ b/sdk/tests/plugins/consumer/imports/probe.json @@ -0,0 +1,268 @@ +{ + "components": { + "schemas": { + "Add": { + "description": "The values to add.", + "properties": { + "a": { + "description": "The first number.", + "format": "int32", + "type": "integer" + }, + "b": { + "description": "The second number.", + "format": "int32", + "type": "integer" + } + }, + "required": [ + "a", + "b" + ], + "type": "object" + }, + "Echo": { + "description": "A large payload.", + "properties": { + "text": { + "description": "The text to return unchanged.", + "type": "string" + } + }, + "required": [ + "text" + ], + "type": "object" + }, + "Empty": { + "description": "Empty parameters for methods without arguments.", + "type": "object" + } + } + }, + "info": { + "title": "sdk-probe", + "version": "0.1.0" + }, + "methods": [ + { + "description": "Publish a control endpoint until EOF.", + "name": "serve", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "null" + } + }, + "summary": "Publish a control endpoint until EOF.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + }, + { + "description": "Add two numbers.\n\nThe result is their sum.", + "name": "add", + "paramStructure": "by-name", + "params": [ + { + "description": "The first number.", + "name": "a", + "required": true, + "schema": { + "description": "The first number.", + "format": "int32", + "type": "integer" + } + }, + { + "description": "The second number.", + "name": "b", + "required": true, + "schema": { + "description": "The second number.", + "format": "int32", + "type": "integer" + } + } + ], + "result": { + "name": "result", + "schema": { + "format": "int32", + "type": "integer" + } + }, + "summary": "Add two numbers.", + "x-sausage-params": { + "$ref": "#/components/schemas/Add" + } + }, + { + "description": "Echo a payload larger than a host I/O buffer.", + "name": "echo", + "paramStructure": "by-name", + "params": [ + { + "description": "The text to return unchanged.", + "name": "text", + "required": true, + "schema": { + "description": "The text to return unchanged.", + "type": "string" + } + } + ], + "result": { + "name": "result", + "schema": { + "type": "string" + } + }, + "summary": "Echo a payload larger than a host I/O buffer.", + "x-sausage-params": { + "$ref": "#/components/schemas/Echo" + } + }, + { + "description": "Return an internal serialization error.", + "name": "serde_error", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "string" + } + }, + "summary": "Return an internal serialization error.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + }, + { + "description": "Return a deliberate error.", + "name": "fail", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "string" + } + }, + "summary": "Return a deliberate error.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + }, + { + "description": "Exercise calls, notifications, and advisory locks.", + "name": "exercise", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "string" + } + }, + "summary": "Exercise calls, notifications, and advisory locks.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + }, + { + "description": "Drop an acquisition before the core grants its lock.", + "name": "late", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "string" + } + }, + "summary": "Drop an acquisition before the core grants its lock.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + }, + { + "description": "Wait until EOF, then run the stop hook.", + "name": "wait", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "string" + } + }, + "summary": "Wait until EOF, then run the stop hook.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + }, + { + "description": "Keep the process alive with a background call.", + "name": "start", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "string" + } + }, + "summary": "Keep the process alive with a background call.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + } + ], + "openrpc": "1.2.6", + "x-sausage-endpoints": { + "control": { + "methods": [ + { + "description": "Report progress and return a stable counter.", + "name": "stats", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "format": "uint64", + "minimum": 0, + "type": "integer" + } + }, + "summary": "Report progress and return a stable counter.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + }, + { + "description": "End the endpoint process after the response.", + "name": "stop", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "type": "null" + } + }, + "summary": "End the endpoint process after the response.", + "x-sausage-params": { + "$ref": "#/components/schemas/Empty" + } + } + ] + } + }, + "x-sausage-grace": 12, + "x-sausage-imports": {} +} \ No newline at end of file diff --git a/sdk/tests/plugins/consumer/src/main.rs b/sdk/tests/plugins/consumer/src/main.rs new file mode 100644 index 0000000..afd8c18 --- /dev/null +++ b/sdk/tests/plugins/consumer/src/main.rs @@ -0,0 +1,25 @@ +//! Consumer of the probe's actual discovered interface. +use sausage_plugin::{Result, rpc}; +sausage_plugin::import!("probe"); +struct Consumer; +#[rpc] +impl Consumer { + /// Route an imported call through the core. + async fn run(&self) -> Result { + Ok(probe::add(probe::Add { a: 2, b: 3 }) + .detach() + .on_notify(|name, _| eprintln!("typed progress: {name}")) + .await? as i32) + } + /// Connect to the provider's endpoint and receive its notification. + async fn endpoint(&self) -> Result { + let control = probe::control::connect("/endpoint").await?; + control + .stats() + .on_notify(|name, _| eprintln!("endpoint notification: {name}")) + .await + } +} +fn main() { + sausage_plugin::run(Consumer) +} diff --git a/sdk/tests/plugins/probe/src/main.rs b/sdk/tests/plugins/probe/src/main.rs index 668f281..89e2824 100644 --- a/sdk/tests/plugins/probe/src/main.rs +++ b/sdk/tests/plugins/probe/src/main.rs @@ -16,14 +16,32 @@ struct Retired { /// The previous hash. previous: String, } -#[rpc(grace_seconds = 12)] +/// A large payload. +#[schema] +struct Echo { + /// The text to return unchanged. + text: String, +} +#[rpc(grace_seconds = 12,endpoints(control=Control))] impl Probe { + /// Publish a control endpoint until EOF. + async fn serve(&self) -> Result<()> { + sausage_plugin::listen("/endpoint", std::rc::Rc::new(Control)).await + } /// Add two numbers. /// /// The result is their sum. fn add(&self, p: Add) -> Result { Ok(p.a + p.b) } + /// Echo a payload larger than a host I/O buffer. + fn echo(&self, p: Echo) -> Result { + Ok(p.text) + } + /// Return an internal serialization error. + fn serde_error(&self) -> Result { + Ok(sausage_plugin::serde_json::from_str::("invalid")?) + } /// Return a deliberate error. fn fail(&self) -> Result { Err(sausage_plugin::Error::new(-32010, "deliberate")) @@ -109,6 +127,20 @@ async fn join( }) .await } +struct Control; +#[rpc] +impl Control { + /// Report progress and return a stable counter. + fn stats(&self, ctx: &Ctx) -> Result { + ctx.notify("endpoint-progress", json!({"value":7}))?; + Ok(7) + } + /// End the endpoint process after the response. + fn stop(&self, ctx: &Ctx) -> Result<()> { + ctx.exit_after_response(); + Ok(()) + } +} fn main() { sausage_plugin::run(Probe) } diff --git a/sdk/tests/sdk.rs b/sdk/tests/sdk.rs index 0b4952e..b51ae55 100644 --- a/sdk/tests/sdk.rs +++ b/sdk/tests/sdk.rs @@ -25,8 +25,12 @@ enum Choice { Second, } struct Plugin; -#[rpc] +#[rpc(endpoints(control=Plugin))] impl Plugin { + /// Return the same named object without a contract change. + fn identity(&self, p: Add) -> Result { + Ok(p) + } /// Choose the first value. fn choice(&self) -> Result { Ok(Choice::First) @@ -45,6 +49,23 @@ impl Plugin { #[test] fn discovery_and_dispatch() { let doc = Plugin.discover(); + assert!( + doc["x-sausage-endpoints"]["control"]["methods"] + .as_array() + .unwrap() + .iter() + .any(|m| m["name"] == "add") + ); + assert!(doc["components"]["schemas"].get("AddResult").is_none()); + assert_eq!( + doc["methods"] + .as_array() + .unwrap() + .iter() + .find(|m| m["name"] == "identity") + .unwrap()["result"]["schema"]["$ref"], + "#/components/schemas/Add" + ); assert!( doc["components"]["schemas"]["Defaulted"] .get("required") @@ -55,8 +76,8 @@ fn discovery_and_dispatch() { json!(["value"]) ); assert_eq!( - doc["components"]["schemas"]["ChoiceResult"]["enum"], - json!(["First", "Second"]) + doc["components"]["schemas"]["Choice"]["oneOf"][0]["description"], + json!("The first choice.") ); let add = doc["methods"] .as_array() @@ -76,7 +97,16 @@ fn discovery_and_dispatch() { .unwrap(), 5 ); - assert!(sausage_plugin::block_on(Plugin.dispatch("add", json!({}), &Ctx::default())).is_err()); + assert_eq!( + sausage_plugin::block_on(Plugin.dispatch("add", json!({}), &Ctx::default())) + .unwrap_err() + .code, + -32602 + ); + let internal: sausage_plugin::Error = sausage_plugin::serde_json::from_str::("bad") + .unwrap_err() + .into(); + assert_eq!(internal.code, -32603); } #[sausage_plugin::test] async fn missing_core_fails_without_wait() { diff --git a/sdk/tests/ui/params.stderr b/sdk/tests/ui/params.stderr index 24ff651..a028ddb 100644 --- a/sdk/tests/ui/params.stderr +++ b/sdk/tests/ui/params.stderr @@ -5,12 +5,12 @@ error[E0277]: RPC params must be a #[schema] struct with named fields | ^^^^^^ the trait `Params` is not implemented for `std::string::String` | help: the trait `Params` is implemented for `sausage_plugin::Empty` - --> src/lib.rs + --> src/discovery.rs | | impl Params for Empty {} | ^^^^^^^^^^^^^^^^^^^^^ -note: required by a bound in `method` - --> src/lib.rs +note: required by a bound in `sausage_plugin::method` + --> src/discovery.rs | | pub fn method(doc: &mut Value, name: &str, docs: &str) { | ^^^^^^ required by this bound in `method` @@ -22,7 +22,7 @@ error[E0277]: RPC params must be a #[schema] struct with named fields | ^^^^^^ the trait `Params` is not implemented for `std::string::String` | help: the trait `Params` is implemented for `sausage_plugin::Empty` - --> src/lib.rs + --> src/discovery.rs | | impl Params for Empty {} | ^^^^^^^^^^^^^^^^^^^^^ diff --git a/sdk/tests/wasi.rs b/sdk/tests/wasi.rs index b75c1f3..116d0b1 100644 --- a/sdk/tests/wasi.rs +++ b/sdk/tests/wasi.rs @@ -34,7 +34,16 @@ impl Process { } fn start(wasm: &Path, dir: &Path) -> Process { let mut child = Command::new("wasmtime") - .args(["run", "--dir"]) + .args([ + "run", + "-S", + "inherit-network=y", + "--env", + "SAUSAGE_PID=probe-00000000-0000-7000-8000-000000000000", + "--env", + "SAUSAGE_PLUGIN=probe", + "--dir", + ]) .arg(format!("{}::/", dir.display())) .arg(wasm) .stdin(Stdio::piped()) @@ -72,7 +81,12 @@ fn protocol() { let d = p.read(); assert_eq!(d["result"]["x-sausage-grace"], 12); assert_eq!( - d["result"]["methods"][0]["description"], + d["result"]["methods"] + .as_array() + .unwrap() + .iter() + .find(|m| m["name"] == "add") + .unwrap()["description"], "Add two numbers.\n\nThe result is their sum." ); assert!(p.child.wait().unwrap().success()); @@ -81,6 +95,18 @@ fn protocol() { assert_eq!(p.read()["result"], 6); assert!(p.child.wait().unwrap().success()); let mut p = start(&wasm, &dir); + p.request("echo", json!({"text":"x".repeat(70_000)})); + assert_eq!(p.read()["result"], "x".repeat(70_000)); + assert!(p.child.wait().unwrap().success()); + let mut p = start(&wasm, &dir); + p.request("echo", json!({"text":5})); + assert_eq!(p.read()["error"]["code"], -32602); + assert!(p.child.wait().unwrap().success()); + let mut p = start(&wasm, &dir); + p.request("serde_error", json!({})); + assert_eq!(p.read()["error"]["code"], -32603); + assert!(p.child.wait().unwrap().success()); + let mut p = start(&wasm, &dir); p.request("fail", json!({})); assert_eq!(p.read()["error"]["code"], -32010); assert!(p.child.wait().unwrap().success()); @@ -107,7 +133,7 @@ fn protocol() { assert!(p.child.wait().unwrap().success()); let mut err = String::new(); std::io::Read::read_to_string(p.child.stderr.as_mut().unwrap(), &mut err).unwrap(); - assert!(err.contains("unknown ID")); + assert!(err.contains("unknown ID 9876")); assert!(err.contains("notification: tick")); let mut p = start(&wasm, &dir); p.request("rpc.retired", json!({"previous":"hash"})); @@ -143,5 +169,77 @@ fn protocol() { assert_eq!(outgoing["method"], "other.background"); p.answer(&outgoing, json!({})); assert!(p.child.wait().unwrap().success()); + let provider_doc = &d["result"]; + let imports = root.join("tests/plugins/consumer/imports"); + std::fs::create_dir_all(&imports).unwrap(); + std::fs::write( + imports.join("probe.json"), + sausage_plugin::serde_json::to_vec_pretty(provider_doc).unwrap(), + ) + .unwrap(); + let generated = sausage_interface::client_rs("probe", provider_doc).unwrap(); + std::fs::write(target.join("client.rs"), generated).unwrap(); + assert!( + Command::new("cargo") + .env("CARGO_TARGET_DIR", &target) + .args(["build", "--offline", "--manifest-path"]) + .arg(root.join("tests/plugins/consumer/Cargo.toml")) + .args(["--target", "wasm32-wasip2"]) + .status() + .unwrap() + .success() + ); + let consumer = target.join("wasm32-wasip2/debug/sdk-consumer.wasm"); + let mut client = start(&consumer, &dir); + client.request("rpc.discover", json!({})); + let consumer_doc = client.read()["result"].clone(); + assert!(client.child.wait().unwrap().success()); + let imported = &consumer_doc["x-sausage-imports"]["probe"]; + assert!(sausage_interface::compat::check(imported, provider_doc).is_empty()); + let mut broken = provider_doc.clone(); + broken["methods"] + .as_array_mut() + .unwrap() + .retain(|m| m["name"] != "add"); + assert!(!sausage_interface::compat::check(imported, &broken).is_empty()); + let mut client = start(&consumer, &dir); + client.request("run", json!({})); + let outgoing = client.read(); + assert_eq!(outgoing["method"], "core.call"); + assert_eq!(outgoing["params"]["method"], "probe.add"); + let mut provider = start(&wasm, &dir); + provider.request("add", outgoing["params"]["params"].clone()); + let reply = provider.read(); + assert!(provider.child.wait().unwrap().success()); + client.send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":outgoing["id"],"name":"typed","params":{}}})); + client.answer(&outgoing, reply["result"].clone()); + assert_eq!(client.read()["result"], 5); + assert!(client.child.wait().unwrap().success()); + let mut provider = start(&wasm, &dir); + provider.request("serve", json!({})); + assert_eq!(provider.read()["result"], Value::Null); + let mut client = start(&consumer, &dir); + client.request("endpoint", json!({})); + assert_eq!(client.read()["result"], 7); + assert!(client.child.wait().unwrap().success()); + let mut stderr = String::new(); + std::io::Read::read_to_string(client.child.stderr.as_mut().unwrap(), &mut stderr).unwrap(); + assert!(stderr.contains("endpoint notification: endpoint-progress")); + let path = dir.join("endpoint"); + let original = std::fs::read_to_string(&path).unwrap(); + std::fs::write(&path, original.replace("probe-00000000", "probe-11111111")).unwrap(); + let mut client = start(&consumer, &dir); + client.request("endpoint", json!({})); + assert!( + client.read()["error"]["message"] + .as_str() + .unwrap() + .contains("identity mismatch") + ); + assert!(client.child.wait().unwrap().success()); + std::fs::write(path, original).unwrap(); + drop(provider.input.take()); + assert_eq!(provider.read()["params"]["name"], "stopped"); + assert!(provider.child.wait().unwrap().success()); std::fs::remove_dir_all(dir).unwrap(); }