Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3c91747ce6 |
+2
-1
@@ -109,7 +109,7 @@ Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`.
|
||||
|
||||
Requests to `broker`. The answer is a message whose `reply_to` is the ID of
|
||||
the request: the request's `type` in past tense (`spawned`, `upgraded`,
|
||||
`rolled_back`, `stopped`, `started`, `subscribed`, `unsubscribed`,
|
||||
`rolled_back`, `stopped`, `started`, `removed`, `subscribed`, `unsubscribed`,
|
||||
`scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
|
||||
|
||||
| `type` | Fields | Effect |
|
||||
@@ -118,6 +118,7 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
|
||||
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version |
|
||||
| `rollback` | `actor`, `version` | go back to an older version |
|
||||
| `stop`, `start` | `actor` | stop or start an actor |
|
||||
| `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<id>/` (all versions, `data/`, `public/`, logs); answer `{"type": "removed", "actor"}`. Messages to it are then undeliverable, and the ID can be spawned again. An unknown actor or a native one (`broker`, `console`, `cargo`, `fs`) is an `error` |
|
||||
| `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
|
||||
| `scan` | | boot new or changed `actor.wasm` files |
|
||||
| `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) |
|
||||
|
||||
+32
-2
@@ -1,6 +1,6 @@
|
||||
//! The broker: message routing, mailboxes, subscriptions, the version store,
|
||||
//! and the requests to `broker` (spawn, upgrade, rollback, stop, start,
|
||||
//! subscribe, unsubscribe, scan, exit). `supervisor.rs` runs each WASM actor.
|
||||
//! remove, subscribe, unsubscribe, scan, exit). `supervisor.rs` runs each WASM actor.
|
||||
//!
|
||||
//! The broker accepts only `.wasm` files. It does not know any source language.
|
||||
//! The world folder is the source of truth: `scan` boots every
|
||||
@@ -351,6 +351,35 @@ impl Broker {
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
/// Stop an actor (normal shutdown) and delete its folder `actor/<id>/`.
|
||||
/// Afterwards the ID is unknown: messages to it are undeliverable, and
|
||||
/// it can be spawned again from scratch.
|
||||
pub async fn remove_actor(&self, id: &str) -> Result<Value> {
|
||||
if NATIVE.contains(&id) {
|
||||
bail!("{id} is part of the host and cannot be removed");
|
||||
}
|
||||
if !world::valid_id(id) {
|
||||
bail!("invalid actor ID {id:?}");
|
||||
}
|
||||
let lock = self.op_lock(id);
|
||||
let _guard = lock.lock().await;
|
||||
let dir = self.world.actor_dir(id);
|
||||
if !self.is_supervised(id) && !dir.is_dir() {
|
||||
bail!("no actor with ID {id}");
|
||||
}
|
||||
if self.is_supervised(id) {
|
||||
self.control(id, |reply| Control::Shutdown { reply }).await?;
|
||||
}
|
||||
// Without its mailbox, messages to `id` are undeliverable.
|
||||
self.controls.lock().unwrap().remove(id);
|
||||
self.mailboxes.lock().unwrap().remove(id);
|
||||
self.metas.lock().unwrap().remove(id);
|
||||
if dir.exists() {
|
||||
std::fs::remove_dir_all(&dir).with_context(|| format!("delete {}", dir.display()))?;
|
||||
}
|
||||
Ok(json!({"type": "removed", "actor": id}))
|
||||
}
|
||||
|
||||
/// Boot the actors in `actor/`. Called at start and on `scan`.
|
||||
///
|
||||
/// - An `actor.wasm` equal to the installed version starts as is.
|
||||
@@ -455,6 +484,7 @@ impl Broker {
|
||||
}
|
||||
"stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await,
|
||||
"start" => self.control(&actor()?, |reply| Control::Start { reply }).await,
|
||||
"remove" => self.remove_actor(&actor()?).await,
|
||||
"subscribe" | "unsubscribe" => {
|
||||
let publisher = actor()?;
|
||||
let on = ty == "subscribe";
|
||||
@@ -474,7 +504,7 @@ impl Broker {
|
||||
self.exit.notify_one();
|
||||
Ok(json!({"type": "exiting"}))
|
||||
}
|
||||
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, scan, exit"),
|
||||
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, scan, exit"),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -494,6 +494,61 @@ async fn explicit_rollback_keeps_current_data() {
|
||||
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");
|
||||
assert!(unknown.body["reason"].as_str().unwrap().contains("remove"));
|
||||
host.stop().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
|
||||
let world = create_world("cargo-actor");
|
||||
|
||||
Reference in New Issue
Block a user