SAU-357 implementer
9cbd4f94aa
Ignore legacy grace metadata in both activation paths. Keep superseded delivery, admission ordering, discovery exclusion and the separate EOF stop budget. Document the uniform policy and cover both windows and opt-in drain. Card: SAU-357
400 lines
14 KiB
Rust
400 lines
14 KiB
Rust
//! Exercise SDK plugins through the real host, without a fake core.
|
|
use sausage::{Host, world::World};
|
|
use serde_json::{Value, json};
|
|
use std::{path::PathBuf, sync::OnceLock, time::Duration};
|
|
|
|
fn fixtures() -> &'static PathBuf {
|
|
static TARGET: OnceLock<PathBuf> = OnceLock::new();
|
|
TARGET.get_or_init(|| {
|
|
for (package, binary) in [
|
|
("examples/notes", "notes"),
|
|
("examples/consumer", "consumer"),
|
|
("examples/www", "www"),
|
|
("examples/secrets", "secrets"),
|
|
("sdk/tests/plugins/consumer", "sdk-consumer"),
|
|
] {
|
|
super::common::build(package, binary);
|
|
}
|
|
super::common::sdk_probe().parent().unwrap().to_owned()
|
|
})
|
|
}
|
|
struct TestWorld(PathBuf);
|
|
impl TestWorld {
|
|
fn new() -> Self {
|
|
let path = std::env::temp_dir().join(format!("sausage-sdk-{}", uuid::Uuid::now_v7()));
|
|
let world = World::create(&path).unwrap();
|
|
for (plugin, binary) in [
|
|
("notes", "notes"),
|
|
("consumer", "consumer"),
|
|
("www", "www"),
|
|
("secrets", "secrets"),
|
|
("probe", "sdk-probe"),
|
|
("typed", "sdk-consumer"),
|
|
] {
|
|
std::fs::create_dir_all(world.plugin_dir(plugin)).unwrap();
|
|
std::fs::copy(
|
|
fixtures().join(format!("{binary}.wasm")),
|
|
world.plugin_dir(plugin).join("plugin.wasm"),
|
|
)
|
|
.unwrap();
|
|
}
|
|
Self(path)
|
|
}
|
|
async fn start(&self) -> Host {
|
|
Host::start_with_cache(
|
|
&self.0,
|
|
std::path::Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")),
|
|
)
|
|
.await
|
|
.unwrap()
|
|
}
|
|
}
|
|
impl Drop for TestWorld {
|
|
fn drop(&mut self) {
|
|
let _ = std::fs::remove_dir_all(&self.0);
|
|
}
|
|
}
|
|
async fn call(host: &Host, method: &str, params: Value) -> Value {
|
|
tokio::time::timeout(Duration::from_secs(60), host.call(method, params))
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
}
|
|
|
|
/// Concurrent read-modify-write calls retain every note under the SDK lock.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn notes_lock_preserves_parallel_additions() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(
|
|
call(&host, "notes.add", json!({"text":"first"})).await["count"],
|
|
1
|
|
);
|
|
let calls = (0..12)
|
|
.map(|n| {
|
|
let h = host.clone();
|
|
tokio::spawn(
|
|
async move { call(&h, "notes.add", json!({"text":format!("note-{n}")})).await },
|
|
)
|
|
})
|
|
.collect::<Vec<_>>();
|
|
for task in calls {
|
|
task.await.unwrap();
|
|
}
|
|
let notes = call(&host, "notes.list", json!({})).await;
|
|
assert_eq!(notes.as_array().unwrap().len(), 13);
|
|
assert_eq!(notes[0], "first");
|
|
for n in 0..12 {
|
|
assert!(
|
|
notes
|
|
.as_array()
|
|
.unwrap()
|
|
.contains(&json!(format!("note-{n}")))
|
|
);
|
|
}
|
|
host.stop().await;
|
|
}
|
|
/// The example consumer invokes notes through its generated import.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn generated_notes_import() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(
|
|
call(&host, "consumer.note", json!({"text":"typed"})).await,
|
|
1
|
|
);
|
|
assert_eq!(call(&host, "notes.list", json!({})).await, json!(["typed"]));
|
|
host.stop().await;
|
|
}
|
|
/// Discovery retains the complete document and satisfies the consumer contract.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn discovery_is_compatible() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let actual = call(&host, "notes.rpc.discover", json!({})).await;
|
|
let consumer = call(&host, "consumer.rpc.discover", json!({})).await;
|
|
let expected: Value = serde_json::from_str(include_str!(
|
|
"../../../examples/consumer/imports/notes.json"
|
|
))
|
|
.unwrap();
|
|
assert_eq!(actual, expected);
|
|
assert_eq!(actual["openrpc"], "1.2.6");
|
|
assert!(actual["components"]["schemas"]["AddParams"].is_object());
|
|
assert!(
|
|
actual["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.find(|m| m["name"] == "list")
|
|
.unwrap()
|
|
.get("x-sausage-params")
|
|
.is_none()
|
|
);
|
|
assert!(actual["components"]["schemas"].get("Empty").is_none());
|
|
assert!(
|
|
sausage_interface::compat::check(&consumer["x-sausage-imports"]["notes"], &actual)
|
|
.is_empty()
|
|
);
|
|
host.stop().await;
|
|
}
|
|
/// Detached SDK calls receive progress and forward it through core.notify.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn detached_call_notifications() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let mut events = host.subscribe();
|
|
assert_eq!(call(&host, "probe.relay", json!({})).await, 42);
|
|
let event = tokio::time::timeout(Duration::from_secs(5), events.recv())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(event["params"]["name"], "step");
|
|
host.stop().await;
|
|
}
|
|
/// Detached typed calls complete and detached services survive their original caller.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn detached_service_survives_parent() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(call(&host, "typed.run", json!({})).await, 5);
|
|
let child = call(&host, "probe.launch", json!({})).await;
|
|
tokio::time::timeout(Duration::from_secs(5), async {
|
|
while host.processes().values().any(|p| p.method == "launch") {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert!(host.processes().contains_key(child.as_str().unwrap()));
|
|
host.stop().await;
|
|
}
|
|
/// A core.stop request closes stdin and executes the SDK stop hook.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stop_runs_hook() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let pid = call(&host, "probe.service", json!({})).await;
|
|
call(&host, "core.stop", json!({"pid":pid})).await;
|
|
assert_eq!(std::fs::read(world.0.join("stopped")).unwrap(), b"yes");
|
|
host.stop().await;
|
|
}
|
|
/// PROTOCOL 3.7: an idle SDK listener opts into drain and exits before EOF/grace.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn superseded_idle_listener_drains_before_grace() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let (notice, mut notices) = tokio::sync::mpsc::unbounded_channel();
|
|
host.register_native(
|
|
"other",
|
|
std::sync::Arc::new(move |_, method, _| {
|
|
let notice = notice.clone();
|
|
Box::pin(async move {
|
|
assert_eq!(method, "superseded");
|
|
notice.send(std::time::Instant::now()).unwrap();
|
|
Ok(json!({}))
|
|
})
|
|
}),
|
|
);
|
|
let pid = call(&host, "probe.drain_service", json!({})).await;
|
|
// Replace with a different real component; force only skips unrelated test consumers.
|
|
std::fs::copy(
|
|
fixtures().join("notes.wasm"),
|
|
world.0.join("state/next.wasm"),
|
|
)
|
|
.unwrap();
|
|
call(
|
|
&host,
|
|
"core.activate",
|
|
json!({"plugin":"probe","wasm":"/state/next.wasm","force":true}),
|
|
)
|
|
.await;
|
|
// Activation preparation/discovery is outside upgrade grace. Time from the
|
|
// hook's outgoing call, not from the start of candidate compilation.
|
|
let before = tokio::time::timeout(Duration::from_secs(60), notices.recv())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
// Waiting on its lifetime lock is an exit event, not process-liveness polling.
|
|
call(&host, "core.lock", json!({"path":"/drain-owner"})).await;
|
|
assert!(before.elapsed() < Duration::from_secs(5));
|
|
assert!(!host.processes().contains_key(pid.as_str().unwrap()));
|
|
assert!(
|
|
!world
|
|
.0
|
|
.join(format!("stopped-{}", pid.as_str().unwrap()))
|
|
.exists(),
|
|
"drain must not run on_stop (discovery processes may run their own hooks)"
|
|
);
|
|
host.stop().await;
|
|
}
|
|
|
|
/// SDK framing carries payloads above 64 KiB in both directions.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn large_messages() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let text = "z".repeat(192 * 1024);
|
|
assert_eq!(call(&host, "probe.echo", json!({"text":text})).await, text);
|
|
host.stop().await;
|
|
}
|
|
/// Normal SDK stdout never produces the discarded-output journal warning.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn no_discarded_stdout() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
call(&host, "probe.echo", json!({"text":"framing"})).await;
|
|
host.stop().await;
|
|
assert_eq!(
|
|
call(&host, "core.log", json!({"grep":"discarded stdout"})).await,
|
|
json!([])
|
|
);
|
|
}
|
|
/// Two SDK processes exchange endpoint requests and progress at a world path.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn world_endpoint() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
call(&host, "probe.service", json!({})).await;
|
|
assert_eq!(call(&host, "typed.world_endpoint", json!({})).await, 7);
|
|
host.stop().await;
|
|
}
|
|
/// The HTTP example exposes its self endpoint through the process namespace.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn www_control_through_proc() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(
|
|
call(&host, "consumer.demo", json!({"text":"web","port":0})).await,
|
|
0
|
|
);
|
|
host.stop().await;
|
|
}
|
|
/// The vault example publishes a fixed endpoint in the shared run namespace.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn secrets_fixed_endpoint() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
call(&host, "secrets.serve", json!({})).await;
|
|
assert_eq!(call(&host, "consumer.vault", json!({})).await, "hello");
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.6: a background task retains its SDK process until the task completes.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn background_task_retains_process() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let pid = call(&host, "probe.background_hold", json!({})).await;
|
|
assert!(host.processes().contains_key(pid.as_str().unwrap()));
|
|
assert_eq!(call(&host, "notes.list", json!({})).await, json!([]));
|
|
assert!(host.processes().contains_key(pid.as_str().unwrap()));
|
|
std::fs::write(world.0.join("release-background"), "release").unwrap();
|
|
tokio::time::timeout(Duration::from_secs(10), async {
|
|
while host.processes().contains_key(pid.as_str().unwrap()) {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
std::fs::read_to_string(world.0.join("background-ended")).unwrap(),
|
|
"done"
|
|
);
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 5: stale endpoint identities and interfaces fail through genuine SDK processes.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn endpoint_handshake_rejects_mismatch() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let pid = call(&host, "probe.service", json!({})).await;
|
|
let path = world.0.join("state/control");
|
|
let original = std::fs::read_to_string(&path).unwrap();
|
|
let words: Vec<_> = original.split_whitespace().collect();
|
|
for changed in [
|
|
format!("{} wrong-owner {}\n", words[0], words[2]),
|
|
format!("{} {} wrong-interface\n", words[0], words[1]),
|
|
] {
|
|
std::fs::write(&path, changed).unwrap();
|
|
let error = host
|
|
.call("typed.world_endpoint", json!({}))
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(error["code"], -32000);
|
|
assert!(error["message"].as_str().unwrap().contains("mismatch"));
|
|
assert!(host.processes().contains_key(pid.as_str().unwrap()));
|
|
}
|
|
std::fs::write(path, original).unwrap();
|
|
assert_eq!(call(&host, "typed.world_endpoint", json!({})).await, 7);
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 10.1 and 5: discovery through core retains params, endpoints, imports, and grace.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn discovery_extensions_through_core() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let provider = call(&host, "probe.rpc.discover", json!({})).await;
|
|
assert_eq!(provider["openrpc"], "1.2.6");
|
|
assert_eq!(provider["x-sausage-grace"], 12);
|
|
assert!(
|
|
provider["x-sausage-endpoints"]["control"]["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|m| m["name"] == "stats")
|
|
);
|
|
let add = provider["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.find(|m| m["name"] == "add")
|
|
.unwrap();
|
|
assert_eq!(add["paramStructure"], "by-name");
|
|
let reference = add["x-sausage-params"]["$ref"].as_str().unwrap();
|
|
assert_eq!(
|
|
provider.pointer(reference.trim_start_matches('#')).unwrap()["type"],
|
|
"object"
|
|
);
|
|
let consumer = call(&host, "typed.rpc.discover", json!({})).await;
|
|
assert!(
|
|
consumer["x-sausage-imports"]["probe"]["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|m| m["name"] == "add")
|
|
);
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.2: rpc.discover is SDK control traffic and endpoint control is not a main method.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn reserved_rpc_control_names() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert!(call(&host, "probe.rpc.discover", json!({})).await["methods"].is_array());
|
|
for method in ["probe.rpc.endpoint", "probe.rpc.unknown"] {
|
|
assert_eq!(
|
|
host.call(method, json!({})).await.unwrap_err()["code"],
|
|
-32601
|
|
);
|
|
}
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: the SDK's leading newline isolates an unfinished diagnostic from its response.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn leading_newline_protects_sdk_response() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(
|
|
call(&host, "probe.framed_output", json!({})).await,
|
|
"intact"
|
|
);
|
|
host.stop().await;
|
|
let warnings = call(&host, "core.log", json!({"grep":"discarded stdout"})).await;
|
|
assert_eq!(warnings.as_array().unwrap().len(), 1);
|
|
assert!(!warnings.to_string().contains("partial diagnostic"));
|
|
}
|