sausage-plugin part 1: run, #[rpc], #[schema], rpc.discover, calls, locks, Ctx, hooks

Checkpoint. Known issues, fixed in part 2: short writes on stdout,
byte-wise stdin reads, Result suffix on all result schemas, lost enum
variant docs, serde errors as invalid params, missing item docs.
This commit is contained in:
2026-10-23 00:05:37 +01:00
parent 4e41061ed4
commit 01712e745a
30 changed files with 1722 additions and 93 deletions
+633 -1
View File
@@ -1 +1,633 @@
//! SDK for Sausage plugins.
//! 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<T> = std::result::Result<T, Error>;
/// 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<Value>,
}
impl Error {
pub fn new(code: i64, message: impl Into<String>) -> 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<std::io::Error> for Error {
fn from(e: std::io::Error) -> Self {
Self::new(-32603, e.to_string())
}
}
impl From<serde_json::Error> 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<T: SchemaType> SchemaType for Vec<T> {}
impl<T: SchemaType> SchemaType for Option<T> {}
/// 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<Value>>;
fn stop(&self) -> LocalBoxFuture<'_, ()> {
Box::pin(async {})
}
}
/// Context of an incoming call.
#[derive(Default)]
pub struct Ctx {
exit: std::cell::Cell<bool>,
}
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<dyn FnMut(&str, Value)>;
struct Pending {
tx: oneshot::Sender<Result<Value>>,
notify: Option<Notify>,
unlock: Option<std::rc::Rc<Lease>>,
}
struct Lease {
path: String,
held: std::cell::Cell<bool>,
}
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<u64, Pending>,
#[cfg(target_os = "wasi")]
tasks: usize,
#[cfg(target_os = "wasi")]
wake: Option<std::task::Waker>,
#[cfg(target_os = "wasi")]
exit: bool,
}
thread_local! { static STATE: RefCell<State> = 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<Notify>,
unlock: Option<std::rc::Rc<Lease>>,
}
pub fn call(method: impl Into<String>, 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<Value>;
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<Path>) -> 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<String>,
}
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<Guard>;
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<Path>, 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: Future>(f: F) -> F::Output {
wstd::runtime::block_on(f)
}
#[cfg(not(target_os = "wasi"))]
pub fn block_on<F: Future>(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: Future>(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::<Value>(&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: Future>(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<Output = ()> + '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<Output = ()> + '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");
}
}
/// Serve the main interface until its work ends or stdin reaches EOF.
#[cfg(target_os = "wasi")]
pub fn run<T: Interface>(value: T) {
block_on(serve(value));
}
#[cfg(not(target_os = "wasi"))]
pub fn run<T: Interface>(_value: T) {
panic!("run requires WASI stdin/stdout");
}
#[cfg(target_os = "wasi")]
async fn serve<T: Interface>(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::<Value>(&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<P: Params, R: SchemaType>(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::<P>()).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::<R>()).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<Vec<Value>> = 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::<Option<Vec<Vec<Value>>>>()
.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 {}