1272 lines
51 KiB
Rust
1272 lines
51 KiB
Rust
mod common;
|
|
|
|
use std::time::Duration;
|
|
|
|
use common::*;
|
|
use sausage::{Envelope, Host, World};
|
|
use serde_json::{Value, json};
|
|
use tokio::sync::mpsc;
|
|
|
|
async fn filesystem_world(label: &str) -> (World, Host, mpsc::Receiver<Envelope>) {
|
|
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)
|
|
}
|
|
|
|
async fn receive_shutdown_messages(
|
|
mailbox: &mut mpsc::Receiver<Envelope>,
|
|
id: &str,
|
|
expected: &str,
|
|
) {
|
|
let (mut result, mut reply, mut publish) = (false, false, false);
|
|
tokio::time::timeout(Duration::from_secs(10), async {
|
|
while !result || !reply || !publish {
|
|
let message = mailbox.recv().await.expect("console mailbox closed");
|
|
if message.reply_to.as_deref() == Some(id) {
|
|
assert_eq!(message.body["type"], expected);
|
|
result = true;
|
|
} else if message.reply_to.as_deref() == Some("broker/shutdown") {
|
|
assert_eq!(message.from, "shutdown-output");
|
|
assert_eq!(message.body["type"], "shutdown-reply");
|
|
reply = true;
|
|
} else {
|
|
assert_eq!(message.from, "shutdown-output");
|
|
assert_eq!(message.reply_to, None);
|
|
assert_eq!(message.body["type"], "shutdown-publish");
|
|
publish = true;
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
.expect("shutdown output or lifecycle reply did not arrive");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn global_config_is_readable_owner_written_and_atomically_replaced() {
|
|
let world = create_world("global-config");
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let wasm = toolbox_wasm().clone();
|
|
for actor in ["shared", "reader"] {
|
|
let source = passing_source(&world, actor);
|
|
spawn_type(&host, &mut mailbox, actor, &wasm, &source).await;
|
|
}
|
|
let initial = ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": "/config"}), "content").await;
|
|
assert_eq!(initial.body["text"], "{}\n");
|
|
for path in ["/config", "/global/config", "/global/config.tmp", "/actor/shared/data/../../../../global/config"] {
|
|
ask_type(&host, &mut mailbox, "reader", json!({"type": "write", "path": path, "text": "not allowed"}), "error").await;
|
|
}
|
|
ask_type(&host, &mut mailbox, "reader", json!({"type": "rename", "path": "/config", "to": "/global/stolen"}), "error").await;
|
|
// The root view stays read-only even for the owner; publication uses its
|
|
// writable directory alias, with a sibling temporary file and rename.
|
|
ask_type(&host, &mut mailbox, "shared", json!({"type": "write", "path": "/config", "text": "not allowed"}), "error").await;
|
|
let next = "{\"llm\":{\"thinking\":\"medium\"}}\n";
|
|
ask_type(&host, &mut mailbox, "shared", json!({"type": "write", "path": "/global/config.tmp", "text": next}), "ok").await;
|
|
let before = ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": "/config"}), "content").await;
|
|
assert_eq!(before.body["text"], "{}\n");
|
|
// A native/world reader holding the old inode sees the complete old file.
|
|
let mut old = std::fs::File::open(world.root.join("config")).unwrap();
|
|
ask_type(&host, &mut mailbox, "shared", json!({"type": "rename", "path": "/global/config.tmp", "to": "/global/config"}), "ok").await;
|
|
let after = ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": "/config"}), "content").await;
|
|
assert_eq!(after.body["text"], next);
|
|
assert_eq!(std::fs::read_to_string(world.root.join("config")).unwrap(), next);
|
|
assert_eq!(std::fs::read_to_string(world.resolve("reader", "/config").unwrap()).unwrap(), next);
|
|
let mut previous = String::new();
|
|
std::io::Read::read_to_string(&mut old, &mut previous).unwrap();
|
|
assert_eq!(previous, "{}\n");
|
|
assert!(!world.global_dir().join("config.tmp").exists());
|
|
// The mount must not give shared writable access to any other world data.
|
|
ask_type(&host, &mut mailbox, "shared", json!({"type": "write", "path": "/global/../actor/reader/data/escape", "text": "no"}), "error").await;
|
|
host.stop().await;
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let persisted = ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": "/config"}), "content").await;
|
|
assert_eq!(persisted.body["text"], next);
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn global_config_candidates_cannot_change_the_live_file() {
|
|
let world = create_world("candidate-global-config");
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let source = source_with_test(&world, "shared", json!({"steps": [
|
|
{"send": {"type": "write", "path": "/global/next", "text": "{\"test\":true}"}, "expect": {"type": "ok"}},
|
|
{"send": {"type": "rename", "path": "/global/next", "to": "/global/config"}, "expect": {"type": "ok"}},
|
|
{"send": {"type": "read", "path": "/config"}, "expect": {"type": "content", "text": "{\"test\":true}"}}
|
|
]}));
|
|
spawn_type(&host, &mut mailbox, "shared", toolbox_wasm(), &source).await;
|
|
assert_eq!(std::fs::read_to_string(world.config_path()).unwrap(), "{}\n");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn global_config_is_added_to_existing_worlds_without_copying_actor_settings() {
|
|
let world = create_world("old-world-config");
|
|
std::fs::remove_file(world.root.join("config")).unwrap();
|
|
std::fs::remove_dir_all(world.global_dir()).unwrap();
|
|
world.ensure_actor_dirs("shared").unwrap();
|
|
// Synthetic marker only; no actual credentials are read in this test.
|
|
std::fs::write(world.data_dir("shared").join("config.json"), "legacy-private-marker").unwrap();
|
|
let (host, _) = start(&world).await;
|
|
assert_eq!(std::fs::read_to_string(world.root.join("config")).unwrap(), "{}\n");
|
|
assert_eq!(std::fs::read_to_string(world.data_dir("shared").join("config.json")).unwrap(), "legacy-private-marker");
|
|
host.stop().await;
|
|
// Preserve an operator-provided global config when adding the mounts.
|
|
std::fs::remove_file(world.root.join("config")).unwrap();
|
|
std::fs::remove_dir_all(world.global_dir()).unwrap();
|
|
std::fs::write(world.root.join("config"), "{\"operator\":true}").unwrap();
|
|
let (host, _) = start(&world).await;
|
|
assert_eq!(std::fs::read_to_string(world.root.join("config")).unwrap(), "{\"operator\":true}");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[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 broker_does_not_answer_messages_with_reply_to() {
|
|
let world = create_world("broker-reply-to");
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let untapped = request(&host, &mut mailbox, json!({"type": "untap"})).await;
|
|
assert_eq!(untapped.body["type"], "untapped");
|
|
|
|
host.broker.send(
|
|
"console",
|
|
Some("broker"),
|
|
None,
|
|
Some("upstream-message"),
|
|
json!({"type": "untap"}),
|
|
);
|
|
assert!(
|
|
tokio::time::timeout(Duration::from_secs(1), mailbox.recv())
|
|
.await
|
|
.is_err(),
|
|
"broker replied to a message that already had reply_to"
|
|
);
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn exit_request_is_answered_and_signals_the_host() {
|
|
let world = create_world("broker-exit");
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let exiting = request(&host, &mut mailbox, json!({"type": "exit"})).await;
|
|
assert_eq!(exiting.body["type"], "exiting");
|
|
tokio::time::timeout(Duration::from_secs(1), host.broker.exit_requested())
|
|
.await
|
|
.expect("exit request did not reach the host");
|
|
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 actors_can_read_but_not_write_other_actor_data() {
|
|
let (_, host, mut mailbox) = filesystem_world("actor-data-sharing").await;
|
|
ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"writer",
|
|
json!({"type": "write", "path": "/data/x", "text": "shared"}),
|
|
"ok",
|
|
)
|
|
.await;
|
|
let shared = ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"reader",
|
|
json!({"type": "read", "path": "/actor/writer/data/x"}),
|
|
"content",
|
|
)
|
|
.await;
|
|
assert_eq!(shared.body["text"], "shared");
|
|
|
|
ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"reader",
|
|
json!({"type": "write", "path": "/actor/writer/data/x", "text": "changed"}),
|
|
"error",
|
|
)
|
|
.await;
|
|
let unchanged = ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"writer",
|
|
json!({"type": "read", "path": "/data/x"}),
|
|
"content",
|
|
)
|
|
.await;
|
|
assert_eq!(unchanged.body["text"], "shared");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn actors_serve_files_under_fs() {
|
|
// "writer" serves its fs/ from memory (toolbox fs_read / fs_write / fs_remove).
|
|
let (_, host, mut mailbox) = filesystem_world("actor-fs").await;
|
|
let file = "/actor/writer/fs/notes/a.txt";
|
|
let write = json!({"type": "write", "path": file, "text": "hello"});
|
|
ask_type(&host, &mut mailbox, "reader", write, "ok").await;
|
|
// The write reached the owner as one fs_write message.
|
|
let direct = json!({"type": "fs_read", "path": "/notes/a.txt"});
|
|
let served = ask_type(&host, &mut mailbox, "writer", direct, "fs_file").await;
|
|
assert_eq!(served.body["text"], "hello");
|
|
|
|
let read = json!({"type": "read", "path": file});
|
|
assert_eq!(ask_type(&host, &mut mailbox, "reader", read, "content").await.body["text"], "hello");
|
|
let list = json!({"type": "list", "path": "/actor/writer/fs/notes"});
|
|
assert_eq!(ask_type(&host, &mut mailbox, "reader", list, "entries").await.body["names"], json!(["a.txt"]));
|
|
let root = json!({"type": "list", "path": "/actor/writer/fs"});
|
|
assert_eq!(ask_type(&host, &mut mailbox, "reader", root, "entries").await.body["names"], json!(["notes"]));
|
|
|
|
// fs_error and undeliverable become errors of the file operation.
|
|
for path in ["/actor/writer/fs/missing.txt", "/actor/nobody/fs/x"] {
|
|
ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": path}), "error").await;
|
|
}
|
|
// An actor that does not answer (here: it is busy reading itself) fails after the timeout.
|
|
let id = host.broker.send("console", Some("reader"), None, None, json!({"type": "read", "path": "/actor/reader/fs/x"}));
|
|
let timed_out = receive_reply(&mut mailbox, &id, Duration::from_secs(20)).await;
|
|
assert_eq!(timed_out.body["type"], "error");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn actor_fs_is_discoverable_only_when_its_source_docs_advertise_it() {
|
|
let world = create_world("actor-fs-discovery");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
|
|
let advertised = passing_source(&world, "advertised-fs");
|
|
std::fs::create_dir_all(advertised.join("src")).unwrap();
|
|
std::fs::write(
|
|
advertised.join("src/main.rs"),
|
|
"//! test actor\n//! Actor-served files: implements fs_read and fs_dir.\n",
|
|
)
|
|
.unwrap();
|
|
spawn_type(&host, &mut mailbox, "served", &wasm, &advertised).await;
|
|
|
|
let undocumented = passing_source(&world, "undocumented-fs");
|
|
std::fs::create_dir_all(undocumented.join("src")).unwrap();
|
|
std::fs::write(undocumented.join("src/main.rs"), "//! test actor without a virtual filesystem\n").unwrap();
|
|
spawn_type(&host, &mut mailbox, "plain", &wasm, &undocumented).await;
|
|
|
|
let reader_source = passing_source(&world, "fs-reader");
|
|
spawn_type(&host, &mut mailbox, "reader", &wasm, &reader_source).await;
|
|
|
|
let served_entries = ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"reader",
|
|
json!({"type": "list", "path": "/actor/served"}),
|
|
"entries",
|
|
)
|
|
.await;
|
|
assert!(served_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
|
|
|
|
let plain_entries = ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"reader",
|
|
json!({"type": "list", "path": "/actor/plain"}),
|
|
"entries",
|
|
)
|
|
.await;
|
|
assert!(!plain_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
|
|
|
|
let path = "/actor/served/fs/readme.txt";
|
|
ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"reader",
|
|
json!({"type": "write", "path": path, "text": "virtual document"}),
|
|
"ok",
|
|
)
|
|
.await;
|
|
let document = ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"reader",
|
|
json!({"type": "read", "path": path}),
|
|
"content",
|
|
)
|
|
.await;
|
|
assert_eq!(document.body["text"], "virtual document");
|
|
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn files_are_plain_posix_files() {
|
|
let (_, host, mut mailbox) = filesystem_world("posix").await;
|
|
// An open file is visible by path, to the writer and to every reader.
|
|
let open = json!({"type": "open", "path": "/public/new.txt", "text": "new"});
|
|
ask_type(&host, &mut mailbox, "writer", open, "opened").await;
|
|
let own = json!({"type": "read", "path": "/public/new.txt"});
|
|
assert_eq!(ask_type(&host, &mut mailbox, "writer", own, "content").await.body["text"], "new");
|
|
let other = json!({"type": "read", "path": "/actor/writer/public/new.txt"});
|
|
assert_eq!(ask_type(&host, &mut mailbox, "reader", other, "content").await.body["text"], "new");
|
|
// It can be renamed while open.
|
|
let rename = json!({"type": "rename", "path": "/public/new.txt", "to": "/public/moved.txt"});
|
|
ask_type(&host, &mut mailbox, "writer", rename, "ok").await;
|
|
ask_type(&host, &mut mailbox, "writer", json!({"type": "close", "path": "/public/new.txt"}), "closed").await;
|
|
let moved = json!({"type": "read", "path": "/actor/writer/public/moved.txt"});
|
|
assert_eq!(ask_type(&host, &mut mailbox, "reader", moved, "content").await.body["text"], "new");
|
|
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");
|
|
// A version that fails its tests never enters versions/.
|
|
assert!(!world.actor_dir("queue").join("versions/2").exists());
|
|
|
|
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<String> = 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"], "2");
|
|
}
|
|
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn rolls_back_a_crash_loop_to_the_previous_version_without_restoring_data() {
|
|
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<Value> = None;
|
|
let mut rolled_back: Option<Value> = None;
|
|
tokio::time::timeout(Duration::from_secs(35), 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 current = ask(
|
|
&host,
|
|
&mut mailbox,
|
|
"crashloop",
|
|
json!({"type": "read", "path": "/data/state"}),
|
|
)
|
|
.await;
|
|
assert_eq!(current.body["text"], "bad-v2");
|
|
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn upgrade_and_rollback_keep_a_stopped_actor_stopped() {
|
|
let world = create_world("upgrade-stopped");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let source = passing_source(&world, "source");
|
|
spawn_type(&host, &mut mailbox, "sleeper", &wasm, &source).await;
|
|
let stopped = request(&host, &mut mailbox, json!({"type": "stop", "actor": "sleeper"})).await;
|
|
assert_eq!(stopped.body["type"], "stopped");
|
|
|
|
let upgraded = upgrade_actor(&host, &mut mailbox, "sleeper", &wasm, &source).await;
|
|
assert_eq!(upgraded.body["type"], "upgraded", "{upgraded:?}");
|
|
assert_eq!(upgraded.body["version"], 2);
|
|
assert_eq!(upgraded.body["running"], false);
|
|
let meta = world.read_meta("sleeper");
|
|
assert_eq!((meta.version, meta.running), (Some(2), false));
|
|
|
|
// Mail waits in the mailbox: the stopped actor reads nothing.
|
|
let waiting = host.broker.send("console", Some("sleeper"), None, None, json!({"type": "version"}));
|
|
assert!(
|
|
tokio::time::timeout(Duration::from_millis(500), receive_reply(&mut mailbox, &waiting, Duration::from_secs(10)))
|
|
.await
|
|
.is_err(),
|
|
"a stopped actor must not answer"
|
|
);
|
|
|
|
let rolled_back = request(&host, &mut mailbox, json!({"type": "rollback", "actor": "sleeper", "version": 1})).await;
|
|
assert_eq!(rolled_back.body["type"], "rolled_back", "{rolled_back:?}");
|
|
assert_eq!(rolled_back.body["running"], false);
|
|
assert!(!world.read_meta("sleeper").running);
|
|
|
|
// `start` runs the installed version and delivers the waiting message.
|
|
let started = request(&host, &mut mailbox, json!({"type": "start", "actor": "sleeper"})).await;
|
|
assert_eq!(started.body["type"], "started");
|
|
let answer = receive_reply(&mut mailbox, &waiting, Duration::from_secs(10)).await;
|
|
assert_eq!(answer.body["version"], "1");
|
|
|
|
// A running actor still restarts on the new version.
|
|
let running = upgrade_actor(&host, &mut mailbox, "sleeper", &wasm, &source).await;
|
|
assert_eq!(running.body["running"], true);
|
|
assert_eq!(ask(&host, &mut mailbox, "sleeper", json!({"type": "version"})).await.body["version"], "3");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn explicit_rollback_keeps_current_data() {
|
|
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);
|
|
let version = ask(&host, &mut mailbox, "recover", json!({"type": "version"})).await;
|
|
assert_eq!(version.body["version"], "1");
|
|
let current = ask(
|
|
&host,
|
|
&mut mailbox,
|
|
"recover",
|
|
json!({"type": "read", "path": "/data/state"}),
|
|
)
|
|
.await;
|
|
assert_eq!(current.body["text"], "version-two");
|
|
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn remove_shuts_an_actor_down_and_deletes_its_folder() {
|
|
let world = create_world("remove");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let source = passing_source(&world, "source");
|
|
spawn_type(&host, &mut mailbox, "doomed", &wasm, &source).await;
|
|
let write = json!({"type": "write", "path": "/data/state", "text": "x"});
|
|
ask_type(&host, &mut mailbox, "doomed", write, "ok").await;
|
|
ask_type(&host, &mut mailbox, "doomed", json!({"type": "arm-shutdown"}), "armed").await;
|
|
|
|
// Normal shutdown: the actor gets "shutdown" and its last message is routed.
|
|
let id = host.broker.send("console", Some("broker"), None, None, json!({"type": "remove", "actor": "doomed"}));
|
|
let (mut removed, mut last_words) = (false, false);
|
|
tokio::time::timeout(Duration::from_secs(10), async {
|
|
while !removed || !last_words {
|
|
let message = mailbox.recv().await.expect("console mailbox closed");
|
|
if message.reply_to.as_deref() == Some(id.as_str()) {
|
|
assert_eq!(message.body, json!({"type": "removed", "actor": "doomed"}));
|
|
removed = true;
|
|
} else if message.reply_to.as_deref() == Some("broker/shutdown") {
|
|
assert_eq!(message.body["type"], "shutdown-reply");
|
|
last_words = true;
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
.expect("remove did not answer or did not shut the actor down");
|
|
assert!(!world.actor_dir("doomed").exists());
|
|
|
|
let gone = ask(&host, &mut mailbox, "doomed", json!({"type": "echo", "value": "hi"})).await;
|
|
assert_eq!(gone.body["type"], "undeliverable");
|
|
let again = request(&host, &mut mailbox, json!({"type": "remove", "actor": "doomed"})).await;
|
|
assert_eq!(again.body["type"], "error");
|
|
let broker = request(&host, &mut mailbox, json!({"type": "remove", "actor": "broker"})).await;
|
|
assert_eq!(broker.body["type"], "error");
|
|
assert!(world.actor_dir("broker").exists());
|
|
|
|
// The ID is free again: a new spawn starts at version 1 with empty data.
|
|
assert_eq!(spawn_type(&host, &mut mailbox, "doomed", &wasm, &source).await.body["version"], 1);
|
|
let read = json!({"type": "read", "path": "/data/state"});
|
|
ask_type(&host, &mut mailbox, "doomed", read, "error").await;
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn unknown_broker_request_lists_the_known_ones() {
|
|
let world = create_world("unknown-request");
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let unknown = request(&host, &mut mailbox, json!({"type": "frobnicate"})).await;
|
|
assert_eq!(unknown.body["type"], "error");
|
|
let reason = unknown.body["reason"].as_str().unwrap();
|
|
assert!(reason.contains("remove") && reason.contains("tap") && reason.contains("watch"), "{reason}");
|
|
// `scan` is retired: the host watches the world folder instead.
|
|
let scan = request(&host, &mut mailbox, json!({"type": "scan"})).await;
|
|
assert_eq!(scan.body["type"], "error");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cargo_actor_runs_unit_tests_by_default_and_can_skip_them() {
|
|
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(180)).await;
|
|
assert_eq!(built.body["type"], "built", "{built:?}");
|
|
assert_eq!(built.body["tests"]["passed"], 1);
|
|
assert_eq!(built.body["tests"]["failed"], 0);
|
|
assert!(built.body["tests"]["output"].as_str().unwrap().contains("unit_tests_run_on_the_host_target"));
|
|
let wasm = built.body["wasm"].as_str().unwrap();
|
|
assert!(wasm.ends_with(".wasm"));
|
|
assert!(host.world.resolve("console", wasm).unwrap().is_file());
|
|
|
|
let failing_test_source = cargo_source("cargo-test-fail");
|
|
let failed_test_id = host.broker.send(
|
|
"console",
|
|
Some("cargo"),
|
|
None,
|
|
None,
|
|
json!({"type": "build", "source": failing_test_source.display().to_string()}),
|
|
);
|
|
let failed_test = receive_reply(&mut mailbox, &failed_test_id, Duration::from_secs(120)).await;
|
|
assert_eq!(failed_test.body["type"], "build_failed", "{failed_test:?}");
|
|
assert_eq!(failed_test.body["tests"]["passed"], 0);
|
|
assert_eq!(failed_test.body["tests"]["failed"], 1);
|
|
assert!(failed_test.body["tests"]["output"].as_str().unwrap().contains("deliberate_unit_test_failure"));
|
|
|
|
let skipped_id = host.broker.send(
|
|
"console",
|
|
Some("cargo"),
|
|
None,
|
|
None,
|
|
json!({"type": "build", "source": failing_test_source.display().to_string(), "skip_tests": true}),
|
|
);
|
|
let skipped = receive_reply(&mut mailbox, &skipped_id, Duration::from_secs(120)).await;
|
|
assert_eq!(skipped.body["type"], "built", "{skipped:?}");
|
|
assert_eq!(skipped.body["tests"]["passed"], 0);
|
|
assert_eq!(skipped.body["tests"]["failed"], 0);
|
|
assert_eq!(skipped.body["tests"]["output"], "skipped (skip_tests=true)");
|
|
|
|
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 invalid = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await;
|
|
assert_eq!(invalid.body["type"], "build_failed", "{invalid:?}");
|
|
assert_eq!(invalid.body["tests"]["failed"], 1);
|
|
assert!(invalid.body["tests"]["output"].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;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn concurrent_spawns_of_one_id_leave_one_live_instance() {
|
|
let world = create_world("concurrent-spawn");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let source = passing_source(&world, "source");
|
|
let ids = [
|
|
spawn_request(&host, "spawn-race", &wasm, &source),
|
|
spawn_request(&host, "spawn-race", &wasm, &source),
|
|
];
|
|
let replies = receive_replies(&mut mailbox, &ids, Duration::from_secs(20)).await;
|
|
let spawned = replies
|
|
.iter()
|
|
.filter(|r| r.body["type"] == "spawned")
|
|
.count();
|
|
let failed = replies.iter().filter(|r| r.body["type"] == "error").count();
|
|
assert_eq!((spawned, failed), (1, 1));
|
|
|
|
let echo = ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"spawn-race",
|
|
json!({"type": "echo", "value": "live"}),
|
|
"echo",
|
|
)
|
|
.await;
|
|
assert_eq!(echo.body["value"], "live");
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
let list = ask_type(
|
|
&host,
|
|
&mut mailbox,
|
|
"spawn-race",
|
|
json!({"type": "list", "path": "/data"}),
|
|
"entries",
|
|
)
|
|
.await;
|
|
let starts = list.body["names"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.filter(|n| n.as_str().unwrap().starts_with("start-"))
|
|
.count();
|
|
assert_eq!(starts, 1);
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn stop_finishes_when_actor_stdin_is_full() {
|
|
let world = create_world("full-stdin-stop");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let source = passing_source(&world, "source");
|
|
spawn_type(&host, &mut mailbox, "full-stdin", &wasm, &source).await;
|
|
let ignore = json!({"type": "ignore-stdin"});
|
|
ask_type(&host, &mut mailbox, "full-stdin", ignore, "ignoring").await;
|
|
let large = json!({"type": "echo", "value": "x".repeat(256 * 1024)});
|
|
host.broker
|
|
.send("console", Some("full-stdin"), None, None, large);
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
let stop = json!({"type": "stop", "actor": "full-stdin"});
|
|
let id = host
|
|
.broker
|
|
.send("console", Some("broker"), None, None, stop);
|
|
let stopped = tokio::time::timeout(
|
|
Duration::from_secs(6),
|
|
receive_reply(&mut mailbox, &id, Duration::from_secs(10)),
|
|
)
|
|
.await
|
|
.expect("stop exceeded the five-second shutdown deadline");
|
|
assert_eq!(stopped.body["type"], "stopped");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn shutdown_reply_and_publish_survive_stop_and_upgrade() {
|
|
let world = create_world("shutdown-output");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let source = passing_source(&world, "source");
|
|
spawn_type(&host, &mut mailbox, "shutdown-output", &wasm, &source).await;
|
|
let subscribe = json!({"type": "subscribe", "actor": "shutdown-output"});
|
|
ask_type(&host, &mut mailbox, "broker", subscribe, "subscribed").await;
|
|
|
|
for operation in ["stop", "upgrade"] {
|
|
let arm = json!({"type": "arm-shutdown"});
|
|
ask_type(&host, &mut mailbox, "shutdown-output", arm, "armed").await;
|
|
let body = match operation {
|
|
"stop" => json!({"type": "stop", "actor": "shutdown-output"}),
|
|
_ => {
|
|
json!({"type": "upgrade", "actor": "shutdown-output", "wasm": wasm.display().to_string(), "source": source.display().to_string()})
|
|
}
|
|
};
|
|
let id = host
|
|
.broker
|
|
.send("console", Some("broker"), None, None, body);
|
|
receive_shutdown_messages(
|
|
&mut mailbox,
|
|
&id,
|
|
if operation == "stop" {
|
|
"stopped"
|
|
} else {
|
|
"upgraded"
|
|
},
|
|
)
|
|
.await;
|
|
if operation == "stop" {
|
|
let start = json!({"type": "start", "actor": "shutdown-output"});
|
|
ask_type(&host, &mut mailbox, "broker", start, "started").await;
|
|
}
|
|
}
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn candidates_receive_version_and_preexisting_spawn_state() {
|
|
let world = create_world("candidate-context");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let version_step =
|
|
json!({"send": {"type": "version"}, "expect": {"type": "version", "version": "1"}});
|
|
let version_source = source_with_test(&world, "version", json!({"steps": [version_step]}));
|
|
let state_step = json!({"send": {"type": "read", "path": "/data/state.json"}, "expect": {"type": "content", "text": "seeded"}});
|
|
let state_source = source_with_test(&world, "state", json!({"steps": [state_step]}));
|
|
let state_dir = world.data_dir("stateful");
|
|
std::fs::create_dir_all(&state_dir).unwrap();
|
|
std::fs::write(state_dir.join("state.json"), "seeded").unwrap();
|
|
let ids = [
|
|
spawn_request(&host, "versioned", &wasm, &version_source),
|
|
spawn_request(&host, "stateful", &wasm, &state_source),
|
|
];
|
|
let replies = receive_replies(&mut mailbox, &ids, Duration::from_secs(20)).await;
|
|
let successful = replies
|
|
.iter()
|
|
.filter(|reply| reply.body["type"] == "spawned")
|
|
.count();
|
|
assert_eq!(successful, 2);
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn running_actors_have_sockets_and_candidates_do_not() {
|
|
// A TCP server on the host that answers each line with "echo: <line>".
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
|
let address = listener.local_addr().unwrap().to_string();
|
|
tokio::spawn(async move {
|
|
while let Ok((stream, _)) = listener.accept().await {
|
|
tokio::spawn(async move {
|
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
|
|
let (read, mut write) = stream.into_split();
|
|
let mut lines = tokio::io::BufReader::new(read).lines();
|
|
while let Ok(Some(line)) = lines.next_line().await {
|
|
let _ = write.write_all(format!("echo: {line}\n").as_bytes()).await;
|
|
}
|
|
});
|
|
}
|
|
});
|
|
|
|
let world = create_world("sockets");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let tcp = json!({"type": "tcp", "address": address, "text": "hi"});
|
|
// The candidate test expects the connection to fail.
|
|
let offline = source_with_test(&world, "offline", json!({"steps": [{"send": tcp, "expect": {"type": "error"}}]}));
|
|
assert_eq!(spawn_actor(&host, &mut mailbox, "net", &wasm, &offline).await.body["type"], "spawned");
|
|
|
|
let answer = ask(&host, &mut mailbox, "net", tcp).await;
|
|
assert_eq!(answer.body, json!({"type": "tcp", "line": "echo: hi"}));
|
|
host.stop().await;
|
|
}
|
|
|
|
/// Every message that arrives until the mailbox is quiet for `quiet`.
|
|
async fn drain(mailbox: &mut mpsc::Receiver<Envelope>, quiet: Duration) -> Vec<Envelope> {
|
|
let mut messages = Vec::new();
|
|
while let Ok(Some(message)) = tokio::time::timeout(quiet, mailbox.recv()).await {
|
|
messages.push(message);
|
|
}
|
|
messages
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn tap_copies_every_routed_message_once() {
|
|
let world = create_world("tap");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let source = passing_source(&world, "source");
|
|
spawn_type(&host, &mut mailbox, "tapped", &wasm, &source).await;
|
|
let subscribe = json!({"type": "subscribe", "actor": "tapped"});
|
|
assert_eq!(request(&host, &mut mailbox, subscribe).await.body["type"], "subscribed");
|
|
assert_eq!(request(&host, &mut mailbox, json!({"type": "tap"})).await.body["type"], "tapping");
|
|
|
|
let echo = json!({"type": "echo", "value": "hi"});
|
|
let echo_id = host.broker.send("console", Some("tapped"), None, None, echo.clone());
|
|
let publish = json!({"type": "publish", "value": "news"});
|
|
host.broker.send("console", Some("tapped"), None, None, publish);
|
|
let messages = drain(&mut mailbox, Duration::from_secs(2)).await;
|
|
let copies: Vec<&Value> = messages
|
|
.iter()
|
|
.filter(|m| m.from == "broker" && m.body["type"] == "tapped")
|
|
.map(|m| &m.body)
|
|
.collect();
|
|
// 4 copies (2 requests, 1 reply, 1 publish) + the reply + the publish itself.
|
|
assert_eq!((copies.len(), messages.len()), (4, 6), "{messages:?}");
|
|
for copy in &copies {
|
|
assert!(copy["ts"].as_u64().unwrap() > 0, "{copy}");
|
|
}
|
|
let count = |want: Value| copies.iter().filter(|c| c["message"] == want).count();
|
|
let request_copy = json!({"id": echo_id, "from": "console", "to": "tapped", "reply_to": null, "body": echo});
|
|
assert_eq!(count(request_copy), 1, "{copies:?}");
|
|
let reply_copies = copies
|
|
.iter()
|
|
.filter(|c| c["message"]["reply_to"] == echo_id.as_str() && c["message"]["from"] == "tapped" && c["message"]["to"] == "console")
|
|
.count();
|
|
assert_eq!(reply_copies, 1, "{copies:?}");
|
|
let publish_copy = copies.iter().find(|c| c["message"]["body"]["type"] == "published").expect("publish not tapped");
|
|
assert_eq!(publish_copy["message"]["from"], "tapped");
|
|
assert_eq!(publish_copy["message"]["to"], Value::Null);
|
|
|
|
assert_eq!(request(&host, &mut mailbox, json!({"type": "untap"})).await.body["type"], "untapped");
|
|
let echo_id = host.broker.send("console", Some("tapped"), None, None, json!({"type": "echo", "value": "quiet"}));
|
|
let messages = drain(&mut mailbox, Duration::from_secs(1)).await;
|
|
assert_eq!(messages.len(), 1, "{messages:?}");
|
|
assert_eq!(messages[0].reply_to.as_deref(), Some(echo_id.as_str()));
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn tap_survives_a_restart() {
|
|
let world = create_world("tap-restart");
|
|
let (host, mut mailbox) = start(&world).await;
|
|
assert_eq!(request(&host, &mut mailbox, json!({"type": "tap"})).await.body["type"], "tapping");
|
|
host.stop().await;
|
|
assert!(world.read_meta("console").tapped);
|
|
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let id = host.broker.send("console", Some("nobody"), None, None, json!({"type": "hello"}));
|
|
let copy = tokio::time::timeout(Duration::from_secs(10), async {
|
|
loop {
|
|
let m = mailbox.recv().await.expect("console mailbox closed");
|
|
if m.body["type"] == "tapped" && m.body["message"]["id"] == id.as_str() {
|
|
return m;
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
.expect("no tapped copy after the restart");
|
|
assert_eq!(copy.body["message"]["to"], "nobody");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_full_tapper_mailbox_drops_copies_without_blocking() {
|
|
let world = create_world("tap-full");
|
|
let (host, mut mailbox) = start(&world).await;
|
|
// "ghost" taps and never reads its mailbox.
|
|
let mut ghost = host.mailbox("ghost");
|
|
let mut sink = host.mailbox("sink");
|
|
host.broker.send("ghost", Some("broker"), None, None, json!({"type": "tap"}));
|
|
tokio::time::timeout(Duration::from_secs(10), async {
|
|
while !world.read_meta("ghost").tapped {
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
})
|
|
.await
|
|
.expect("tap was not saved");
|
|
|
|
let total = sausage::broker::MAILBOX_SIZE + 500;
|
|
for n in 0..total {
|
|
host.broker.send("console", Some("sink"), None, None, json!({"type": "n", "n": n}));
|
|
assert_eq!(sink.try_recv().unwrap().body["n"], n);
|
|
}
|
|
assert!(mailbox.try_recv().is_err(), "a dropped copy reported something to the sender");
|
|
let mut held = 0;
|
|
while ghost.try_recv().is_ok() {
|
|
held += 1;
|
|
}
|
|
assert_eq!(held, sausage::broker::MAILBOX_SIZE);
|
|
host.broker.send("console", Some("sink"), None, None, json!({"type": "after"}));
|
|
assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped");
|
|
host.stop().await;
|
|
}
|
|
|
|
/// Ask `actor` again and again (it may not exist yet) until `want` holds for
|
|
/// the answer.
|
|
async fn ask_until(
|
|
host: &Host,
|
|
mailbox: &mut mpsc::Receiver<Envelope>,
|
|
actor: &str,
|
|
body: Value,
|
|
want: impl Fn(&Value) -> bool,
|
|
) -> Envelope {
|
|
tokio::time::timeout(Duration::from_secs(30), async {
|
|
loop {
|
|
let reply = ask(host, mailbox, actor, body.clone()).await;
|
|
if want(&reply.body) {
|
|
return reply;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap_or_else(|_| panic!("{actor} never gave the wanted answer"))
|
|
}
|
|
|
|
/// The `fs_changed` bodies among `messages` (all from `fs`).
|
|
fn fs_events(messages: Vec<Envelope>) -> Vec<Value> {
|
|
messages
|
|
.into_iter()
|
|
.filter(|m| m.body["type"] == "fs_changed")
|
|
.map(|m| {
|
|
assert_eq!(m.from, "fs");
|
|
m.body
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_new_actor_wasm_on_disk_becomes_an_actor() {
|
|
let world = create_world("watch-spawn");
|
|
let (host, mut mailbox) = start(&world).await;
|
|
// Any writer counts: here the test itself, like a shell on the host.
|
|
let tests = world.source_dir("dropped").join("tests");
|
|
std::fs::create_dir_all(&tests).unwrap();
|
|
let test = json!({"steps": [{"send": {"type": "echo", "value": "candidate"}, "expect": {"type": "echo", "value": "candidate"}}]});
|
|
std::fs::write(tests.join("candidate.json"), test.to_string()).unwrap();
|
|
std::fs::copy(toolbox_wasm(), world.wasm_path("dropped")).unwrap();
|
|
|
|
let echo = json!({"type": "echo", "value": "hi"});
|
|
let reply = ask_until(&host, &mut mailbox, "dropped", echo, |b| b["type"] == "echo").await;
|
|
assert_eq!(reply.from, "dropped");
|
|
assert_eq!(world.read_meta("dropped").version, Some(1));
|
|
let meta = std::fs::read_to_string(world.version_dir("dropped", 1).join("meta.json")).unwrap();
|
|
assert_eq!(serde_json::from_str::<Value>(&meta).unwrap()["requested_by"], "scan");
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_changed_actor_wasm_on_disk_is_an_upgrade() {
|
|
let world = create_world("watch-upgrade");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let source = passing_source(&world, "source");
|
|
spawn_type(&host, &mut mailbox, "changing", &wasm, &source).await;
|
|
|
|
// The same program plus a custom section: a different file.
|
|
let mut changed = std::fs::read(&wasm).unwrap();
|
|
changed.extend_from_slice(&[0, 4, 1, b'x', b'v', b'2']);
|
|
std::fs::write(world.wasm_path("changing"), changed).unwrap();
|
|
|
|
let version = json!({"type": "version"});
|
|
ask_until(&host, &mut mailbox, "changing", version, |b| b["version"] == "2").await;
|
|
assert_eq!(world.read_meta("changing").version, Some(2));
|
|
assert_eq!(world.versions("changing"), vec![1, 2]);
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn watchers_get_one_event_per_change_from_any_writer() {
|
|
let (world, host, mut mailbox) = filesystem_world("watch-events").await;
|
|
let watch = json!({"type": "watch", "path": "/actor/writer/public/"});
|
|
let watching = request(&host, &mut mailbox, watch).await;
|
|
assert_eq!(watching.body, json!({"type": "watching", "path": "/actor/writer/public"}));
|
|
assert!(world.read_meta("console").watches.contains("/actor/writer/public"));
|
|
drain(&mut mailbox, Duration::from_millis(300)).await;
|
|
|
|
let quiet = Duration::from_millis(600);
|
|
let event = |path: &str, op: &str| json!({"type": "fs_changed", "path": format!("/actor/writer/public/{path}"), "op": op});
|
|
|
|
// An actor creates a file: one create, not create + write.
|
|
let write = json!({"type": "write", "path": "/public/a.txt", "text": "a"});
|
|
ask_type(&host, &mut mailbox, "writer", write, "ok").await;
|
|
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("a.txt", "create")]);
|
|
|
|
// The host (a shell, git, ...) changes it.
|
|
std::fs::write(world.public_dir("writer").join("a.txt"), "b").unwrap();
|
|
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("a.txt", "write")]);
|
|
|
|
// A rename is one event with `to`.
|
|
let rename = json!({"type": "rename", "path": "/public/a.txt", "to": "/public/b.txt"});
|
|
ask_type(&host, &mut mailbox, "writer", rename, "ok").await;
|
|
let mut renamed = event("a.txt", "rename");
|
|
renamed["to"] = json!("/actor/writer/public/b.txt");
|
|
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![renamed]);
|
|
|
|
std::fs::remove_file(world.public_dir("writer").join("b.txt")).unwrap();
|
|
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("b.txt", "remove")]);
|
|
|
|
// Nothing outside the watched path.
|
|
let outside = json!({"type": "write", "path": "/data/x", "text": "x"});
|
|
ask_type(&host, &mut mailbox, "writer", outside, "ok").await;
|
|
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), Vec::<Value>::new());
|
|
|
|
let unwatch = json!({"type": "unwatch", "path": "/actor/writer/public"});
|
|
let unwatched = request(&host, &mut mailbox, unwatch).await;
|
|
assert_eq!(unwatched.body, json!({"type": "unwatched", "path": "/actor/writer/public"}));
|
|
assert!(world.read_meta("console").watches.is_empty());
|
|
let write = json!({"type": "write", "path": "/public/c.txt", "text": "c"});
|
|
ask_type(&host, &mut mailbox, "writer", write, "ok").await;
|
|
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), Vec::<Value>::new());
|
|
host.stop().await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_watcher_overflow_sends_one_rescan_and_runs_a_full_pass() {
|
|
let world = create_world("watch-overflow");
|
|
let wasm = toolbox_wasm().clone();
|
|
let (host, mut mailbox) = start(&world).await;
|
|
let mut bystander = host.mailbox("bystander");
|
|
let watch = json!({"type": "watch", "path": "/actor/elsewhere"});
|
|
assert_eq!(request(&host, &mut mailbox, watch).await.body["type"], "watching");
|
|
|
|
// A dropped actor that the full pass finds (before the watcher's quiet time).
|
|
let tests = world.source_dir("planted").join("tests");
|
|
std::fs::create_dir_all(&tests).unwrap();
|
|
let test = json!({"steps": [{"send": {"type": "echo", "value": "candidate"}, "expect": {"type": "echo", "value": "candidate"}}]});
|
|
std::fs::write(tests.join("candidate.json"), test.to_string()).unwrap();
|
|
std::fs::copy(&wasm, world.wasm_path("planted")).unwrap();
|
|
let report = host.broker.clone().overflowed().await;
|
|
assert!(report.iter().any(|r| r["type"] == "spawned" && r["actor"] == "planted"), "{report:?}");
|
|
|
|
let rescan = json!({"type": "fs_changed", "path": "/actor", "op": "rescan"});
|
|
assert_eq!(fs_events(drain(&mut mailbox, Duration::from_millis(600)).await), vec![rescan]);
|
|
assert!(bystander.try_recv().is_err(), "an actor without watches got the rescan");
|
|
host.stop().await;
|
|
}
|