From bfd9339d25030df3a9de4ce44736737ec6daaeab Mon Sep 17 00:00:00 2001 From: Surma Date: Thu, 15 Oct 2026 11:24:07 +0100 Subject: [PATCH] Integration tests for routing, filesystem, upgrades, rollback, cargo, and paths --- cli/tests/actors/toolbox/Cargo.lock | 105 ++++ cli/tests/actors/toolbox/Cargo.toml | 9 + cli/tests/actors/toolbox/src/main.rs | 111 ++++ cli/tests/common/mod.rs | 216 ++++++++ cli/tests/crates/cargo-fail/Cargo.toml | 6 + cli/tests/crates/cargo-fail/src/main.rs | 3 + cli/tests/crates/cargo-valid/Cargo.toml | 6 + cli/tests/crates/cargo-valid/src/main.rs | 1 + cli/tests/host.rs | 641 +++++++++++++++++++++++ 9 files changed, 1098 insertions(+) create mode 100644 cli/tests/actors/toolbox/Cargo.lock create mode 100644 cli/tests/actors/toolbox/Cargo.toml create mode 100644 cli/tests/actors/toolbox/src/main.rs create mode 100644 cli/tests/common/mod.rs create mode 100644 cli/tests/crates/cargo-fail/Cargo.toml create mode 100644 cli/tests/crates/cargo-fail/src/main.rs create mode 100644 cli/tests/crates/cargo-valid/Cargo.toml create mode 100644 cli/tests/crates/cargo-valid/src/main.rs create mode 100644 cli/tests/host.rs diff --git a/cli/tests/actors/toolbox/Cargo.lock b/cli/tests/actors/toolbox/Cargo.lock new file mode 100644 index 0000000..7b62d62 --- /dev/null +++ b/cli/tests/actors/toolbox/Cargo.lock @@ -0,0 +1,105 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.154" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "syn" +version = "3.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "toolbox" +version = "0.1.0" +dependencies = [ + "serde_json", +] + +[[package]] +name = "unicode-ident" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/cli/tests/actors/toolbox/Cargo.toml b/cli/tests/actors/toolbox/Cargo.toml new file mode 100644 index 0000000..5a9a35e --- /dev/null +++ b/cli/tests/actors/toolbox/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "toolbox" +version = "0.1.0" +edition = "2024" + +[dependencies] +serde_json = "1" + +[workspace] diff --git a/cli/tests/actors/toolbox/src/main.rs b/cli/tests/actors/toolbox/src/main.rs new file mode 100644 index 0000000..133fad1 --- /dev/null +++ b/cli/tests/actors/toolbox/src/main.rs @@ -0,0 +1,111 @@ +use std::collections::HashMap; +use std::io::{self, BufRead, Write}; +use std::time::Duration; + +use serde_json::{Value, json}; + +fn reply(out: &mut impl Write, message: &Value, body: Value) { + writeln!( + out, + "{}", + json!({"to": message["from"], "reply_to": message["id"], "body": body}) + ) + .unwrap(); + out.flush().unwrap(); +} + +fn io_error(error: io::Error) -> Value { + json!({"type": "error", "error": error.to_string()}) +} + +fn main() { + let actor = std::env::var("SAUSAGE_ACTOR").unwrap_or_default(); + let version = std::env::var("SAUSAGE_VERSION").unwrap_or_else(|_| "candidate".to_string()); + if actor == "crashloop" && version == "2" { + std::fs::write("/data/state", "bad-v2").unwrap(); + panic!("version 2 crashes on startup"); + } + + let mut open = HashMap::::new(); + let stdin = io::stdin(); + let mut out = io::stdout().lock(); + for line in stdin.lock().lines() { + let Ok(line) = line else { break }; + let Ok(message) = serde_json::from_str::(&line) else { + continue; + }; + let body = &message["body"]; + let path = body["path"].as_str().unwrap_or_default().to_string(); + let text = body["text"].as_str().unwrap_or_default(); + let response = match body["type"].as_str() { + Some("shutdown") => { + if actor == "queue" { + let _ = std::fs::write("/data/shutdown", "stopping"); + std::thread::sleep(Duration::from_millis(800)); + } + break; + } + Some("echo") => json!({"type": "echo", "value": body["value"]}), + Some("publish") => { + writeln!( + out, + "{}", + json!({"body": {"type": "published", "value": body["value"]}}) + ) + .unwrap(); + out.flush().unwrap(); + continue; + } + Some("version") => json!({"type": "version", "version": version}), + Some("ordered") => json!({"type": "ordered", "n": body["n"], "version": version}), + Some("write") => match std::fs::write(&path, text) { + Ok(()) => json!({"type": "ok"}), + Err(error) => io_error(error), + }, + Some("open") => { + let result = (|| -> io::Result<()> { + let mut file = std::fs::File::create(&path)?; + file.write_all(text.as_bytes())?; + open.insert(path.clone(), file); + Ok(()) + })(); + match result { + Ok(()) => json!({"type": "opened"}), + Err(error) => io_error(error), + } + } + Some("append") => { + let result = (|| -> io::Result<()> { + let mut file = std::fs::OpenOptions::new().append(true).open(&path)?; + file.write_all(text.as_bytes())?; + open.insert(path.clone(), file); + Ok(()) + })(); + match result { + Ok(()) => json!({"type": "opened"}), + Err(error) => io_error(error), + } + } + Some("close") => json!({"type": "closed", "closed": open.remove(&path).is_some()}), + Some("read") => match std::fs::read_to_string(&path) { + Ok(text) => json!({"type": "content", "text": text}), + Err(error) => io_error(error), + }, + Some("list") => match std::fs::read_dir(&path) { + Ok(entries) => { + let mut names: Vec = entries + .filter_map(|entry| entry.ok()) + .map(|entry| entry.file_name().to_string_lossy().to_string()) + .collect(); + names.sort(); + json!({"type": "entries", "names": names}) + } + Err(error) => io_error(error), + }, + Some("crash") if version == "1" => panic!("asked to crash"), + Some("crash") => json!({"type": "replayed", "version": version}), + _ => json!({"type": "error", "error": "unknown message"}), + }; + reply(&mut out, &message, response); + } +} diff --git a/cli/tests/common/mod.rs b/cli/tests/common/mod.rs new file mode 100644 index 0000000..cc695cc --- /dev/null +++ b/cli/tests/common/mod.rs @@ -0,0 +1,216 @@ +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::sync::OnceLock; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use sausage::{Envelope, Host, World}; +use serde_json::{Value, json}; +use tokio::sync::mpsc; + +const REPLY_TIMEOUT: Duration = Duration::from_secs(10); +static WORLD_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +pub fn repo_root() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .unwrap() + .to_path_buf() +} + +pub fn create_world(label: &str) -> World { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let sequence = WORLD_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let path = repo_root() + .join("target/test-worlds") + .join(format!("{label}-{}-{nonce}-{sequence}", std::process::id())); + World::create(&path).unwrap() +} + +pub fn actor_fixture_source(name: &str) -> PathBuf { + repo_root().join("cli/tests/actors").join(name) +} + +pub fn cargo_source(name: &str) -> PathBuf { + repo_root().join("cli/tests/crates").join(name) +} + +pub fn toolbox_wasm() -> &'static PathBuf { + static WASM: OnceLock = OnceLock::new(); + WASM.get_or_init(|| build_fixture("toolbox")) +} + +fn build_fixture(name: &str) -> PathBuf { + let root = repo_root(); + let manifest = actor_fixture_source(name).join("Cargo.toml"); + let target_dir = root.join("target/test-actors"); + let output = Command::new("cargo") + .args([ + "build", + "--release", + "--target", + "wasm32-wasip2", + "--target-dir", + ]) + .arg(&target_dir) + .arg("--manifest-path") + .arg(&manifest) + .current_dir(&root) + .output() + .expect("run cargo to build the test actor"); + assert!( + output.status.success(), + "building fixture {name} failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + let wasm = target_dir + .join("wasm32-wasip2/release") + .join(format!("{name}.wasm")); + assert!( + wasm.is_file(), + "fixture build did not create {}", + wasm.display() + ); + wasm +} + +pub fn passing_source(world: &World, name: &str) -> PathBuf { + source_with_test( + world, + name, + json!({ + "steps": [{ + "send": {"type": "echo", "value": "candidate"}, + "expect": {"type": "echo", "value": "candidate"} + }] + }), + ) +} + +pub fn source_with_test(world: &World, name: &str, test: Value) -> PathBuf { + let source = world.source_dir("console").join("test-fixtures").join(name); + std::fs::create_dir_all(source.join("tests")).unwrap(); + std::fs::write( + source.join("tests/candidate.json"), + serde_json::to_vec(&test).unwrap(), + ) + .unwrap(); + source +} + +pub async fn start(world: &World) -> (Host, mpsc::Receiver) { + let host = Host::start(&world.root).await.unwrap(); + let mailbox = host.mailbox("console"); + (host, mailbox) +} + +pub async fn request(host: &Host, mailbox: &mut mpsc::Receiver, body: Value) -> Envelope { + let id = host + .broker + .send("console", Some("broker"), None, None, body); + receive_reply(mailbox, &id, REPLY_TIMEOUT).await +} + +pub async fn receive_message( + mailbox: &mut mpsc::Receiver, + timeout: Duration, +) -> Envelope { + tokio::time::timeout(timeout, mailbox.recv()) + .await + .expect("timed out waiting for a message") + .expect("console mailbox closed") +} + +pub async fn receive_reply( + mailbox: &mut mpsc::Receiver, + id: &str, + timeout: Duration, +) -> Envelope { + tokio::time::timeout(timeout, async { + loop { + let envelope = mailbox.recv().await.expect("console mailbox closed"); + if envelope.reply_to.as_deref() == Some(id) { + return envelope; + } + } + }) + .await + .unwrap_or_else(|_| panic!("timed out waiting for reply to {id}")) +} + +pub async fn ask( + host: &Host, + mailbox: &mut mpsc::Receiver, + actor: &str, + body: Value, +) -> Envelope { + let id = host.broker.send("console", Some(actor), None, None, body); + receive_reply(mailbox, &id, REPLY_TIMEOUT).await +} + +pub async fn spawn_actor( + host: &Host, + mailbox: &mut mpsc::Receiver, + actor: &str, + wasm: &Path, + source: &Path, +) -> Envelope { + request( + host, + mailbox, + json!({ + "type": "spawn", + "actor": actor, + "wasm": wasm.display().to_string(), + "source": source.display().to_string() + }), + ) + .await +} + +pub async fn upgrade_actor( + host: &Host, + mailbox: &mut mpsc::Receiver, + actor: &str, + wasm: &Path, + source: &Path, +) -> Envelope { + request( + host, + mailbox, + json!({ + "type": "upgrade", + "actor": actor, + "wasm": wasm.display().to_string(), + "source": source.display().to_string() + }), + ) + .await +} + +pub async fn wait_for_parked(world: &World, actor: &str, message_id: &str) { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let status_path = world.actor_dir(actor).join("status.json"); + let parked_id = std::fs::read_to_string(status_path) + .ok() + .and_then(|text| serde_json::from_str::(&text).ok()) + .and_then(|status| { + status + .pointer("/parked_message/id") + .and_then(Value::as_str) + .map(str::to_string) + }); + if parked_id.as_deref() == Some(message_id) { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("actor did not park the crashed message"); +} diff --git a/cli/tests/crates/cargo-fail/Cargo.toml b/cli/tests/crates/cargo-fail/Cargo.toml new file mode 100644 index 0000000..6403b77 --- /dev/null +++ b/cli/tests/crates/cargo-fail/Cargo.toml @@ -0,0 +1,6 @@ +[package] +name = "cargo-fail" +version = "0.1.0" +edition = "2024" + +[workspace] diff --git a/cli/tests/crates/cargo-fail/src/main.rs b/cli/tests/crates/cargo-fail/src/main.rs new file mode 100644 index 0000000..57d1382 --- /dev/null +++ b/cli/tests/crates/cargo-fail/src/main.rs @@ -0,0 +1,3 @@ +fn main() { + missing_function(); +} diff --git a/cli/tests/crates/cargo-valid/Cargo.toml b/cli/tests/crates/cargo-valid/Cargo.toml new file mode 100644 index 0000000..8355632 --- /dev/null +++ b/cli/tests/crates/cargo-valid/Cargo.toml @@ -0,0 +1,6 @@ +[package] +name = "cargo-valid" +version = "0.1.0" +edition = "2024" + +[workspace] diff --git a/cli/tests/crates/cargo-valid/src/main.rs b/cli/tests/crates/cargo-valid/src/main.rs new file mode 100644 index 0000000..f328e4d --- /dev/null +++ b/cli/tests/crates/cargo-valid/src/main.rs @@ -0,0 +1 @@ +fn main() {} diff --git a/cli/tests/host.rs b/cli/tests/host.rs new file mode 100644 index 0000000..3ba74d6 --- /dev/null +++ b/cli/tests/host.rs @@ -0,0 +1,641 @@ +mod common; + +use std::time::Duration; + +use common::*; +use sausage::fs::TMP_PREFIX; +use sausage::{Envelope, Host, World}; +use serde_json::{Value, json}; +use tokio::sync::mpsc; + +async fn filesystem_world(label: &str) -> (World, Host, mpsc::Receiver) { + let world = create_world(label); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + for actor in ["writer", "reader"] { + let source = passing_source(&world, actor); + let spawned = spawn_actor(&host, &mut mailbox, actor, &wasm, &source).await; + assert_eq!(spawned.body["type"], "spawned"); + } + (world, host, mailbox) +} + +#[tokio::test] +async fn routes_replies_publishes_and_reports_undeliverable() { + let world = create_world("routing"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + + let missing_id = host.broker.send( + "console", + Some("missing"), + None, + None, + json!({"type": "hello"}), + ); + let missing = receive_reply(&mut mailbox, &missing_id, Duration::from_secs(10)).await; + assert_eq!(missing.body["type"], "undeliverable"); + assert_eq!(missing.body["to"], "missing"); + + let source = passing_source(&world, "router"); + let spawned = spawn_actor(&host, &mut mailbox, "router", &wasm, &source).await; + assert_eq!(spawned.body["type"], "spawned"); + + let echo = ask( + &host, + &mut mailbox, + "router", + json!({"type": "echo", "value": "hello"}), + ) + .await; + assert_eq!(echo.from, "router"); + assert_eq!(echo.body, json!({"type": "echo", "value": "hello"})); + + let subscribed = request( + &host, + &mut mailbox, + json!({"type": "subscribe", "actor": "router"}), + ) + .await; + assert_eq!(subscribed.body["type"], "subscribed"); + + host.broker.send( + "console", + Some("router"), + None, + None, + json!({"type": "publish", "value": "news"}), + ); + let published = receive_message(&mut mailbox, Duration::from_secs(10)).await; + assert_eq!(published.from, "router"); + assert_eq!( + published.body, + json!({"type": "published", "value": "news"}) + ); + assert_eq!(published.reply_to, None); + + host.stop().await; +} + +#[tokio::test] +async fn filesystem_permissions_and_public_sharing() { + let (_, host, mut mailbox) = filesystem_world("filesystem-permissions").await; + for path in [ + "/actor/reader/public/denied.txt", + "/actor/writer/public/denied.txt", + ] { + let denied = ask( + &host, + &mut mailbox, + "writer", + json!({"type": "write", "path": path, "text": "denied"}), + ) + .await; + assert_eq!(denied.body["type"], "error", "write to {path} succeeded"); + } + + let written = ask( + &host, + &mut mailbox, + "writer", + json!({"type": "write", "path": "/public/shared.txt", "text": "public"}), + ) + .await; + assert_eq!(written.body["type"], "ok"); + let shared = ask( + &host, + &mut mailbox, + "reader", + json!({"type": "read", "path": "/actor/writer/public/shared.txt"}), + ) + .await; + assert_eq!(shared.body, json!({"type": "content", "text": "public"})); + host.stop().await; +} + +#[tokio::test] +async fn filesystem_hides_new_and_truncated_writes_until_close() { + let (_, host, mut mailbox) = filesystem_world("filesystem-commit").await; + let opened = ask( + &host, + &mut mailbox, + "writer", + json!({"type": "open", "path": "/public/new.txt", "text": "new"}), + ) + .await; + assert_eq!(opened.body["type"], "opened"); + let invisible = ask( + &host, + &mut mailbox, + "reader", + json!({"type": "read", "path": "/actor/writer/public/new.txt"}), + ) + .await; + assert_eq!(invisible.body["type"], "error"); + let listing = ask( + &host, + &mut mailbox, + "writer", + json!({"type": "list", "path": "/public"}), + ) + .await; + let names = listing.body["names"].as_array().unwrap(); + assert!( + names + .iter() + .all(|name| !name.as_str().unwrap().starts_with(TMP_PREFIX)) + ); + assert!(names.iter().all(|name| name.as_str() != Some("new.txt"))); + let closed = ask( + &host, + &mut mailbox, + "writer", + json!({"type": "close", "path": "/public/new.txt"}), + ) + .await; + assert_eq!(closed.body["closed"], true); + let visible = ask( + &host, + &mut mailbox, + "reader", + json!({"type": "read", "path": "/actor/writer/public/new.txt"}), + ) + .await; + assert_eq!(visible.body["text"], "new"); + + ask( + &host, + &mut mailbox, + "writer", + json!({"type": "write", "path": "/public/truncated.txt", "text": "old"}), + ) + .await; + ask( + &host, + &mut mailbox, + "writer", + json!({"type": "open", "path": "/public/truncated.txt", "text": "new"}), + ) + .await; + let old_during_truncate = ask( + &host, + &mut mailbox, + "reader", + json!({"type": "read", "path": "/actor/writer/public/truncated.txt"}), + ) + .await; + assert_eq!(old_during_truncate.body["text"], "old"); + ask( + &host, + &mut mailbox, + "writer", + json!({"type": "close", "path": "/public/truncated.txt"}), + ) + .await; + let new_after_truncate = ask( + &host, + &mut mailbox, + "reader", + json!({"type": "read", "path": "/actor/writer/public/truncated.txt"}), + ) + .await; + assert_eq!(new_after_truncate.body["text"], "new"); + host.stop().await; +} + +#[tokio::test] +async fn filesystem_append_is_visible_before_close() { + let (_, host, mut mailbox) = filesystem_world("filesystem-append").await; + ask( + &host, + &mut mailbox, + "writer", + json!({"type": "write", "path": "/public/append.txt", "text": "base"}), + ) + .await; + ask( + &host, + &mut mailbox, + "writer", + json!({"type": "append", "path": "/public/append.txt", "text": "+tail"}), + ) + .await; + let appended = ask( + &host, + &mut mailbox, + "reader", + json!({"type": "read", "path": "/actor/writer/public/append.txt"}), + ) + .await; + assert_eq!(appended.body["text"], "base+tail"); + let closed = ask( + &host, + &mut mailbox, + "writer", + json!({"type": "close", "path": "/public/append.txt"}), + ) + .await; + assert_eq!(closed.body["closed"], true); + host.stop().await; +} + +#[tokio::test] +async fn filesystem_crash_discards_an_open_truncated_write() { + let (world, host, mut mailbox) = filesystem_world("filesystem-crash").await; + ask( + &host, + &mut mailbox, + "writer", + json!({"type": "write", "path": "/public/crash.txt", "text": "old"}), + ) + .await; + ask( + &host, + &mut mailbox, + "writer", + json!({"type": "open", "path": "/public/crash.txt", "text": "uncommitted"}), + ) + .await; + let crash_id = host.broker.send( + "console", + Some("writer"), + None, + None, + json!({"type": "crash"}), + ); + wait_for_parked(&world, "writer", &crash_id).await; + let after_crash = ask( + &host, + &mut mailbox, + "reader", + json!({"type": "read", "path": "/actor/writer/public/crash.txt"}), + ) + .await; + assert_eq!(after_crash.body["text"], "old"); + host.stop().await; +} + +#[tokio::test] +async fn filesystem_rejects_paths_that_escape_data() { + let (_, host, mut mailbox) = filesystem_world("filesystem-path-escape").await; + let escaped = ask( + &host, + &mut mailbox, + "writer", + json!({"type": "write", "path": "/data/../../..", "text": "outside"}), + ) + .await; + assert_eq!(escaped.body["type"], "error"); + host.stop().await; +} + +#[tokio::test] +async fn upgrades_test_candidates_keep_old_versions_and_queue_messages_in_order() { + let world = create_world("upgrade"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let passing = passing_source(&world, "passing"); + let failing = source_with_test( + &world, + "failing", + json!({ + "steps": [{ + "send": {"type": "echo", "value": "candidate"}, + "expect": {"type": "wrong"} + }] + }), + ); + assert_eq!( + spawn_actor(&host, &mut mailbox, "queue", &wasm, &passing) + .await + .body["type"], + "spawned" + ); + + let failed = upgrade_actor(&host, &mut mailbox, "queue", &wasm, &failing).await; + assert_eq!(failed.body["type"], "error"); + let old_version = ask(&host, &mut mailbox, "queue", json!({"type": "version"})).await; + assert_eq!(old_version.body["version"], "1"); + + let upgrade_id = host.broker.send( + "console", + Some("broker"), + None, + None, + json!({"type": "upgrade", "actor": "queue", "wasm": wasm.display().to_string(), "source": passing.display().to_string()}), + ); + tokio::time::timeout(Duration::from_secs(10), async { + while !world.data_dir("queue").join("shutdown").is_file() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("old actor did not enter shutdown during upgrade"); + + let mut message_ids = Vec::new(); + for n in 1..=3 { + message_ids.push(host.broker.send( + "console", + Some("queue"), + None, + None, + json!({"type": "ordered", "n": n}), + )); + } + + let mut upgrade_result = None; + let mut actor_replies = Vec::new(); + tokio::time::timeout(Duration::from_secs(15), async { + while upgrade_result.is_none() || actor_replies.len() < message_ids.len() { + let envelope = mailbox.recv().await.expect("console mailbox closed"); + if envelope.reply_to.as_deref() == Some(upgrade_id.as_str()) { + upgrade_result = Some(envelope); + } else if message_ids + .iter() + .any(|id| envelope.reply_to.as_deref() == Some(id.as_str())) + { + actor_replies.push(envelope); + } else { + panic!("unexpected message during queued upgrade: {envelope:?}"); + } + } + }) + .await + .expect("upgrade did not deliver queued messages"); + + assert_eq!(upgrade_result.unwrap().body["type"], "upgraded"); + let actual_ids: Vec = actor_replies + .iter() + .map(|message| message.reply_to.clone().unwrap()) + .collect(); + assert_eq!(actual_ids, message_ids); + for (index, envelope) in actor_replies.iter().enumerate() { + assert_eq!(envelope.from, "queue"); + assert_eq!(envelope.body["type"], "ordered"); + assert_eq!(envelope.body["n"], json!(index + 1)); + assert_eq!(envelope.body["version"], "3"); + } + + host.stop().await; +} + +#[tokio::test] +async fn redelivers_a_crashed_message_after_the_next_upgrade() { + let world = create_world("redelivery"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + assert_eq!( + spawn_actor(&host, &mut mailbox, "parked", &wasm, &source) + .await + .body["type"], + "spawned" + ); + + let crash_id = host.broker.send( + "console", + Some("parked"), + None, + None, + json!({"type": "crash"}), + ); + wait_for_parked(&world, "parked", &crash_id).await; + assert!(matches!( + mailbox.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + )); + + let upgrade_id = host.broker.send( + "console", + Some("broker"), + None, + None, + json!({"type": "upgrade", "actor": "parked", "wasm": wasm.display().to_string(), "source": source.display().to_string()}), + ); + let mut upgraded = None; + let mut replayed = None; + tokio::time::timeout(Duration::from_secs(15), async { + while upgraded.is_none() || replayed.is_none() { + let envelope = mailbox.recv().await.expect("console mailbox closed"); + if envelope.reply_to.as_deref() == Some(upgrade_id.as_str()) { + upgraded = Some(envelope); + } else if envelope.reply_to.as_deref() == Some(crash_id.as_str()) { + replayed = Some(envelope); + } else { + panic!("unexpected message during parked delivery: {envelope:?}"); + } + } + }) + .await + .expect("upgrade did not redeliver the parked message"); + + assert_eq!(upgraded.unwrap().body["type"], "upgraded"); + assert_eq!( + replayed.unwrap().body, + json!({"type": "replayed", "version": "2"}) + ); + host.stop().await; +} + +#[tokio::test] +async fn rolls_back_a_crash_loop_and_restores_its_data_snapshot() { + let world = create_world("crash-loop"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + assert_eq!( + spawn_actor(&host, &mut mailbox, "crashloop", &wasm, &source) + .await + .body["type"], + "spawned" + ); + assert_eq!( + ask( + &host, + &mut mailbox, + "crashloop", + json!({"type": "write", "path": "/data/state", "text": "stable"}) + ) + .await + .body["type"], + "ok" + ); + + let upgrade_id = host.broker.send( + "console", + Some("broker"), + None, + None, + json!({"type": "upgrade", "actor": "crashloop", "wasm": wasm.display().to_string(), "source": source.display().to_string()}), + ); + let mut upgraded: Option = None; + let mut rolled_back: Option = None; + tokio::time::timeout(Duration::from_secs(20), async { + while upgraded.is_none() || rolled_back.is_none() { + let envelope = mailbox.recv().await.expect("console mailbox closed"); + let kind = envelope.body["type"].as_str().map(str::to_string); + match kind.as_deref() { + Some("upgraded") if envelope.reply_to.as_deref() == Some(upgrade_id.as_str()) => { + upgraded = Some(envelope.body); + } + Some("rolled_back") => rolled_back = Some(envelope.body), + other => panic!("unexpected automatic rollback message: {other:?}, {envelope:?}"), + } + } + }) + .await + .expect("crash loop did not trigger automatic rollback"); + + assert_eq!(upgraded.unwrap()["version"], 2); + let rollback = rolled_back.unwrap(); + assert_eq!(rollback["actor"], "crashloop"); + assert_eq!(rollback["from"], 2); + assert_eq!(rollback["to"], 1); + assert_eq!(rollback["reason"], "crash loop after upgrade"); + + let version = ask(&host, &mut mailbox, "crashloop", json!({"type": "version"})).await; + assert_eq!(version.body["version"], "1"); + let restored = ask( + &host, + &mut mailbox, + "crashloop", + json!({"type": "read", "path": "/data/state"}), + ) + .await; + assert_eq!(restored.body["text"], "stable"); + + host.stop().await; +} + +#[tokio::test] +async fn explicit_rollback_restores_the_previous_version_and_snapshot() { + let world = create_world("explicit-rollback"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + assert_eq!( + spawn_actor(&host, &mut mailbox, "recover", &wasm, &source) + .await + .body["type"], + "spawned" + ); + ask( + &host, + &mut mailbox, + "recover", + json!({"type": "write", "path": "/data/state", "text": "version-one"}), + ) + .await; + + let upgraded = upgrade_actor(&host, &mut mailbox, "recover", &wasm, &source).await; + assert_eq!(upgraded.body["type"], "upgraded"); + ask( + &host, + &mut mailbox, + "recover", + json!({"type": "write", "path": "/data/state", "text": "version-two"}), + ) + .await; + + let rolled_back = request( + &host, + &mut mailbox, + json!({"type": "rollback", "actor": "recover", "version": 1}), + ) + .await; + assert_eq!(rolled_back.body["type"], "rolled_back"); + assert_eq!(rolled_back.body["from"], 2); + assert_eq!(rolled_back.body["to"], 1); + assert_eq!(rolled_back.body["restored_state"], true); + let version = ask(&host, &mut mailbox, "recover", json!({"type": "version"})).await; + assert_eq!(version.body["version"], "1"); + let restored = ask( + &host, + &mut mailbox, + "recover", + json!({"type": "read", "path": "/data/state"}), + ) + .await; + assert_eq!(restored.body["text"], "version-one"); + + host.stop().await; +} + +#[tokio::test] +async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() { + let world = create_world("cargo-actor"); + let (host, mut mailbox) = start(&world).await; + let valid_source = cargo_source("cargo-valid"); + let valid_id = host.broker.send( + "console", + Some("cargo"), + None, + None, + json!({"type": "build", "source": valid_source.display().to_string()}), + ); + let built = receive_reply(&mut mailbox, &valid_id, Duration::from_secs(120)).await; + assert_eq!(built.body["type"], "built"); + let wasm = built.body["wasm"].as_str().unwrap(); + assert!(wasm.ends_with(".wasm")); + assert!(host.world.resolve("console", wasm).unwrap().is_file()); + + let invalid_source = cargo_source("cargo-fail"); + let invalid_id = host.broker.send( + "console", + Some("cargo"), + None, + None, + json!({"type": "build", "source": invalid_source.display().to_string()}), + ); + let failed = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await; + assert_eq!(failed.body["type"], "build_failed"); + assert!( + failed.body["errors"] + .as_str() + .unwrap() + .contains("missing_function") + ); + + host.stop().await; +} + +#[tokio::test] +async fn rejects_invalid_actor_ids() { + let world = create_world("invalid-id"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let invalid = request( + &host, + &mut mailbox, + json!({"type": "spawn", "actor": "bad/actor", "wasm": wasm.display().to_string()}), + ) + .await; + assert_eq!(invalid.body["type"], "error"); + assert!( + invalid.body["reason"] + .as_str() + .unwrap() + .contains("invalid actor ID") + ); + host.stop().await; +} + +#[tokio::test] +async fn rejects_parent_paths() { + let world = create_world("parent-path"); + let (host, mut mailbox) = start(&world).await; + let parent_path = request( + &host, + &mut mailbox, + json!({"type": "spawn", "actor": "safe", "wasm": "/public/../escape.wasm"}), + ) + .await; + assert_eq!(parent_path.body["type"], "error"); + assert!( + parent_path.body["reason"] + .as_str() + .unwrap() + .contains("must not contain '..'") + ); + host.stop().await; +}