diff --git a/cli/tests/activation.rs b/cli/tests/activation.rs index 31cd6e9..998884f 100644 --- a/cli/tests/activation.rs +++ b/cli/tests/activation.rs @@ -12,12 +12,10 @@ fn fixture(version: usize) -> PathBuf { static FILES: OnceLock> = OnceLock::new(); FILES.get_or_init(|| { let root = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap(); - let target = root.join(format!( - "target/compat/activation-guests/{}", - std::process::id() - )); + let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/activation-guests")); (1..=3) .map(|version| { + let target = target.join(version.to_string()); let status = std::process::Command::new("cargo") .env("FIXTURE_VERSION", version.to_string()) .args([ @@ -67,14 +65,11 @@ impl TestWorld { std::fs::write(self.0.join(format!("state/{name}/1.json")), old.to_string()).unwrap(); std::fs::write(self.0.join(format!("state/{name}/2.json")), new.to_string()).unwrap(); } - /// Starts a host with an independent compile cache shared by these tests. + /// Starts a host with a persistent compile cache shared by the test suites. async fn start(&self) -> Host { Host::start_with_cache( &self.0, - &Path::new(env!("CARGO_MANIFEST_DIR")) - .parent() - .unwrap() - .join("target/compat/runtime-cache"), + Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")), ) .await .unwrap() @@ -85,9 +80,9 @@ impl Drop for TestWorld { let _ = std::fs::remove_dir_all(&self.0); } } -/// Builds a method-only document with a short upgrade grace period. +/// Builds a method-only document without an upgrade delay. fn document(name: &str) -> Value { - json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}],"x-sausage-grace":0.05}) + json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}],"x-sausage-grace":0}) } /// Adds an import of a provider's method list. fn consumer(provider: &str, expected: Value) -> Value { @@ -187,8 +182,7 @@ async fn review_discovery_answer_stops() { marker(world.0.join("state/notes/discovery-ended-2"), "ended").await; host.stop().await; } -/// M4: a rejected candidate's discovery process never becomes a retired installed version. -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +/// Checks that a rejected discovery never receives a retirement hook. async fn review_candidate_is_not_old_version() { use sha2::{Digest, Sha256}; let world = TestWorld::new(); @@ -246,9 +240,7 @@ async fn review_shutdown_suppresses_hooks() { marker(world.0.join("state/notes/service"), "1").await; activate(&host, "notes").await.unwrap(); host.stop().await; - for _ in 0..100 { - tokio::task::yield_now().await; - } + assert!(host.processes().is_empty()); assert!(!world.0.join("state/notes/retired").exists()); assert_eq!( std::fs::read_to_string(world.0.join("state/notes/service")).unwrap(), @@ -270,8 +262,7 @@ async fn review_shutdown_suppresses_hooks() { 1 ); } -/// L15: retirement waits for new admissions of a reverted version, not just its first snapshot. -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +/// Checks retirement against live admissions after a revert. async fn review_retirement_checks_live_version() { let world = TestWorld::new(); let mut old = document("hold"); @@ -340,7 +331,11 @@ async fn review_watcher_is_shallow() { let source = world.0.join("plugins/notes/source"); std::fs::create_dir_all(&source).unwrap(); std::fs::set_permissions(&source, std::fs::Permissions::from_mode(0o0)).unwrap(); - let result = Host::start_with_cache(&world.0, &world.0.join("compile-cache")).await; + let result = Host::start_with_cache( + &world.0, + Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")), + ) + .await; std::fs::set_permissions(&source, std::fs::Permissions::from_mode(0o700)).unwrap(); let host = result.unwrap(); activate(&host, "notes").await.unwrap(); @@ -355,7 +350,11 @@ async fn review_watcher_failure_is_nonfatal() { let inaccessible = world.0.join("plugins/inaccessible"); std::fs::create_dir_all(&inaccessible).unwrap(); std::fs::set_permissions(&inaccessible, std::fs::Permissions::from_mode(0o0)).unwrap(); - let result = Host::start_with_cache(&world.0, &world.0.join("compile-cache")).await; + let result = Host::start_with_cache( + &world.0, + Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")), + ) + .await; std::fs::set_permissions(&inaccessible, std::fs::Permissions::from_mode(0o700)).unwrap(); let host = result.unwrap(); assert!( @@ -368,9 +367,21 @@ async fn review_watcher_failure_is_nonfatal() { activate(&host, "notes").await.unwrap(); host.stop().await; } -/// L14: rapid service upgrades never overlap singleton instances. +/// PROTOCOL 8 (L14): rapid service upgrades never overlap singleton instances. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn review_singleton_restart() { + singleton_restart_cycles(4).await; +} + +/// PROTOCOL 8 (L14): 128 upgrade/revert cycles stress the singleton restart race. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "stress: run by nix flake check"] +async fn singleton_restart_stress() { + singleton_restart_cycles(128).await; +} + +/// Exercises service restarts without duplicating the normal and stress setup. +async fn singleton_restart_cycles(cycles: usize) { let world = TestWorld::new(); world.plugin("notes", document("serve"), document("serve")); std::fs::write( @@ -379,10 +390,12 @@ async fn review_singleton_restart() { ) .unwrap(); let host = world.start().await; - for _ in 0..128 { + for _ in 0..cycles { activate(&host, "notes").await.unwrap(); call(&host, "core.revert", json!({"plugin":"notes"})).await; } + activate(&host, "notes").await.unwrap(); + marker(world.0.join("state/notes/service"), "2").await; host.stop().await; assert!(!world.0.join("state/notes/duplicate-service").exists()); } @@ -422,14 +435,28 @@ async fn review_dropped_activation_completes() { .unwrap(); host.stop().await; } -/// M4: discovery times out and its process stops after either an answer or a timeout. +/// PROTOCOL 10.1: discovery fails after its fixed ten-second deadline and receives EOF. +/// PROTOCOL 3.7: old processes retain their declared grace while new calls use new code. +/// PROTOCOL 3.7: retirement waits for all live admissions, including reverted versions. +/// PROTOCOL 3.7: rejected discovery candidates never receive retirement hooks. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn upgrade_and_discovery_deadlines() { + tokio::join!( + review_discovery_deadline_and_stop(), + old_process_grace_and_new_calls(), + review_retirement_checks_live_version(), + review_candidate_is_not_old_version(), + ); +} + +/// Checks the fixed discovery deadline and subsequent process stop. async fn review_discovery_deadline_and_stop() { let world = TestWorld::new(); let mut next = document("add"); - next["x-test-wait"] = json!(true); + next["x-test-no-answer"] = json!(true); world.plugin("notes", document("add"), next); let host = world.start().await; + let before = std::time::Instant::now(); let result = tokio::time::timeout(Duration::from_secs(15), activate(&host, "notes")).await; assert!( result.unwrap().unwrap_err()["message"] @@ -437,6 +464,7 @@ async fn review_discovery_deadline_and_stop() { .unwrap() .contains("discovery") ); + assert!(before.elapsed() >= Duration::from_millis(9900)); assert!( !host .processes() @@ -720,8 +748,7 @@ async fn graph_missing_import() { assert!(dot.as_str().unwrap().contains("absent")); host.stop().await; } -/// PROTOCOL 3.7: new calls use new code while old calls retain their grace period. -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +/// Checks the declared upgrade grace and dispatch to the new version. async fn old_process_grace_and_new_calls() { let world = TestWorld::new(); let mut old = document("add"); diff --git a/cli/tests/agent.rs b/cli/tests/agent.rs index e5f745b..ddb2891 100644 --- a/cli/tests/agent.rs +++ b/cli/tests/agent.rs @@ -4,6 +4,7 @@ use http_body_util::{BodyExt, Full}; use hyper::{Response, server::conn::http1, service::service_fn}; use hyper_util::rt::TokioIo; use sausage::{Host, world::World}; +use sausage_vfs::Layer; use serde_json::{Value, json}; use std::{convert::Infallible, path::PathBuf, sync::OnceLock, time::Duration}; use tokio::{ @@ -114,7 +115,7 @@ fn fixtures() -> &'static PathBuf { .parent() .unwrap() .to_owned(); - let target = root.join("target/sdk/agent-guests"); + let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests")); for args in [ vec![ "build", @@ -165,7 +166,7 @@ impl TestWorld { async fn start(&self) -> Host { Host::start_with_cache( &self.0, - &PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../target/sdk/agent-cache"), + std::path::Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")), ) .await .unwrap() @@ -349,24 +350,27 @@ async fn concurrent_chats_serialize_complete_turns() { let first = chat(&host, "first concurrent request"); let (_, reply) = server.next().await; let second = chat(&host, "second concurrent request"); + let view = host.filesystem().await.unwrap(); tokio::time::timeout(Duration::from_secs(10), async { - while host - .processes() - .values() - .filter(|p| p.plugin == "agent" && p.method == "chat") - .count() - != 2 - { + loop { + let file = view + .open("/proc/core/locks.json", sausage_vfs::Open::default()) + .await + .unwrap(); + let locks: Value = + serde_json::from_slice(&file.read(0, 65536).await.unwrap()).unwrap(); + if locks["/state/agent/history.jsonl"]["waiting"] + .as_array() + .is_some_and(|waiting| waiting.len() == 1) + { + break; + } tokio::task::yield_now().await; } }) .await .unwrap(); - assert!( - tokio::time::timeout(Duration::from_millis(200), server.requests.recv()) - .await - .is_err() - ); + assert!(server.requests.try_recv().is_err()); reply .send((200, server.answer("first concurrent answer", &[]))) .unwrap(); diff --git a/cli/tests/console.rs b/cli/tests/console.rs index d5fb6c7..9dbba31 100644 --- a/cli/tests/console.rs +++ b/cli/tests/console.rs @@ -14,7 +14,7 @@ fn fixture() -> PathBuf { .parent() .unwrap() .to_owned(); - let target = root.join("target/core/console-guests"); + let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests")); let status = std::process::Command::new("cargo") .args([ "build", @@ -51,10 +51,7 @@ impl TestWorld { Self(root) } async fn start(&self) -> Host { - let cache = PathBuf::from(env!("CARGO_MANIFEST_DIR")) - .parent() - .unwrap() - .join("target/core/console-cache"); + let cache = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")); Host::start_with_cache(&self.0, &cache).await.unwrap() } } diff --git a/cli/tests/core.rs b/cli/tests/core.rs index 25b1683..0b6cec0 100644 --- a/cli/tests/core.rs +++ b/cli/tests/core.rs @@ -9,7 +9,7 @@ fn fixture() -> PathBuf { .parent() .unwrap() .to_path_buf(); - let target = root.join("target/core/guests"); + let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests")); let status = std::process::Command::new("cargo") .args([ "build", @@ -38,7 +38,12 @@ impl TestWorld { Self(p) } async fn start(&self) -> Host { - Host::start(&self.0).await.unwrap() + Host::start_with_cache( + &self.0, + std::path::Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")), + ) + .await + .unwrap() } } impl Drop for TestWorld { @@ -902,8 +907,17 @@ async fn native_kill_drops_future_and_releases_locks() { h.stop().await; } -/// PROTOCOL 3.6: shutdown stops busy processes in parallel after five seconds. +/// 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; @@ -1132,16 +1146,31 @@ async fn two_plugins_execute_calls_at_the_same_time() { 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.sleep", json!({"ms":1500})).await }); - while !h.processes().values().any(|p| p.method == "sleep") { - tokio::task::yield_now().await; - } + 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()); - slow.await.unwrap().unwrap(); + call(&h, "core.stop", json!({"pid":pid})).await; + assert_eq!(slow.await.unwrap().unwrap_err()["code"], -32002); h.stop().await; } @@ -1809,8 +1838,7 @@ async fn notification_core_call_forgets_detached_child() { h.stop().await; } -/// PROTOCOL 3.6: three generations that ignore EOF stop together within one five-second deadline. -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +/// 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; @@ -1941,6 +1969,8 @@ async fn core_exit_sets_executable_status_75() { 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) .stdin(std::process::Stdio::null()) diff --git a/cli/tests/plugins/activation/src/main.rs b/cli/tests/plugins/activation/src/main.rs index 8924562..cf361f2 100644 --- a/cli/tests/plugins/activation/src/main.rs +++ b/cli/tests/plugins/activation/src/main.rs @@ -11,8 +11,9 @@ fn deferred(request: &Value, plugin: &str, lines: &mut impl Iterator { + if document["x-test-no-answer"] == true { + for _ in &mut lines {} + return; + } if document["x-test-wait"] == true { std::fs::write(format!("/state/{plugin}/discover-started"), "ready").unwrap(); + let before = std::time::Instant::now(); while !std::path::Path::new(&format!("/state/{plugin}/discover-release")).exists() { - std::thread::sleep(std::time::Duration::from_millis(1)); + assert!(before.elapsed() < std::time::Duration::from_secs(30)); } } document @@ -105,8 +111,9 @@ fn main() { send(json!({"jsonrpc":"2.0","id":request["id"],"result":result})); if request["method"] == "rpc.discover" && result["x-test-linger"] == true { if result["x-test-stubborn"] == true { + let before = std::time::Instant::now(); while !std::path::Path::new(&format!("/state/{plugin}/discovery-exit")).exists() { - std::thread::sleep(std::time::Duration::from_millis(1)); + assert!(before.elapsed() < std::time::Duration::from_secs(30)); } } else { for _ in lines {} diff --git a/cli/tests/plugins/probe/src/main.rs b/cli/tests/plugins/probe/src/main.rs index c4ab3ca..0a8182f 100644 --- a/cli/tests/plugins/probe/src/main.rs +++ b/cli/tests/plugins/probe/src/main.rs @@ -18,7 +18,6 @@ fn chat(params: &Value) -> Value { send(json!({"jsonrpc":"2.0", "method":"core.notify", "params":{ "name":"text", "params":{"text":"interim"} }})); - std::thread::sleep(std::time::Duration::from_millis(30)); json!({"text":params["text"]}) } fn discover(params: &Value) -> Value { @@ -65,12 +64,13 @@ fn notification(params: &Value) -> Value { "method": "probe.write", "params": params })); - for _ in 0..200 { + let before = std::time::Instant::now(); + loop { let contents = std::fs::read_to_string(params["path"].as_str().unwrap()).ok(); if contents.as_deref() == params["text"].as_str() { break; } - std::thread::sleep(std::time::Duration::from_millis(10)); + assert!(before.elapsed() < std::time::Duration::from_secs(10)); } json!({}) } @@ -82,6 +82,7 @@ fn unknown_id() { })); } fn wait_eof(lines: &mut Lines) { + std::fs::write("/state/waiting", "yes").unwrap(); for _ in lines {} std::fs::write("/state/stopped", "yes").unwrap(); } @@ -90,11 +91,6 @@ fn busy() -> Value { std::hint::spin_loop(); } } -fn sleep(params: &Value) -> Value { - let duration = std::time::Duration::from_millis(params["ms"].as_u64().unwrap_or(100)); - std::thread::sleep(duration); - json!({}) -} fn exit_without_answer() { eprintln!("last line"); } @@ -140,7 +136,6 @@ fn notify() -> Value { "params": {"n": 1} } })); - std::thread::sleep(std::time::Duration::from_millis(30)); json!({}) } fn serve(request: &Value, lines: &mut Lines) { @@ -288,7 +283,7 @@ fn locked_write(lines: &mut Lines) -> Value { json!({"held":true}) ); std::fs::write("/state/loop", "new").unwrap(); - std::thread::sleep(std::time::Duration::from_millis(100)); + outgoing_call(&json!({"method":"probe.echo","params":{}}), lines); assert_eq!(std::fs::read_to_string("/state/loop").unwrap(), "new"); json!({}) } @@ -305,11 +300,12 @@ fn unlock_answer(request: &Value, lines: &mut Lines) { } fn notification_child(lines: &mut Lines) -> Value { send(json!({"jsonrpc":"2.0","method":"probe.orphan_notify","params":{}})); + let before = std::time::Instant::now(); let pid = loop { if let Ok(pid) = std::fs::read_to_string("/state/orphan") { break pid; } - std::thread::sleep(std::time::Duration::from_millis(1)); + assert!(before.elapsed() < std::time::Duration::from_secs(10)); }; send(json!({"jsonrpc":"2.0","id":7,"method":"core.stop","params":{"pid":pid}})); let mut count = 0; @@ -351,8 +347,9 @@ fn forget(lines: &mut Lines) -> Value { send( json!({"jsonrpc":"2.0","method":"core.call","params":{"method":"probe.forgotten","params":{},"detach":true}}), ); + let before = std::time::Instant::now(); while !std::path::Path::new("/state/forgotten").exists() { - std::thread::sleep(std::time::Duration::from_millis(1)); + assert!(before.elapsed() < std::time::Duration::from_secs(10)); } send(json!({"jsonrpc":"2.0","id":"barrier","method":"probe.echo","params":{}})); let response = receive(lines); @@ -438,7 +435,6 @@ fn dispatch(request: &Value, lines: &mut Lines) -> Option { return None; } "busy" => busy(), - "sleep" => sleep(params), "exit" => { exit_without_answer(); return None; diff --git a/cli/tests/sdk.rs b/cli/tests/sdk.rs index 22cfa2a..a9d74e4 100644 --- a/cli/tests/sdk.rs +++ b/cli/tests/sdk.rs @@ -10,7 +10,7 @@ fn fixtures() -> &'static PathBuf { .parent() .unwrap() .to_owned(); - let target = root.join("target/sdk/real-guests"); + let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests")); for package in [ "examples/notes", "examples/consumer", @@ -63,7 +63,7 @@ impl TestWorld { async fn start(&self) -> Host { Host::start_with_cache( &self.0, - &PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../target/sdk/real-cache"), + std::path::Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")), ) .await .unwrap()