SAU-135 implementer
168795b408
Make legacy core.exit share core.restart's supervisor gate and pending request. Seed the stable agent prompt and boot comments with the approval and shutdown contract, and check real init output under a deadline. Cover both aliases' gate, response grace, coalescing and exit status. Card: SAU-135
1998 lines
66 KiB
Rust
1998 lines
66 KiB
Rust
//! End-to-end host behavior with a raw WASI guest.
|
|
use super::common::probe as fixture;
|
|
use sausage::{Host, world::World};
|
|
use serde_json::{Value, json};
|
|
use std::path::PathBuf;
|
|
struct TestWorld(PathBuf);
|
|
impl TestWorld {
|
|
fn new() -> Self {
|
|
let p = std::env::temp_dir().join(format!("sausage-{}", uuid::Uuid::now_v7()));
|
|
let w = World::create(&p).unwrap();
|
|
std::fs::create_dir_all(w.plugin_dir("probe")).unwrap();
|
|
std::fs::copy(fixture(), w.plugin_dir("probe").join("plugin.wasm")).unwrap();
|
|
Self(p)
|
|
}
|
|
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(h: &Host, m: &str, p: Value) -> Value {
|
|
tokio::time::timeout(std::time::Duration::from_secs(20), h.call(m, p))
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
}
|
|
|
|
/// PROTOCOL 3.3: a closed stdin does not discard a response already on stdout.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn closed_stdin_drains_stdout() {
|
|
let w = TestWorld::new();
|
|
std::fs::write(w.0.join("state/early"), "answer").unwrap();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
call(&h, "probe.echo", json!({"large":"x".repeat(2*1024*1024)})).await,
|
|
"early"
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3: failed imports remain process failures, even with initial input in flight.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn failed_imports_preserve_process_status() {
|
|
let w = TestWorld::new();
|
|
let path = w.0.join("plugins/probe/plugin.wasm");
|
|
let mut bytes = std::fs::read(&path).unwrap();
|
|
for index in 0..bytes.len() - 5 {
|
|
if &bytes[index..index + 5] == b"wasi:" {
|
|
bytes[index..index + 5].copy_from_slice(b"fake:");
|
|
}
|
|
}
|
|
std::fs::write(path, bytes).unwrap();
|
|
let h = w.start().await;
|
|
let error = tokio::time::timeout(
|
|
std::time::Duration::from_secs(20),
|
|
h.call("probe.echo", json!({"large":"x".repeat(2*1024*1024)})),
|
|
)
|
|
.await
|
|
.unwrap()
|
|
.unwrap_err();
|
|
assert_eq!(error["code"], -32002);
|
|
assert!(error["data"]["status"].as_str().unwrap().contains("fake:"));
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3: a closed stdin preserves the actual exit status and final stderr.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn closed_stdin_preserves_failure_details() {
|
|
let w = TestWorld::new();
|
|
std::fs::write(w.0.join("state/early"), "fail").unwrap();
|
|
let h = w.start().await;
|
|
let error = tokio::time::timeout(
|
|
std::time::Duration::from_secs(20),
|
|
h.call("probe.echo", json!({"large":"x".repeat(2*1024*1024)})),
|
|
)
|
|
.await
|
|
.unwrap()
|
|
.unwrap_err();
|
|
assert_eq!(error["code"], -32002);
|
|
// WASIp2 command.run represents a nonzero exit as its unit error result.
|
|
assert_eq!(error["data"]["status"], "exit 1");
|
|
assert!(
|
|
error["data"]["lines"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|line| line["line"] == "last early log")
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 7: invalid stderr bytes do not stop later log lines or the guest.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stderr_uses_lossy_utf8_and_continues() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let pid = call(&h, "probe.invalid_stderr", json!({})).await;
|
|
call(&h, "core.stop", json!({"pid":pid})).await;
|
|
let logs = call(&h, "core.log", json!({"pid":pid})).await;
|
|
assert!(
|
|
logs.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|line| line["line"] == "\u{fffd}")
|
|
);
|
|
assert!(
|
|
logs.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|line| line["line"] == "after invalid UTF-8")
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3 and 7: stderr drains even while a large initial stdin write blocks.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stderr_flood_does_not_block_stdin() {
|
|
let w = TestWorld::new();
|
|
std::fs::write(w.0.join("state/early"), "flood").unwrap();
|
|
let h = w.start().await;
|
|
let payload = json!({"large":"x".repeat(1024*1024)});
|
|
assert_eq!(call(&h, "probe.echo", payload.clone()).await, payload);
|
|
h.stop().await;
|
|
let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
|
|
let count: i64 = db
|
|
.query_row(
|
|
"SELECT count(*) FROM journal WHERE line LIKE 'flood %'",
|
|
[],
|
|
|row| row.get(0),
|
|
)
|
|
.unwrap();
|
|
assert!(count >= 5000);
|
|
}
|
|
|
|
/// PROTOCOL 3.6: shutdown rejects new processes, including calls after stop returns.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn shutdown_rejects_new_processes() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
h.stop().await;
|
|
assert_eq!(
|
|
h.call("probe.echo", json!({})).await.unwrap_err()["code"],
|
|
-32600
|
|
);
|
|
assert!(h.processes().is_empty());
|
|
}
|
|
|
|
/// PROTOCOL 6: stdout unlock takes effect before the next result reaches its caller.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stdout_unlock_precedes_result() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
for _ in 0..10 {
|
|
let pid = call(&h, "probe.unlock_answer", json!({})).await;
|
|
assert_eq!(
|
|
call(
|
|
&h,
|
|
"core.lock",
|
|
json!({"path":"ordered-unlock","wait":false})
|
|
)
|
|
.await,
|
|
json!({"held":true})
|
|
);
|
|
call(&h, "core.unlock", json!({"path":"ordered-unlock"})).await;
|
|
call(&h, "core.stop", json!({"pid":pid})).await;
|
|
}
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.5: a notification-started process has no call to receive progress.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn notification_process_has_no_progress_destination() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(call(&h, "probe.notification_child", json!({})).await, 0);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: an explicit null error does not replace a successful result.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn null_error_is_success() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(call(&h, "probe.null_error", json!({})).await, 42);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 7: a journal insertion failure does not poison the host or stop later writes.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn journal_recovers_after_sqlite_write_failure() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
|
|
db.busy_timeout(std::time::Duration::from_secs(5)).unwrap();
|
|
db.execute_batch("CREATE TRIGGER reject_writes BEFORE INSERT ON journal BEGIN SELECT RAISE(FAIL,'test failure'); END;").unwrap();
|
|
assert_eq!(
|
|
call(&h, "probe.echo", json!({"failed_log":true})).await,
|
|
json!({"failed_log":true})
|
|
);
|
|
call(&h, "core.log", json!({})).await;
|
|
db.execute_batch("DROP TRIGGER reject_writes;").unwrap();
|
|
assert_eq!(
|
|
call(&h, "probe.echo", json!({"recovered":true})).await,
|
|
json!({"recovered":true})
|
|
);
|
|
let entries = call(&h, "core.log", json!({"grep":"recovered"})).await;
|
|
assert!(!entries.as_array().unwrap().is_empty());
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: trailing slashes preserve live proc folders and generated plugin folders.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn proc_directories_accept_trailing_slashes() {
|
|
use sausage_vfs::{Layer, Open};
|
|
let w = TestWorld::new();
|
|
let document = json!({"methods":[]});
|
|
std::fs::create_dir_all(w.0.join("state/probe")).unwrap();
|
|
std::fs::write(w.0.join("state/probe/discover.json"), document.to_string()).unwrap();
|
|
let h = w.start().await;
|
|
let view = h.filesystem().await.unwrap();
|
|
tokio::time::timeout(std::time::Duration::from_secs(20), async {
|
|
loop {
|
|
if view.list("/proc/core/plugins/probe").await.is_ok() {
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
let pid = call(&h, "probe.serve", json!({})).await;
|
|
for path in [
|
|
"/proc/".to_string(),
|
|
"/proc/core/".into(),
|
|
"/proc/core/plugins/".into(),
|
|
"/proc/core/plugins/probe/".into(),
|
|
format!("/proc/{}/", pid.as_str().unwrap()),
|
|
] {
|
|
assert!(view.stat(&path, true).await.unwrap().directory);
|
|
view.list(&path).await.unwrap();
|
|
view.open(
|
|
&path,
|
|
Open {
|
|
directory: true,
|
|
..Open::default()
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
}
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.4: Unix replacements with the same size and mtime use the new component.
|
|
#[cfg(unix)]
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn component_identity_includes_inode_and_ctime() {
|
|
let w = TestWorld::new();
|
|
let path = w.0.join("plugins/probe/plugin.wasm");
|
|
let mut bytes = std::fs::read(&path).unwrap();
|
|
bytes.extend_from_slice(&[0, 2, 1, b'a']);
|
|
std::fs::write(&path, &bytes).unwrap();
|
|
let modified = std::fs::metadata(&path).unwrap().modified().unwrap();
|
|
let h = w.start().await;
|
|
let old = call(&h, "probe.serve", json!({})).await;
|
|
*bytes.last_mut().unwrap() = b'b';
|
|
std::fs::write(&path, &bytes).unwrap();
|
|
std::fs::File::options()
|
|
.write(true)
|
|
.open(&path)
|
|
.unwrap()
|
|
.set_times(std::fs::FileTimes::new().set_modified(modified))
|
|
.unwrap();
|
|
let new = call(&h, "probe.serve", json!({})).await;
|
|
let table = h.processes();
|
|
assert_ne!(
|
|
table[old.as_str().unwrap()].version,
|
|
table[new.as_str().unwrap()].version
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.6 and 6: stopping cancels a native process's queued lock request immediately.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stopping_removes_queued_lock_acquisitions() {
|
|
use sausage_vfs::{Layer, Open};
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "core.lock", json!({"path":"queued-stop"})).await;
|
|
h.register_native(
|
|
"queued",
|
|
std::sync::Arc::new(|ctx, _, _| {
|
|
Box::pin(async move {
|
|
ctx.notify("ready", json!({"pid":ctx.pid}));
|
|
ctx.call("core.lock", json!({"path":"queued-stop"})).await
|
|
})
|
|
}),
|
|
);
|
|
let mut events = h.subscribe();
|
|
let host = h.clone();
|
|
let pending = tokio::spawn(async move { host.call("queued.wait", json!({})).await });
|
|
let event = events.recv().await.unwrap();
|
|
let pid = event["params"]["params"]["pid"].clone();
|
|
let view = h.filesystem().await.unwrap();
|
|
tokio::time::timeout(std::time::Duration::from_secs(10), async {
|
|
loop {
|
|
let f = view
|
|
.open("/proc/core/locks.json", Open::default())
|
|
.await
|
|
.unwrap();
|
|
let locks: Value = serde_json::from_slice(&f.read(0, 65536).await.unwrap()).unwrap();
|
|
if locks["queued-stop"]["waiting"] == json!([pid]) {
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
tokio::time::timeout(
|
|
std::time::Duration::from_secs(1),
|
|
h.call("core.stop", json!({"pid":pid})),
|
|
)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert!(pending.await.unwrap().is_err());
|
|
call(&h, "core.unlock", json!({"path":"queued-stop"})).await;
|
|
assert_eq!(
|
|
call(&h, "core.lock", json!({"path":"queued-stop","wait":false})).await,
|
|
json!({"held":true})
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: json rpc lines carry results.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn json_rpc_lines_carry_results() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(call(&h, "probe.echo", json!({"x":1})).await, json!({"x":1}));
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.2: method names split at the first dot.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn method_names_split_at_the_first_dot() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
call(&h, "probe.rpc.discover", json!({"x":2})).await,
|
|
json!({"x":2})
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3: processes can call other plugins.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn processes_can_call_other_plugins() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
call(
|
|
&h,
|
|
"probe.call",
|
|
json!({"method":"probe.echo","params":{"nested":true}})
|
|
)
|
|
.await,
|
|
json!({"nested":true})
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3: pending ids distinguish numbers from strings.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn pending_ids_distinguish_numbers_from_strings() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
call(&h, "probe.multicall", json!({})).await,
|
|
json!({"count":2})
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3: notifications start attached processes.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn notifications_start_attached_processes() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(
|
|
&h,
|
|
"probe.notification",
|
|
json!({"path":"/state/notification","text":"delivered"}),
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/notification")).unwrap(),
|
|
"delivered"
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3: unknown response ids are dropped.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn unknown_response_ids_are_dropped() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
h.call("probe.unknownid", json!({})).await.unwrap_err()["code"],
|
|
-32002
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.8: process environment identifies the caller.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn process_environment_identifies_the_caller() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let env = call(&h, "probe.env", json!({})).await;
|
|
assert_eq!(env["plugin"], "probe");
|
|
assert_eq!(env["parent"], "console");
|
|
assert!(env["pid"].as_str().unwrap().starts_with("probe-"));
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.6: processes read and write the world.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn processes_read_and_write_the_world() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(
|
|
&h,
|
|
"probe.write",
|
|
json!({"path":"/state/test","text":"hello"}),
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
call(&h, "probe.read", json!({"path":"/state/test"})).await,
|
|
"hello"
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.9: missing plugins return minus 32001.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn missing_plugins_return_minus_32001() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
h.call("absent.run", json!({})).await.unwrap_err()["code"],
|
|
-32001
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 9: unimplemented core methods return method not found.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn unimplemented_core_methods_return_method_not_found() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
h.call("core.unknown", json!({})).await.unwrap_err()["code"],
|
|
-32601
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: params must be objects.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn params_must_be_objects() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
h.call("probe.echo", json!([])).await.unwrap_err()["code"],
|
|
-32602
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.9: exit without answer reports status and last logs.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn exit_without_answer_reports_status_and_last_logs() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let e = h.call("probe.exit", json!({})).await.unwrap_err();
|
|
assert_eq!(e["code"], -32002);
|
|
assert!(
|
|
e["data"]["lines"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|l| l["line"] == "last line")
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.5: progress notifications reach the caller.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn progress_notifications_reach_the_caller() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let mut notifications = h.subscribe();
|
|
call(&h, "probe.notify", json!({})).await;
|
|
assert_eq!(
|
|
notifications.recv().await.unwrap()["params"]["name"],
|
|
"progress"
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.5: every stdout notification reaches the caller before the following result.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stdout_notifications_precede_the_result() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let mut notifications = h.subscribe();
|
|
for _ in 0..10 {
|
|
assert_eq!(call(&h, "probe.burst", json!({"count":64})).await, 64);
|
|
let mut call_id = None;
|
|
for index in 0..64 {
|
|
// A delayed notification fails here, even if it arrives later.
|
|
let notification = notifications.try_recv().unwrap();
|
|
assert_eq!(notification["params"]["name"], "burst");
|
|
assert_eq!(notification["params"]["params"]["index"], index);
|
|
if let Some(id) = &call_id {
|
|
assert_eq!(¬ification["params"]["call"], id);
|
|
} else {
|
|
call_id = Some(notification["params"]["call"].clone());
|
|
}
|
|
}
|
|
assert!(matches!(
|
|
notifications.try_recv(),
|
|
Err(tokio::sync::broadcast::error::TryRecvError::Empty)
|
|
));
|
|
}
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: non message stdout warns once without its content.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn non_message_stdout_warns_once_without_its_content() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.junk", json!({})).await;
|
|
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
|
while !h.processes().is_empty() {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
let starts = call(&h, "core.log", json!({"grep":"start junk"})).await;
|
|
let pid = &starts[0]["pid"];
|
|
let warnings = call(&h, "core.log", json!({"pid":pid,"grep":"discarded stdout"})).await;
|
|
assert_eq!(warnings.as_array().unwrap().len(), 1);
|
|
assert!(!warnings.to_string().contains("junk"));
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: stderr lines enter the journal.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stderr_lines_enter_the_journal() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.junk", json!({})).await;
|
|
h.stop().await;
|
|
let log = call(&h, "core.log", json!({"grep":"stderr line"})).await;
|
|
assert_eq!(log.as_array().unwrap().len(), 1);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 6: try lock reports held without waiting.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn try_lock_reports_held_without_waiting() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
call(&h, "core.lock", json!({"path":"x"})).await,
|
|
json!({"held":true})
|
|
);
|
|
assert_eq!(
|
|
call(&h, "core.lock", json!({"path":"x","wait":false})).await,
|
|
json!({"held":false})
|
|
);
|
|
call(&h, "core.unlock", json!({"path":"x"})).await;
|
|
assert_eq!(
|
|
call(&h, "probe.lock", json!({"path":"x"})).await,
|
|
json!({"held":true})
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 9: kill ends a process at once.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn kill_ends_a_process_at_once() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let pid = call(&h, "probe.serve", json!({})).await;
|
|
call(&h, "core.kill", json!({"pid":pid})).await;
|
|
assert!(!h.processes().contains_key(pid.as_str().unwrap()));
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 9: native calls have process ids and results.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn native_calls_have_process_ids_and_results() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
h.register_native(
|
|
"native",
|
|
std::sync::Arc::new(|ctx, method, params| {
|
|
Box::pin(async move {
|
|
ctx.notify("native", json!({}));
|
|
Ok(json!({"pid":ctx.pid,"method":method,"params":params}))
|
|
})
|
|
}),
|
|
);
|
|
let result = call(&h, "native.echo", json!({"x":1})).await;
|
|
assert_eq!(result["method"], "echo");
|
|
assert!(result["pid"].as_str().unwrap().starts_with("native-"));
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 2: reserved world roots prevent startup.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn reserved_world_roots_prevent_startup() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
for name in ["proc", "run", "self"] {
|
|
std::fs::create_dir(w.0.join(name)).unwrap();
|
|
assert!(World::open(&w.0).is_err());
|
|
std::fs::remove_dir(w.0.join(name)).unwrap();
|
|
}
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 8: boot waits for each readiness result before the next service.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn boot_waits_for_readiness_in_order() {
|
|
let w = TestWorld::new();
|
|
std::fs::write(
|
|
w.0.join("config/boot.yaml"),
|
|
serde_yaml_ng::to_string(&json!([
|
|
{"method":"probe.write","params":{"path":"/state/boot","text":"ready"}},
|
|
{"method":"probe.read","params":{"path":"/state/boot"}}
|
|
]))
|
|
.unwrap(),
|
|
)
|
|
.unwrap();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/boot")).unwrap(),
|
|
"ready"
|
|
);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.6: attached children stop when the parent ends.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn attached_children_stop_when_the_parent_ends() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let attached = call(
|
|
&h,
|
|
"probe.call",
|
|
json!({"method":"probe.serve","params":{}}),
|
|
)
|
|
.await;
|
|
tokio::time::timeout(std::time::Duration::from_secs(7), async {
|
|
while h.processes().contains_key(attached.as_str().unwrap()) {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.4: detached calls have no parent and outlive the caller.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn detached_calls_have_no_parent_and_outlive_the_caller() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let detached = call(
|
|
&h,
|
|
"probe.call",
|
|
json!({"method":"core.call","params":{"method":"probe.serve","params":{},"detach":true}}),
|
|
)
|
|
.await;
|
|
assert!(h.processes()[detached.as_str().unwrap()].detached);
|
|
call(&h, "core.stop", json!({"pid":detached})).await;
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.8: busy guests do not block other calls.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn busy_guests_do_not_block_other_calls() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let hc = h.clone();
|
|
let busy = tokio::spawn(async move { hc.call("probe.busy", json!({})).await });
|
|
let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
|
|
loop {
|
|
if let Some((pid, _)) = h.processes().into_iter().find(|(_, p)| p.method == "busy") {
|
|
break pid;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
call(&h, "probe.echo", json!({"works":true})).await,
|
|
json!({"works":true})
|
|
);
|
|
call(&h, "core.kill", json!({"pid":pid})).await;
|
|
assert_eq!(busy.await.unwrap().unwrap_err()["code"], -32002);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 6: same process lock acquisitions queue.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn same_process_lock_acquisitions_queue() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
|
|
h.register_native(
|
|
"locks",
|
|
std::sync::Arc::new(|ctx, _, _| {
|
|
Box::pin(async move {
|
|
assert_eq!(
|
|
ctx.call("core.lock", json!({"path":"fifo"})).await.unwrap(),
|
|
json!({"held":true})
|
|
);
|
|
assert_eq!(
|
|
ctx.call("core.lock", json!({"path":"fifo","wait":false}))
|
|
.await
|
|
.unwrap(),
|
|
json!({"held":false})
|
|
);
|
|
let mut second = Box::pin(ctx.call("core.lock", json!({"path":"fifo"})));
|
|
std::future::poll_fn(|cx| {
|
|
assert!(second.as_mut().poll(cx).is_pending());
|
|
std::task::Poll::Ready(())
|
|
})
|
|
.await;
|
|
ctx.call("core.unlock", json!({"path":"fifo"}))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(second.await.unwrap(), json!({"held":true}));
|
|
Ok(json!({}))
|
|
})
|
|
}),
|
|
);
|
|
call(&h, "locks.test", json!({})).await;
|
|
h.stop().await;
|
|
assert_eq!(
|
|
call(&h, "core.lock", json!({"path":"fifo","wait":false})).await,
|
|
json!({"held":true})
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.8: tcp udp and dns.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn tcp_udp_and_dns() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
|
|
use tokio::io::AsyncWriteExt;
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
|
let tcp = format!("localhost:{}", listener.local_addr().unwrap().port());
|
|
let udp = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
|
|
let udp_addr = udp.local_addr().unwrap().to_string();
|
|
let server = tokio::spawn(async move {
|
|
let (mut stream, _) = listener.accept().await.unwrap();
|
|
stream.write_all(b"tcp").await.unwrap();
|
|
});
|
|
assert_eq!(
|
|
call(&h, "probe.network", json!({"tcp":tcp,"udp":udp_addr})).await,
|
|
"tcp"
|
|
);
|
|
let mut buf = [0; 3];
|
|
udp.recv_from(&mut buf).await.unwrap();
|
|
assert_eq!(&buf, b"udp");
|
|
server.await.unwrap();
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 6: native kill drops future and releases locks.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn native_kill_drops_future_and_releases_locks() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
|
|
use std::sync::{
|
|
Arc,
|
|
atomic::{AtomicBool, Ordering},
|
|
};
|
|
struct Dropped(Arc<AtomicBool>);
|
|
impl Drop for Dropped {
|
|
fn drop(&mut self) {
|
|
self.0.store(true, Ordering::SeqCst);
|
|
}
|
|
}
|
|
let dropped = Arc::new(AtomicBool::new(false));
|
|
let flag = dropped.clone();
|
|
h.register_native(
|
|
"native",
|
|
Arc::new(move |ctx, _, _| {
|
|
let flag = flag.clone();
|
|
Box::pin(async move {
|
|
let _guard = Dropped(flag);
|
|
ctx.call("core.lock", json!({"path":"native-lock"}))
|
|
.await
|
|
.unwrap();
|
|
ctx.notify("ready", json!({"pid":ctx.pid}));
|
|
std::future::pending().await
|
|
})
|
|
}),
|
|
);
|
|
let mut notifications = h.subscribe();
|
|
let hc = h.clone();
|
|
let pending = tokio::spawn(async move { hc.call("native.wait", json!({})).await });
|
|
let notification =
|
|
tokio::time::timeout(std::time::Duration::from_secs(10), notifications.recv())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
let pid = ¬ification["params"]["params"]["pid"];
|
|
assert_eq!(
|
|
call(&h, "core.lock", json!({"path":"native-lock","wait":false})).await,
|
|
json!({"held":false})
|
|
);
|
|
call(&h, "core.kill", json!({"pid":pid})).await;
|
|
tokio::time::timeout(std::time::Duration::from_secs(1), async {
|
|
while !dropped.load(Ordering::SeqCst) {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
call(&h, "core.lock", json!({"path":"native-lock","wait":false})).await,
|
|
json!({"held":true})
|
|
);
|
|
assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.6: host shutdown stops busy processes together after five seconds.
|
|
/// PROTOCOL 3.6: core.stop terminates an attached tree within the same five-second bound.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stop_deadlines() {
|
|
tokio::join!(
|
|
shutdown_stops_busy_processes_in_parallel_after_five_seconds(),
|
|
attached_tree_stops_in_parallel(),
|
|
);
|
|
}
|
|
|
|
/// Checks the shutdown half of the shared deadline test.
|
|
async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
|
|
let mut tasks = Vec::new();
|
|
for _ in 0..2 {
|
|
let hc = h.clone();
|
|
tasks.push(tokio::spawn(async move {
|
|
hc.call("probe.busy", json!({})).await
|
|
}));
|
|
}
|
|
tokio::time::timeout(std::time::Duration::from_secs(20), async {
|
|
while h
|
|
.processes()
|
|
.values()
|
|
.filter(|p| p.plugin == "probe" && p.method == "busy")
|
|
.count()
|
|
!= 2
|
|
{
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
let before = std::time::Instant::now();
|
|
h.stop().await;
|
|
assert!(before.elapsed() >= std::time::Duration::from_millis(4900));
|
|
assert!(before.elapsed() < std::time::Duration::from_secs(7));
|
|
for task in tasks {
|
|
assert_eq!(task.await.unwrap().unwrap_err()["code"], -32002);
|
|
}
|
|
assert!(h.processes().is_empty());
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.6: stopping rejects calls but allows notifications.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stopping_rejects_calls_but_allows_notifications() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
|
|
use std::sync::{Arc, Mutex};
|
|
let (tx, rx) = tokio::sync::oneshot::channel();
|
|
let tx = Arc::new(Mutex::new(Some(tx)));
|
|
let gate = Arc::new(tokio::sync::Notify::new());
|
|
let release = gate.clone();
|
|
h.register_native(
|
|
"native",
|
|
Arc::new(move |ctx, _, _| {
|
|
let tx = tx.lock().unwrap().take().unwrap();
|
|
let gate = gate.clone();
|
|
Box::pin(async move {
|
|
tx.send(ctx.clone()).ok();
|
|
gate.notified().await;
|
|
Ok(json!({}))
|
|
})
|
|
}),
|
|
);
|
|
let hc = h.clone();
|
|
let pending = tokio::spawn(async move { hc.call("native.wait", json!({})).await });
|
|
let ctx = rx.await.unwrap();
|
|
let hc = h.clone();
|
|
let pid = ctx.pid.clone();
|
|
let stop = tokio::spawn(async move { hc.call("core.stop", json!({"pid":pid})).await });
|
|
tokio::time::timeout(std::time::Duration::from_secs(2), async {
|
|
while h.processes()[&ctx.pid].state != "stopping" {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
ctx.call("probe.echo", json!({})).await.unwrap_err()["code"],
|
|
-32600
|
|
);
|
|
let mut notifications = h.subscribe();
|
|
ctx.call("core.notify", json!({"name":"stop-hook","params":{}}))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
notifications.recv().await.unwrap()["params"]["name"],
|
|
"stop-hook"
|
|
);
|
|
release.notify_one();
|
|
assert_eq!(pending.await.unwrap().unwrap(), json!({}));
|
|
assert_eq!(stop.await.unwrap().unwrap(), json!({}));
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 9: stop closes stdin and waits for exit.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stop_closes_stdin_and_waits_for_exit() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let hc = h.clone();
|
|
let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await });
|
|
let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
|
|
loop {
|
|
if let Some((pid, _)) = h
|
|
.processes()
|
|
.into_iter()
|
|
.find(|(_, p)| p.method == "wait_eof")
|
|
{
|
|
break pid;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
call(&h, "core.stop", json!({"pid":pid})).await;
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/stopped")).unwrap(),
|
|
"yes"
|
|
);
|
|
assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3: process ids are uuidv7 and versions are sha256.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn process_ids_are_uuidv7_and_versions_are_sha256() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
use sha2::{Digest, Sha256};
|
|
let pid = call(&h, "probe.serve", json!({"version":1})).await;
|
|
let pid = pid.as_str().unwrap();
|
|
let info = h.processes()[pid].clone();
|
|
assert_eq!(
|
|
info.version,
|
|
format!("{:x}", Sha256::digest(std::fs::read(fixture()).unwrap()))
|
|
);
|
|
assert_eq!(info.plugin, "probe");
|
|
assert_eq!(info.method, "serve");
|
|
assert_eq!(info.params, json!({"version":1}));
|
|
assert_eq!(info.parent.as_deref(), Some("console"));
|
|
assert_eq!(info.state, "running");
|
|
assert!(!info.detached);
|
|
assert!(info.start_time > 0);
|
|
let uuid = uuid::Uuid::parse_str(pid.strip_prefix("probe-").unwrap()).unwrap();
|
|
assert_eq!(uuid.get_version_num(), 7);
|
|
call(&h, "core.stop", json!({"pid":pid})).await;
|
|
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 7: journal filters and entries persist in sqlite.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn journal_filters_and_entries_persist_in_sqlite() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let pid = call(&h, "probe.serve", json!({})).await;
|
|
call(&h, "core.stop", json!({"pid":pid})).await;
|
|
let entries = call(
|
|
&h,
|
|
"core.log",
|
|
json!({"plugin":"probe","pid":pid,"limit":2}),
|
|
)
|
|
.await;
|
|
assert_eq!(entries.as_array().unwrap().len(), 2);
|
|
let entry = &entries[0];
|
|
let id = &entry["id"];
|
|
let single = call(&h, "core.log", json!({"since":id,"until":id})).await;
|
|
assert_eq!(single, json!([entry]));
|
|
h.stop().await;
|
|
drop(h);
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
call(&h, "core.log", json!({"since":id,"until":id})).await,
|
|
single
|
|
);
|
|
h.stop().await;
|
|
assert!(
|
|
std::fs::read(w.0.join("state/core/journal.sqlite"))
|
|
.unwrap()
|
|
.starts_with(b"SQLite format 3")
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: blank stdout lines do not warn.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn blank_stdout_lines_do_not_warn() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.blank", json!({})).await;
|
|
h.stop().await;
|
|
let warnings = call(&h, "core.log", json!({"grep":"discarded stdout"})).await;
|
|
assert_eq!(warnings, json!([]));
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: large requests and responses cross pipe capacity.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn large_requests_and_responses_cross_pipe_capacity() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let params = json!({"text":"x".repeat(192*1024)});
|
|
assert_eq!(call(&h, "probe.echo", params.clone()).await, params);
|
|
let outgoing = json!({"method":"probe.echo","params":params});
|
|
assert_eq!(call(&h, "probe.call", outgoing).await, params);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.8: two plugins execute calls at the same time.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn two_plugins_execute_calls_at_the_same_time() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let other = w.0.join("plugins/other");
|
|
std::fs::create_dir_all(&other).unwrap();
|
|
std::fs::copy(fixture(), other.join("plugin.wasm")).unwrap();
|
|
let hc = h.clone();
|
|
let slow = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await });
|
|
let pid = tokio::time::timeout(std::time::Duration::from_secs(10), async {
|
|
while !w.0.join("state/waiting").exists() {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
loop {
|
|
if let Some((pid, _)) = h
|
|
.processes()
|
|
.into_iter()
|
|
.find(|(_, p)| p.method == "wait_eof")
|
|
{
|
|
break pid;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
call(&h, "other.echo", json!({"ready":true})).await,
|
|
json!({"ready":true})
|
|
);
|
|
assert!(!slow.is_finished());
|
|
call(&h, "core.stop", json!({"pid":pid})).await;
|
|
assert_eq!(slow.await.unwrap().unwrap_err()["code"], -32002);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.3: request id is the process id.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn request_id_is_the_process_id() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let pid = call(&h, "probe.request_id", json!({})).await;
|
|
let uuid =
|
|
uuid::Uuid::parse_str(pid.as_str().unwrap().strip_prefix("probe-").unwrap()).unwrap();
|
|
assert_eq!(uuid.get_version_num(), 7);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 9: core call returns the target result.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn core_call_returns_the_target_result() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
call(
|
|
&h,
|
|
"core.call",
|
|
json!({"method":"probe.echo","params":{"n":1}})
|
|
)
|
|
.await,
|
|
json!({"n":1})
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 9: core unlock releases the named lock.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn core_unlock_releases_the_named_lock() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "core.lock", json!({"path":"unlock"})).await;
|
|
call(&h, "core.unlock", json!({"path":"unlock"})).await;
|
|
assert_eq!(
|
|
call(&h, "core.lock", json!({"path":"unlock","wait":false})).await,
|
|
json!({"held":true})
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 6: locks do not block file writes.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn locks_do_not_block_file_writes() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "core.lock", json!({"path":"/state/advisory"})).await;
|
|
call(
|
|
&h,
|
|
"probe.write",
|
|
json!({"path":"/state/advisory","text":"allowed"}),
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/advisory")).unwrap(),
|
|
"allowed"
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 6: crashes release all process locks.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn crashes_release_all_process_locks() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
h.register_native(
|
|
"crash",
|
|
std::sync::Arc::new(|ctx, _, _| {
|
|
Box::pin(async move {
|
|
ctx.call("core.lock", json!({"path":"crash"}))
|
|
.await
|
|
.unwrap();
|
|
panic!("test crash");
|
|
})
|
|
}),
|
|
);
|
|
assert_eq!(
|
|
h.call("crash.run", json!({})).await.unwrap_err()["code"],
|
|
-32002
|
|
);
|
|
assert_eq!(
|
|
call(&h, "core.lock", json!({"path":"crash","wait":false})).await,
|
|
json!({"held":true})
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 7: journal entries have uuid time identity stream and line.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn journal_entries_have_uuid_time_identity_stream_and_line() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.junk", json!({})).await;
|
|
h.stop().await;
|
|
let entries = call(&h, "core.log", json!({"plugin":"probe"})).await;
|
|
for entry in entries.as_array().unwrap() {
|
|
assert_eq!(
|
|
uuid::Uuid::parse_str(entry["id"].as_str().unwrap())
|
|
.unwrap()
|
|
.get_version_num(),
|
|
7
|
|
);
|
|
assert!(entry["time"].as_str().unwrap().contains('T'));
|
|
assert!(entry["time"].as_str().unwrap().ends_with('Z'));
|
|
assert!(entry["pid"].as_str().unwrap().starts_with("probe-"));
|
|
assert!(["stderr", "core"].contains(&entry["stream"].as_str().unwrap()));
|
|
assert!(entry["line"].is_string());
|
|
}
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 7: journal has wal and identity indexes.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn journal_has_wal_and_identity_indexes() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
h.stop().await;
|
|
let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
|
|
assert_eq!(
|
|
db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0))
|
|
.unwrap(),
|
|
"wal"
|
|
);
|
|
for name in ["journal_pid", "journal_plugin"] {
|
|
assert_eq!(
|
|
db.query_row(
|
|
"SELECT count(*) FROM sqlite_master WHERE type='index' AND name=?1",
|
|
[name],
|
|
|r| r.get::<_, i64>(0)
|
|
)
|
|
.unwrap(),
|
|
1
|
|
);
|
|
}
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 6: acquisitions from different processes keep their FIFO positions.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn lock_acquisitions_from_different_processes_are_fifo() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
|
|
let gate = std::sync::Arc::new(tokio::sync::Semaphore::new(0));
|
|
let release = gate.clone();
|
|
h.register_native(
|
|
"queued",
|
|
std::sync::Arc::new(move |ctx, _, _| {
|
|
let tx = tx.clone();
|
|
let gate = gate.clone();
|
|
Box::pin(async move {
|
|
tx.send(ctx).unwrap();
|
|
gate.acquire().await.unwrap().forget();
|
|
Ok(json!({}))
|
|
})
|
|
}),
|
|
);
|
|
let first_host = h.clone();
|
|
let first = tokio::spawn(async move { first_host.call("queued.wait", json!({})).await });
|
|
let second_host = h.clone();
|
|
let second = tokio::spawn(async move { second_host.call("queued.wait", json!({})).await });
|
|
let first_ctx = rx.recv().await.unwrap();
|
|
let second_ctx = rx.recv().await.unwrap();
|
|
call(&h, "core.lock", json!({"path":"ordered"})).await;
|
|
let mut first_lock = Box::pin(first_ctx.call("core.lock", json!({"path":"ordered"})));
|
|
let mut second_lock = Box::pin(second_ctx.call("core.lock", json!({"path":"ordered"})));
|
|
std::future::poll_fn(|cx| {
|
|
assert!(first_lock.as_mut().poll(cx).is_pending());
|
|
assert!(second_lock.as_mut().poll(cx).is_pending());
|
|
std::task::Poll::Ready(())
|
|
})
|
|
.await;
|
|
call(&h, "core.unlock", json!({"path":"ordered"})).await;
|
|
assert_eq!(first_lock.await.unwrap(), json!({"held":true}));
|
|
std::future::poll_fn(|cx| {
|
|
assert!(second_lock.as_mut().poll(cx).is_pending());
|
|
std::task::Poll::Ready(())
|
|
})
|
|
.await;
|
|
first_ctx
|
|
.call("core.unlock", json!({"path":"ordered"}))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(second_lock.await.unwrap(), json!({"held":true}));
|
|
release.add_permits(2);
|
|
first.await.unwrap().unwrap();
|
|
second.await.unwrap().unwrap();
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.5: notifications after the result do not reach the caller.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn notifications_after_the_result_are_dropped() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let (tx, rx) = tokio::sync::oneshot::channel();
|
|
let tx = std::sync::Mutex::new(Some(tx));
|
|
h.register_native(
|
|
"native",
|
|
std::sync::Arc::new(move |ctx, _, _| {
|
|
tx.lock().unwrap().take().unwrap().send(ctx).ok();
|
|
Box::pin(async { Ok(json!({})) })
|
|
}),
|
|
);
|
|
let mut notifications = h.subscribe();
|
|
call(&h, "native.answer", json!({})).await;
|
|
let ctx = rx.await.unwrap();
|
|
ctx.notify("late", json!({}));
|
|
assert!(matches!(
|
|
notifications.try_recv(),
|
|
Err(tokio::sync::broadcast::error::TryRecvError::Empty)
|
|
));
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: non-UTF-8 stdout is discarded, not treated as blank framing.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn non_utf8_stdout_is_not_blank_framing() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.invalid_utf8", json!({})).await;
|
|
h.stop().await;
|
|
let warnings = call(&h, "core.log", json!({"grep":"discarded stdout"})).await;
|
|
assert_eq!(warnings.as_array().unwrap().len(), 1);
|
|
}
|
|
|
|
/// PROTOCOL 8: a failed service enters the journal and does not prevent later services.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn failed_boot_services_are_logged_and_boot_continues() {
|
|
let w = TestWorld::new();
|
|
let services = json!([
|
|
{"method":"absent.start","params":{}},
|
|
{"method":"probe.write","params":{"path":"/state/continued","text":"yes"}}
|
|
]);
|
|
std::fs::write(
|
|
w.0.join("config/boot.yaml"),
|
|
serde_json::to_vec(&services).unwrap(),
|
|
)
|
|
.unwrap();
|
|
let h = w.start().await;
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/continued")).unwrap(),
|
|
"yes"
|
|
);
|
|
let errors = call(
|
|
&h,
|
|
"core.log",
|
|
json!({"plugin":"core","grep":"no such plugin"}),
|
|
)
|
|
.await;
|
|
assert_eq!(errors.as_array().unwrap().len(), 1);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 8: boot services are detached and remain alive after their readiness result.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn boot_services_are_detached() {
|
|
let w = TestWorld::new();
|
|
std::fs::write(
|
|
w.0.join("config/boot.yaml"),
|
|
br#"[{"method":"probe.serve","params":{}}]"#,
|
|
)
|
|
.unwrap();
|
|
let h = w.start().await;
|
|
let processes = h.processes();
|
|
let services: Vec<_> = processes.values().filter(|p| p.method == "serve").collect();
|
|
assert_eq!(services.len(), 1);
|
|
assert!(services[0].detached);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 7: the core records start parameters, stop, kill, and exit status.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn core_lifetime_events_enter_the_journal() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let stopped = call(&h, "probe.serve", json!({"n":1})).await;
|
|
call(&h, "core.stop", json!({"pid":stopped})).await;
|
|
let killed = call(&h, "probe.tree", json!({"depth":0})).await;
|
|
call(&h, "core.kill", json!({"pid":killed})).await;
|
|
let entries = call(&h, "core.log", json!({"plugin":"probe","limit":100})).await;
|
|
let lines: Vec<_> = entries
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.map(|entry| entry["line"].as_str().unwrap())
|
|
.collect();
|
|
assert!(lines.contains(&"start serve {\"n\":1}"));
|
|
for line in ["stop", "kill", "exit 0", "killed"] {
|
|
assert!(lines.contains(&line), "missing {line}");
|
|
}
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: the root merges world files and mounts, and cross-mount rename fails.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_root_and_mount_boundaries() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.filesystem", json!({"topic":"root"})).await;
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/root")).unwrap(),
|
|
"world"
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: a process publishes self files read-only under its own live proc folder.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_self_and_proc_lifetime() {
|
|
use sausage_vfs::Layer;
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let view = h.filesystem().await.unwrap();
|
|
let pid = call(&h, "probe.filesystem", json!({"topic":"self"})).await;
|
|
call(&h, "core.stop", json!({"pid":pid})).await;
|
|
assert!(
|
|
view.stat(&format!("/proc/{}", pid.as_str().unwrap()), true)
|
|
.await
|
|
.is_err()
|
|
);
|
|
let live = call(&h, "probe.self_serve", json!({})).await;
|
|
call(&h, "probe.filesystem", json!({"topic":"isolated"})).await;
|
|
assert_eq!(
|
|
call(
|
|
&h,
|
|
"probe.read",
|
|
json!({"path":format!("/proc/{}/endpoint",live.as_str().unwrap())})
|
|
)
|
|
.await,
|
|
"endpoint"
|
|
);
|
|
call(&h, "core.stop", json!({"pid":live})).await;
|
|
assert!(
|
|
view.stat(&format!("/proc/{}", live.as_str().unwrap()), true)
|
|
.await
|
|
.is_err()
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: core-owned self names reject creation and replacement.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_reserved_self_names() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.filesystem", json!({"topic":"reserved"})).await;
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: generated logs capture a finite snapshot at open.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_log_snapshot_ends() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.filesystem", json!({"topic":"snapshot"})).await;
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: proc/core provides protocol, graph, lock, and plugin inspection files.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_core_inspection_files() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(&h, "probe.filesystem", json!({"topic":"core"})).await;
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: proc/core exposes each discovered document and its generated Rust client.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_discovery_document_and_client() {
|
|
use sausage_vfs::{Layer, Open};
|
|
let w = TestWorld::new();
|
|
let document =
|
|
json!({"methods":[{"name":"echo","params":[],"result":{"schema":{"type":"string"}}}]});
|
|
std::fs::create_dir_all(w.0.join("state/probe")).unwrap();
|
|
std::fs::write(w.0.join("state/probe/discover.json"), document.to_string()).unwrap();
|
|
let h = w.start().await;
|
|
let view = h.filesystem().await.unwrap();
|
|
tokio::time::timeout(std::time::Duration::from_secs(20), async {
|
|
loop {
|
|
if view
|
|
.open("/proc/core/plugins/probe/discover.json", Open::default())
|
|
.await
|
|
.is_ok()
|
|
{
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
let published = call(
|
|
&h,
|
|
"probe.read",
|
|
json!({"path":"/proc/core/plugins/probe/discover.json"}),
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
serde_json::from_str::<Value>(published.as_str().unwrap()).unwrap(),
|
|
document
|
|
);
|
|
let client = call(
|
|
&h,
|
|
"probe.read",
|
|
json!({"path":"/proc/core/plugins/probe/client.rs"}),
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
client,
|
|
sausage_interface::client_rs("probe", &document).unwrap()
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: run is shared across processes but empty after a host restart.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_run_is_shared_and_ephemeral() {
|
|
use sausage_vfs::Layer;
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
call(
|
|
&h,
|
|
"probe.write",
|
|
json!({"path":"/run/shared","text":"yes"}),
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
call(&h, "probe.read", json!({"path":"/run/shared"})).await,
|
|
"yes"
|
|
);
|
|
h.stop().await;
|
|
let h = w.start().await;
|
|
assert!(
|
|
h.filesystem()
|
|
.await
|
|
.unwrap()
|
|
.list("/run")
|
|
.await
|
|
.unwrap()
|
|
.is_empty()
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 6: no write from a killed process lands after the next holder acquires its lock.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_kill_drains_writes_before_lock_transfer() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let pid = call(&h, "probe.write_loop", json!({})).await;
|
|
let next = h.clone();
|
|
let waiter = tokio::spawn(async move { call(&next, "probe.locked_write", json!({})).await });
|
|
use sausage_vfs::{Layer, Open};
|
|
let view = h.filesystem().await.unwrap();
|
|
tokio::time::timeout(std::time::Duration::from_secs(20), async {
|
|
loop {
|
|
let file = view
|
|
.open("/proc/core/locks.json", Open::default())
|
|
.await
|
|
.unwrap();
|
|
let locks: Value = serde_json::from_slice(&file.read(0, 65536).await.unwrap()).unwrap();
|
|
if locks["/state/loop"]["waiting"]
|
|
.as_array()
|
|
.is_some_and(|queue| !queue.is_empty())
|
|
{
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
call(&h, "core.kill", json!({"pid":pid})).await;
|
|
waiter.await.unwrap();
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/loop")).unwrap(),
|
|
"new"
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: native processes appear in proc and locks.json identifies holders and waiters.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn filesystem_native_processes_and_lock_snapshot() {
|
|
use sausage_vfs::{Layer, Open};
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
h.register_native(
|
|
"native",
|
|
std::sync::Arc::new(|ctx, _, _| {
|
|
Box::pin(async move {
|
|
ctx.call("core.lock", json!({"path":"live"})).await.unwrap();
|
|
ctx.notify("ready", json!({"pid":ctx.pid}));
|
|
std::future::pending().await
|
|
})
|
|
}),
|
|
);
|
|
let mut events = h.subscribe();
|
|
let host = h.clone();
|
|
let task = tokio::spawn(async move { host.call("native.wait", json!({})).await });
|
|
let event = events.recv().await.unwrap();
|
|
let pid = event["params"]["params"]["pid"].as_str().unwrap();
|
|
let view = h.filesystem().await.unwrap();
|
|
assert!(
|
|
view.list("/proc")
|
|
.await
|
|
.unwrap()
|
|
.iter()
|
|
.any(|e| e.name == pid)
|
|
);
|
|
let info = call(
|
|
&h,
|
|
"probe.read",
|
|
json!({"path":format!("/proc/{pid}/info.json")}),
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
serde_json::from_str::<Value>(info.as_str().unwrap()).unwrap()["version"],
|
|
"native"
|
|
);
|
|
let host = h.clone();
|
|
let waiter = tokio::spawn(async move { host.call("core.lock", json!({"path":"live"})).await });
|
|
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
|
loop {
|
|
let file = view
|
|
.open("/proc/core/locks.json", Open::default())
|
|
.await
|
|
.unwrap();
|
|
let locks: Value = serde_json::from_slice(&file.read(0, 65536).await.unwrap()).unwrap();
|
|
if locks["live"]["waiting"] == json!(["console"]) {
|
|
assert_eq!(locks["live"]["owner"], pid);
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
call(&h, "core.kill", json!({"pid":pid})).await;
|
|
assert_eq!(waiter.await.unwrap().unwrap(), json!({"held":true}));
|
|
assert!(task.await.unwrap().is_err());
|
|
assert!(view.stat(&format!("/proc/{pid}"), true).await.is_err());
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 2: invalid plugin names cannot route calls, and core cannot be shadowed.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn plugin_names_and_core_reservation() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
for name in ["Upper", "bad!", "with space", "é", "", "../probe"] {
|
|
h.register_native(
|
|
name,
|
|
std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!("bad")) })),
|
|
);
|
|
assert_eq!(
|
|
h.call(&format!("{name}.echo"), json!({}))
|
|
.await
|
|
.unwrap_err()["code"],
|
|
-32601
|
|
);
|
|
}
|
|
h.register_native(
|
|
"core",
|
|
std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!("bad")) })),
|
|
);
|
|
assert_eq!(
|
|
h.call("core.unknown", json!({})).await.unwrap_err()["code"],
|
|
-32601
|
|
);
|
|
for name in ["a-z_09", "0"] {
|
|
h.register_native(
|
|
name,
|
|
std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!(true)) })),
|
|
);
|
|
assert_eq!(call(&h, &format!("{name}.echo"), json!({})).await, true);
|
|
}
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: a batch on guest stdout does not execute any of its requests.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn batches_are_discarded() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(call(&h, "probe.batch", json!({})).await, true);
|
|
h.stop().await;
|
|
assert!(!w.0.join("state/batch").exists());
|
|
assert_eq!(
|
|
call(&h, "core.log", json!({"grep":"discarded stdout"}))
|
|
.await
|
|
.as_array()
|
|
.unwrap()
|
|
.len(),
|
|
1
|
|
);
|
|
}
|
|
|
|
/// PROTOCOL 3.3: a second response to an answered call cannot replace its result.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn late_responses_are_dropped() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
assert_eq!(call(&h, "probe.late_answer", json!({})).await, 42);
|
|
assert_eq!(
|
|
call(&h, "probe.echo", json!({"next":true})).await,
|
|
json!({"next":true})
|
|
);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.4: core.call notifications forget detached children after the caller ends.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn notification_core_call_forgets_detached_child() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let parent = call(&h, "probe.forget", json!({})).await;
|
|
let child: Value = tokio::time::timeout(std::time::Duration::from_secs(10), async {
|
|
loop {
|
|
if let Ok(bytes) = std::fs::read(w.0.join("state/forgotten"))
|
|
&& let Ok(value) = serde_json::from_slice::<Value>(&bytes)
|
|
{
|
|
break value;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
tokio::time::timeout(std::time::Duration::from_secs(10), async {
|
|
while h.processes().contains_key(parent["pid"].as_str().unwrap()) {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert!(h.processes()[child["pid"].as_str().unwrap()].detached);
|
|
assert!(
|
|
h.processes()[child["pid"].as_str().unwrap()]
|
|
.parent
|
|
.is_none()
|
|
);
|
|
call(&h, "core.stop", json!({"pid":child["pid"]})).await;
|
|
assert!(!h.processes().contains_key(child["pid"].as_str().unwrap()));
|
|
h.stop().await;
|
|
}
|
|
|
|
/// Checks the attached-tree half of the shared deadline test.
|
|
async fn attached_tree_stops_in_parallel() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let root = call(&h, "probe.tree", json!({"depth":2})).await;
|
|
assert_eq!(
|
|
h.processes()
|
|
.values()
|
|
.filter(|p| p.method == "tree")
|
|
.count(),
|
|
3
|
|
);
|
|
let before = std::time::Instant::now();
|
|
call(&h, "core.stop", json!({"pid":root})).await;
|
|
tokio::time::timeout(std::time::Duration::from_secs(2), async {
|
|
while h.processes().values().any(|p| p.method == "tree") {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert!(before.elapsed() >= std::time::Duration::from_millis(4900));
|
|
assert!(before.elapsed() < std::time::Duration::from_secs(7));
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.8: a nested call receives its caller's process ID as SAUSAGE_PARENT.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn nested_parent_environment() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let result = call(&h, "probe.nested_environment", json!({})).await;
|
|
assert_eq!(result["child"]["parent"], result["caller"]);
|
|
assert_ne!(result["child"]["pid"], result["caller"]);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 8 and 3.8: boot services receive core as SAUSAGE_PARENT.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn service_parent_environment() {
|
|
let w = TestWorld::new();
|
|
std::fs::write(
|
|
w.0.join("config/boot.yaml"),
|
|
br#"[{"method":"probe.forgotten","params":{}}]"#,
|
|
)
|
|
.unwrap();
|
|
let h = w.start().await;
|
|
let env: Value =
|
|
serde_json::from_slice(&std::fs::read(w.0.join("state/forgotten")).unwrap()).unwrap();
|
|
assert_eq!(env["parent"], "core");
|
|
assert_eq!(env["plugin"], "probe");
|
|
assert!(h.processes()[env["pid"].as_str().unwrap()].detached);
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 4: info.json exposes every field of the live process snapshot.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn info_json_has_all_process_fields() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let params = json!({"marker":"complete"});
|
|
let pid = call(&h, "probe.serve", params.clone()).await;
|
|
let text = call(
|
|
&h,
|
|
"probe.read",
|
|
json!({"path":format!("/proc/{}/info.json", pid.as_str().unwrap())}),
|
|
)
|
|
.await;
|
|
let info: Value = serde_json::from_str(text.as_str().unwrap()).unwrap();
|
|
let expected = serde_json::to_value(&h.processes()[pid.as_str().unwrap()]).unwrap();
|
|
assert_eq!(info, expected);
|
|
for field in [
|
|
"plugin",
|
|
"version",
|
|
"method",
|
|
"params",
|
|
"parent",
|
|
"detached",
|
|
"start_time",
|
|
"state",
|
|
] {
|
|
assert!(info.get(field).is_some(), "{field}");
|
|
}
|
|
assert_eq!(info["params"], params);
|
|
assert_eq!(info["parent"], "console");
|
|
assert_eq!(info["state"], "running");
|
|
h.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.6: detached processes persist until stop, kill, or host shutdown explicitly ends them.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn detached_processes_need_explicit_termination() {
|
|
let w = TestWorld::new();
|
|
let h = w.start().await;
|
|
let mut pids = Vec::new();
|
|
for _ in 0..3 {
|
|
let pid = call(
|
|
&h,
|
|
"core.call",
|
|
json!({"method":"probe.serve","params":{},"detach":true}),
|
|
)
|
|
.await;
|
|
assert!(h.processes()[pid.as_str().unwrap()].parent.is_none());
|
|
pids.push(pid);
|
|
}
|
|
call(&h, "probe.echo", json!({})).await;
|
|
for pid in &pids {
|
|
assert!(h.processes().contains_key(pid.as_str().unwrap()));
|
|
}
|
|
call(&h, "core.stop", json!({"pid":pids[0]})).await;
|
|
call(&h, "core.kill", json!({"pid":pids[1]})).await;
|
|
assert!(h.processes().contains_key(pids[2].as_str().unwrap()));
|
|
h.stop().await;
|
|
for pid in pids {
|
|
assert!(!h.processes().contains_key(pid.as_str().unwrap()));
|
|
}
|
|
}
|
|
|
|
/// PROTOCOL 9: restart answers the detached caller, then uses normal stop and status 75.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn core_restart_cleanly_stops_detached_caller() {
|
|
tokio::time::timeout(std::time::Duration::from_secs(30), async {
|
|
let w = TestWorld::new();
|
|
std::fs::write(
|
|
w.0.join("config/boot.yaml"),
|
|
br#"[{"method":"probe.restart","params":{}}]"#,
|
|
)
|
|
.unwrap();
|
|
let output = tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage"))
|
|
.env("SAUSAGE_LOOP", "1")
|
|
.arg("--headless")
|
|
.arg("--cache-dir")
|
|
.arg(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache"))
|
|
.arg("run")
|
|
.arg(&w.0)
|
|
.stdin(std::process::Stdio::null())
|
|
.kill_on_drop(true)
|
|
.output()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
output.status.code(),
|
|
Some(75),
|
|
"{}",
|
|
String::from_utf8_lossy(&output.stderr)
|
|
);
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/restart-answer")).unwrap(),
|
|
"received"
|
|
);
|
|
assert_eq!(
|
|
std::fs::read_to_string(w.0.join("state/restart-stop")).unwrap(),
|
|
"persisted after EOF"
|
|
);
|
|
let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
|
|
let count: i64 = db
|
|
.query_row(
|
|
"SELECT count(*) FROM journal WHERE line = 'restart stop hook finished'",
|
|
[],
|
|
|row| row.get(0),
|
|
)
|
|
.unwrap();
|
|
assert_eq!(count, 1);
|
|
let killed: i64 = db
|
|
.query_row(
|
|
"SELECT count(*) FROM journal WHERE line = 'kill'",
|
|
[],
|
|
|row| row.get(0),
|
|
)
|
|
.unwrap();
|
|
assert_eq!(killed, 0);
|
|
})
|
|
.await
|
|
.unwrap();
|
|
}
|
|
|
|
/// PROTOCOL 9: supervised core.exit from boot uses the restart shutdown path and code 75.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn core_exit_alias_sets_executable_status_75() {
|
|
let w = TestWorld::new();
|
|
std::fs::write(
|
|
w.0.join("config/boot.yaml"),
|
|
br#"[{"method":"core.exit","params":{}}]"#,
|
|
)
|
|
.unwrap();
|
|
let output = tokio::time::timeout(
|
|
std::time::Duration::from_secs(30),
|
|
tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage"))
|
|
.arg("--headless")
|
|
.arg("--cache-dir")
|
|
.arg(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache"))
|
|
.arg("run")
|
|
.arg(&w.0)
|
|
.env("SAUSAGE_LOOP", "1")
|
|
.stdin(std::process::Stdio::null())
|
|
.kill_on_drop(true)
|
|
.output(),
|
|
)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(
|
|
output.status.code(),
|
|
Some(75),
|
|
"{}",
|
|
String::from_utf8_lossy(&output.stderr)
|
|
);
|
|
}
|