diff --git a/PROTOCOL.md b/PROTOCOL.md index 1fd6f83..411e280 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -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//` (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) | diff --git a/cli/src/broker.rs b/cli/src/broker.rs index c90f6e1..fd7bad1 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -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//`. + /// 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 { + 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"), } } diff --git a/cli/tests/host.rs b/cli/tests/host.rs index 81ee5e6..8b92bbe 100644 --- a/cli/tests/host.rs +++ b/cli/tests/host.rs @@ -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");