surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tests: fast host suites

Browse Source 
The timing rules are in two tests (upgrade_and_discovery_deadlines,
stop_deadlines). All other tests use a grace period of 0 or core.stop,
poll with deadlines instead of sleeping, and use the persistent cache in
CARGO_TARGET_TMPDIR. The 128-cycle singleton test is an ignored stress
test that nix flake check runs; the normal run has a 4-cycle sample.
Warm, the activation suite takes 12 s (was 47 s), and cargo test
--workspace takes 65 s.
This commit is contained in:
surma
2026-10-23 08:20:49 +01:00
parent 1386acc924
commit 41818107a2
7 changed files with 134 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/tests/activation.rs
+53 -26
View File
@@ -12,12 +12,10 @@ fn fixture(version: usize) -> PathBuf {
static FILES: OnceLock<Vec<PathBuf>> = 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");
cli/tests/agent.rs
+18 -14
View File
@@ -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();
cli/tests/console.rs
+2 -5
View File
@@ -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()
}
}
cli/tests/core.rs
+40 -10
View File
@@ -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())
cli/tests/plugins/activation/src/main.rs
+10 -3
View File
@@ -11,8 +11,9 @@ fn deferred(request: &Value, plugin: &str, lines: &mut impl Iterator<Item = io::
send(
json!({"jsonrpc":"2.0","id":request["id"],"result":std::env::var("SAUSAGE_PID").unwrap()}),
);
let before = std::time::Instant::now();
while !std::path::Path::new(&format!("/state/{plugin}/release-call")).exists() {
std::thread::sleep(std::time::Duration::from_millis(1));
assert!(before.elapsed() < std::time::Duration::from_secs(30));
}
send(json!({"jsonrpc":"2.0","id":"deferred","method":"notes.legacy","params":{}}));
for line in lines {
@@ -57,10 +58,15 @@ fn main() {
return;
}
"rpc.discover" => {
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 {}
cli/tests/plugins/probe/src/main.rs
+9 -13
View File
@@ -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<Value> {
return None;
}
"busy" => busy(),
"sleep" => sleep(params),
"exit" => {
exit_without_answer();
return None;
cli/tests/sdk.rs
+2 -2
View File
@@ -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()