broker: remove op deletes an actor (on top of tap/untap)
ci / check (push) Successful in 8s

{"type":"remove","actor"} shuts the actor down normally, drops its
mailbox, supervisor and actor.json entry (so also its subscriptions and
tap), and deletes /actor/<id>/. Native actors cannot be removed.
Re-applied from card-1-broker-remove (3c91747) onto card-2-broker-tap.
This commit is contained in:
Sausage Agent
2026-10-16 11:48:50 +00:00
parent 49453fa9af
commit c1f182d9b6
3 changed files with 91 additions and 4 deletions
+33 -3
View File
@@ -1,7 +1,7 @@
//! The broker: message routing, mailboxes, subscriptions, taps, the version
//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
//! subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs` runs each
//! WASM actor.
//! remove, subscribe, unsubscribe, tap, untap, 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
@@ -377,6 +377,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.
@@ -481,6 +510,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";
@@ -506,7 +536,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, tap, untap, scan, exit"),
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, scan, exit"),
}
}
+56
View File
@@ -494,6 +494,62 @@ 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");
let reason = unknown.body["reason"].as_str().unwrap();
assert!(reason.contains("remove") && reason.contains("tap"), "{reason}");
host.stop().await;
}
#[tokio::test]
async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
let world = create_world("cargo-actor");