196 lines
6.9 KiB
Rust
196 lines
6.9 KiB
Rust
//! Real SDK watch callbacks and raw WASI filesystem mutations share the host namespace.
|
|
use super::common::{CACHE, probe, sdk_probe};
|
|
use sausage::{Host, world::World};
|
|
use serde_json::{Value, json};
|
|
use std::time::Duration;
|
|
|
|
// Same call bound as core.rs; no sleeps or polling.
|
|
async fn call(h: &Host, method: &str, p: Value) -> Value {
|
|
tokio::time::timeout(Duration::from_secs(20), h.call(method, p))
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
}
|
|
async fn event(h: &Host, rx: &mut tokio::sync::broadcast::Receiver<Value>, name: &str) -> Value {
|
|
let result = tokio::time::timeout(Duration::from_secs(20), async {
|
|
loop {
|
|
let e = rx.recv().await.unwrap();
|
|
if e["params"]["name"] == name {
|
|
return e["params"]["params"].clone();
|
|
}
|
|
}
|
|
})
|
|
.await;
|
|
match result {
|
|
Ok(v) => v,
|
|
Err(e) => {
|
|
eprintln!(
|
|
"JOURNAL {}",
|
|
h.call("core.log", json!({"limit":100})).await.unwrap()
|
|
);
|
|
panic!("{e}");
|
|
}
|
|
}
|
|
}
|
|
async fn start() -> (tempfile::TempDir, Host) {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let world = World::create(&dir.path().join("world")).unwrap();
|
|
for (name, wasm) in [("probe", probe()), ("sdk", sdk_probe())] {
|
|
std::fs::create_dir_all(world.plugin_dir(name)).unwrap();
|
|
std::fs::copy(wasm, world.plugin_dir(name).join("plugin.wasm")).unwrap();
|
|
}
|
|
let host = Host::start_with_cache(&world.root, std::path::Path::new(CACHE))
|
|
.await
|
|
.unwrap();
|
|
(dir, host)
|
|
}
|
|
fn watch(h: &Host, path: &str) -> tokio::task::JoinHandle<sausage::RpcResult> {
|
|
let (h, path) = (h.clone(), path.to_owned());
|
|
tokio::spawn(async move { h.call("sdk.watch", json!({"path":path})).await })
|
|
}
|
|
async fn mutate(h: &Host, op: &str, path: &str) {
|
|
call(h, "probe.mutate", json!({"op":op,"path":path})).await;
|
|
}
|
|
/// PROTOCOL 4: pre-arm state is in the first scan; each post-scan mutation wakes
|
|
/// the SDK without completing the watch. Covers descriptor/stream/truncate paths.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn armed_scan_and_later_changes() {
|
|
let (_dir, h) = start().await;
|
|
mutate(&h, "mkdir", "/state/mailbox").await;
|
|
call(
|
|
&h,
|
|
"probe.write",
|
|
json!({"path":"/state/mailbox/before", "text":"before"}),
|
|
)
|
|
.await;
|
|
let mut rx = h.subscribe();
|
|
let task = watch(&h, "/state/mailbox/../mailbox/");
|
|
let armed = event(&h, &mut rx, "armed").await;
|
|
assert_eq!(armed["entries"], json!(["before"]));
|
|
assert!(!task.is_finished());
|
|
for (op, expected) in [
|
|
("descriptor", "descriptor"),
|
|
("stream", "streamptor"),
|
|
("append", "streamptorappend"),
|
|
("size", ""),
|
|
("descriptor", "descriptor"),
|
|
("truncate", ""),
|
|
] {
|
|
mutate(&h, op, "/state/mailbox/before").await;
|
|
let mut scan = event(&h, &mut rx, "changed").await;
|
|
while scan["text"] != expected {
|
|
scan = event(&h, &mut rx, "changed").await;
|
|
}
|
|
}
|
|
mutate(&h, "burst", "/state/mailbox").await;
|
|
let mut scanned = event(&h, &mut rx, "changed").await;
|
|
while !scanned["entries"].as_array().unwrap().contains(&json!("4")) {
|
|
scanned = event(&h, &mut rx, "changed").await;
|
|
}
|
|
while rx.try_recv().is_ok() {}
|
|
mutate(&h, "delete", "/state/mailbox/4").await;
|
|
let mut scanned = event(&h, &mut rx, "changed").await;
|
|
while scanned["entries"].as_array().unwrap().contains(&json!("4")) {
|
|
scanned = event(&h, &mut rx, "changed").await;
|
|
}
|
|
call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
|
|
assert!(task.await.unwrap().is_err());
|
|
h.stop().await;
|
|
}
|
|
/// PROTOCOL 4: atomic publication, rename source/destination, mounted paths,
|
|
/// absent paths, and caller-death cleanup. No external host writes are required.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn rename_delete_and_mounts() {
|
|
let (_dir, h) = start().await;
|
|
for base in ["/state/inbox", "/run/inbox"] {
|
|
let mut rx = h.subscribe();
|
|
let task = watch(&h, base);
|
|
let armed = event(&h, &mut rx, "armed").await;
|
|
assert_eq!(armed["entries"], json!([]));
|
|
let staging = format!("{base}-staging");
|
|
mutate(&h, "mkdir", &staging).await;
|
|
call(
|
|
&h,
|
|
"probe.mutate",
|
|
json!({"op":"rename","path":staging,"to":base}),
|
|
)
|
|
.await;
|
|
event(&h, &mut rx, "changed").await;
|
|
mutate(&h, "atomic", &format!("{base}/published")).await;
|
|
let mut scan = event(&h, &mut rx, "changed").await;
|
|
while !scan["entries"]
|
|
.as_array()
|
|
.unwrap()
|
|
.contains(&json!("published"))
|
|
{
|
|
scan = event(&h, &mut rx, "changed").await;
|
|
}
|
|
assert_eq!(
|
|
call(
|
|
&h,
|
|
"probe.read",
|
|
json!({"path":format!("{base}/published")})
|
|
)
|
|
.await,
|
|
"complete"
|
|
);
|
|
while rx.try_recv().is_ok() {}
|
|
call(
|
|
&h,
|
|
"probe.mutate",
|
|
json!({"op":"rename","path":base,"to":format!("{base}-moved")}),
|
|
)
|
|
.await;
|
|
let mut scan = event(&h, &mut rx, "changed").await;
|
|
while scan["entries"] != json!([]) {
|
|
scan = event(&h, &mut rx, "changed").await;
|
|
}
|
|
call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
|
|
assert!(task.await.unwrap().is_err());
|
|
}
|
|
// /self watches bind to this caller, and mutations are visible via /proc/<pid>.
|
|
let mut rx = h.subscribe();
|
|
let task = watch(&h, "/self");
|
|
let armed = event(&h, &mut rx, "armed").await;
|
|
assert!(
|
|
armed["entries"]
|
|
.as_array()
|
|
.unwrap()
|
|
.contains(&json!("owned"))
|
|
);
|
|
event(&h, &mut rx, "changed").await;
|
|
assert_eq!(
|
|
call(
|
|
&h,
|
|
"probe.read",
|
|
json!({"path":format!("/proc/{}/owned", armed["pid"].as_str().unwrap())})
|
|
)
|
|
.await,
|
|
"visible via proc"
|
|
);
|
|
call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
|
|
assert!(task.await.unwrap().is_err());
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 9: core discovery includes the armed watch and matches its proc projection.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn discovery() {
|
|
let (_dir, h) = start().await;
|
|
let doc = call(&h, "core.rpc.discover", json!({})).await;
|
|
let projected = call(&h, "probe.read", json!({"path":"/proc/core/discover.json"})).await;
|
|
assert_eq!(
|
|
doc,
|
|
serde_json::from_str::<Value>(projected.as_str().unwrap()).unwrap()
|
|
);
|
|
let watch = doc["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.find(|m| m["name"] == "watch")
|
|
.unwrap();
|
|
assert_eq!(watch["params"][0]["name"], "path");
|
|
assert!(watch["description"].as_str().unwrap().contains("armed"));
|
|
h.stop().await;
|
|
}
|