diff --git a/PROTOCOL.md b/PROTOCOL.md index c5d26bf..5324395 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -125,7 +125,7 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`, | `type` | Fields | Effect | | --- | --- | --- | | `spawn` | `actor`, `wasm`, `source`?, `built_by`? | test, then start a new actor | -| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version | +| `upgrade` | `actor`, `wasm`, `source`?, `built_by`?, `preserve_stopped`? (boolean, default `false`) | test, then swap the version; with `preserve_stopped: true`, a stopped actor is compiled and installed without execution and stays stopped | | `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` | @@ -182,10 +182,11 @@ change of `path` or a path under it (`/actor/a` covers `/actor/a/x`, not restarts and upgrades. `unwatch` with the same path ends it. **Versions.** A new version (from `spawn`, `upgrade`, or a changed -`actor.wasm` on disk) is tested first, while the old version keeps running. +`actor.wasm` on disk) is tested first, while the old version keeps running +(except the stopped-preserving upgrade described below). Only when its tests pass does the broker copy it to `versions//` (`actor.wasm`, `source/`, `meta.json`) and switch to it. So `versions/` holds -exactly the versions that were installed, and instances always run from +exactly the versions that were installed (including stopped installations), and instances always run from there. If the tests fail, nothing changes, and the sender gets an `error` with the reason. All versions share the actor's `data/` and `public/`: a new version must read the data of the old one. @@ -214,6 +215,21 @@ instance. A version without test files passes. top-level `actor.wasm` and `source/` a copy of the new version, and starts it. A rollback does the same with the older version. +**Stopped-preserving upgrade.** Send `upgrade` with `"preserve_stopped": true` +to install a stopped actor's new program without starting it, running `main` +(even as a candidate), or reading or sending any of its mail. The host compiles +the WASM and copies the new version, source and metadata as usual, increments +the version, and keeps `actor.json`'s `running: false`. Its mailbox, `data/` +and `public/` stay untouched. Because candidate tests execute `main`, they are +not run in this case: the `upgraded` reply contains +`"tests": {"skipped": "preserve_stopped"}`. Test the build separately before +installing it this way. An explicit `start` later runs the installed version +and delivers the queued mail. No startup/crash probation runs for this stopped +installation. Running actors still use the normal tested swap, even with this +option. Omitting the option (or `false`) retains the existing behavior, +including starting a stopped actor. Explicit stop/start requests serialize +with upgrades of the same actor. + **Crashes.** The broker restarts a crashed actor after 0.5 s, then doubles the delay up to 30 s. A new version that crashes 3 times within 30 seconds after an upgrade goes back to the previous version. The actor that asked for diff --git a/cli/src/broker.rs b/cli/src/broker.rs index b8a5095..1b98fe2 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -313,12 +313,16 @@ impl Broker { Ok(component) } - /// Test a new version where it lies. Only when the tests pass, copy it - /// to `versions//`: that folder holds only versions that run(ran). - /// Returns the component and the test report. - async fn try_version(&self, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<(Component, Value)> { + /// Compile a new version where it lies and, normally, test it. Copy it + /// to `versions//` only on success. A stopped-preserving install skips + /// execution, including candidate tests. Return the component and report. + async fn try_version(&self, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, run_tests: bool) -> Result<(Component, Value)> { let component = self.load(id, wasm)?; - let tests = candidate::test(self, id, meta.version, &component, source).await; + let tests = if run_tests { + candidate::test(self, id, meta.version, &component, source).await + } else { + Ok(json!({"skipped": "preserve_stopped"})) + }; if let Err(e) = &tests { self.log(id, &format!("version {} from {} failed its tests: {e:#}", meta.version, meta.wasm)); } @@ -332,11 +336,11 @@ impl Broker { } /// Make `version` the current `actor.wasm`, `source/`, and `actor.json`. - pub fn install(&self, id: &str, version: u64) -> Result<()> { + pub fn install(&self, id: &str, version: u64, running: bool) -> Result<()> { self.world.install_files(id, version)?; self.update_meta(id, |m| { m.version = Some(version); - m.running = true; + m.running = running; }) } @@ -359,26 +363,29 @@ impl Broker { } async fn create_version(self: &Arc, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result { - let (_, tests) = self.try_version(id, meta, wasm, source).await?; - self.install(id, meta.version)?; + let (_, tests) = self.try_version(id, meta, wasm, source, true).await?; + self.install(id, meta.version, true)?; self.start_supervisor(id, meta.version, true); Ok(json!({"type": "spawned", "actor": id, "version": meta.version, "tests": tests})) } - pub async fn upgrade_actor(self: &Arc, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> Result { + pub async fn upgrade_actor(self: &Arc, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta, preserve_stopped: bool) -> Result { let lock = self.op_lock(id); let _guard = lock.lock().await; if !self.is_supervised(id) { bail!("no WASM actor with ID {id}"); } let version = self.world.next_version(id); - self.upgrade_version(id, &VersionMeta { version, ..meta }, wasm, source).await + self.upgrade_version(id, &VersionMeta { version, ..meta }, wasm, source, preserve_stopped).await } - async fn upgrade_version(self: &Arc, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result { - let (component, tests) = self.try_version(id, meta, wasm, source).await?; + async fn upgrade_version(self: &Arc, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, preserve_stopped: bool) -> Result { + // Lifecycle requests hold the same op lock, so a stopped actor cannot + // start between this decision and the supervisor's installation. + let run_tests = !preserve_stopped || self.meta(id).is_some_and(|m| m.running); + let (component, tests) = self.try_version(id, meta, wasm, source, run_tests).await?; let (version, requester) = (meta.version, meta.requested_by.clone()); - let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, reply }).await?; + let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, preserve_stopped, reply }).await?; result["tests"] = tests; Ok(result) } @@ -517,7 +524,7 @@ impl Broker { let source = source.is_dir().then_some(source.as_path()); let result = match installed { None => self.create_version(id, &vmeta, &wasm, source).await, - Some(_) => self.upgrade_version(id, &vmeta, &wasm, source).await, + Some(_) => self.upgrade_version(id, &vmeta, &wasm, source, false).await, }; Some(result.unwrap_or_else(|e| json!({"type": "error", "actor": id, "reason": format!("{e:#}")}))) } @@ -564,7 +571,11 @@ impl Broker { } "upgrade" => { let (wasm, source, meta) = version_meta()?; - self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta).await + let preserve_stopped = match b.get("preserve_stopped") { + None => false, + Some(value) => value.as_bool().ok_or_else(|| anyhow!("preserve_stopped must be a boolean"))?, + }; + self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta, preserve_stopped).await } "rollback" => { let id = actor()?; @@ -573,8 +584,15 @@ impl Broker { let _guard = lock.lock().await; self.control(&id, |reply| Control::Rollback { version, reply }).await } - "stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await, - "start" => self.control(&actor()?, |reply| Control::Start { reply }).await, + "stop" | "start" => { + let id = actor()?; + let lock = self.op_lock(&id); + let _guard = lock.lock().await; + self.control(&id, |reply| match ty { + "stop" => Control::Stop { reply }, + _ => Control::Start { reply }, + }).await + }, "remove" => self.remove_actor(&actor()?).await, "subscribe" | "unsubscribe" => { let publisher = actor()?; diff --git a/cli/src/supervisor.rs b/cli/src/supervisor.rs index ed76e99..92ad333 100644 --- a/cli/src/supervisor.rs +++ b/cli/src/supervisor.rs @@ -23,7 +23,7 @@ pub type Reply = oneshot::Sender>; pub enum Control { /// Swap to `version` (already in `versions/`, compiled as `component`). - Upgrade { version: u64, component: Component, requester: String, reply: Reply }, + Upgrade { version: u64, component: Component, requester: String, preserve_stopped: bool, reply: Reply }, Rollback { version: u64, reply: Reply }, Stop { reply: Reply }, Start { reply: Reply }, @@ -221,24 +221,29 @@ impl Supervisor { /// Make `version` current. It is compiled first (unless `component` is /// given): if that fails, nothing changes. `data/` and `public/` stay as /// they are. - async fn install(&mut self, version: u64, component: Option) -> Result<()> { + async fn install(&mut self, version: u64, component: Option, running: bool) -> Result<()> { let component = match component { Some(c) => c, None => self.broker.load_version(&self.id, version)?, }; self.stop().await; - self.broker.install(&self.id, version)?; + self.broker.install(&self.id, version, running)?; self.version = version; self.component = Some(component); - self.running = true; + self.running = running; self.restarts = 0; self.restart_at = None; Ok(()) } - async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result { + async fn upgrade(&mut self, version: u64, component: Component, requester: String, preserve_stopped: bool) -> Result { let previous = self.version; - self.install(version, Some(component)).await?; + let running = self.running || !preserve_stopped; + self.install(version, Some(component), running).await?; + if !running { + self.probation = None; + return Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous})); + } self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester }); if let Err(e) = self.start().await { self.rollback(previous).await?; @@ -249,7 +254,7 @@ impl Supervisor { async fn rollback(&mut self, version: u64) -> Result { let from = self.version; - self.install(version, None).await?; + self.install(version, None, true).await?; self.probation = None; self.start().await?; Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version})) @@ -258,8 +263,8 @@ impl Supervisor { /// Handle a control request. Returns `true` when the supervisor ends. async fn handle(&mut self, ctl: Control) -> bool { match ctl { - Control::Upgrade { version, component, requester, reply } => { - let _ = reply.send(self.upgrade(version, component, requester).await); + Control::Upgrade { version, component, requester, preserve_stopped, reply } => { + let _ = reply.send(self.upgrade(version, component, requester, preserve_stopped).await); } Control::Rollback { version, reply } => { let _ = reply.send(self.rollback(version).await); diff --git a/cli/tests/actors/toolbox/src/main.rs b/cli/tests/actors/toolbox/src/main.rs index 6672db7..0808119 100644 --- a/cli/tests/actors/toolbox/src/main.rs +++ b/cli/tests/actors/toolbox/src/main.rs @@ -38,6 +38,10 @@ fn main() { panic!("version 2 crashes on startup"); } + if actor == "stopped-upgrade" { + std::fs::write("/data/main-version", &version).unwrap(); + } + let mut open = HashMap::::new(); // Files that this actor serves at /actor//fs/ (path -> text). let mut served = std::collections::BTreeMap::::new(); diff --git a/cli/tests/host.rs b/cli/tests/host.rs index 005b3c0..b0d0b32 100644 --- a/cli/tests/host.rs +++ b/cli/tests/host.rs @@ -436,6 +436,113 @@ async fn upgrades_test_candidates_keep_old_versions_and_queue_messages_in_order( host.stop().await; } +#[tokio::test] +async fn stopped_preserving_upgrade_installs_without_execution_and_keeps_all_mail() { + let world = create_world("stopped-upgrade"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "initial"); + spawn_type(&host, &mut mailbox, "stopped-upgrade", &wasm, &source).await; + ask_type(&host, &mut mailbox, "stopped-upgrade", json!({"type": "version"}), "version").await; + let stopped = request(&host, &mut mailbox, json!({"type": "stop", "actor": "stopped-upgrade"})).await; + assert_eq!(stopped.body["type"], "stopped"); + + // A distinct valid component: append an empty custom section named x. + let mut bytes = std::fs::read(&wasm).unwrap(); + bytes.extend_from_slice(&[0, 2, 1, b'x']); + let new_wasm = world.public_dir("console").join("new.wasm"); + std::fs::write(&new_wasm, &bytes).unwrap(); + // This would fail if candidate main/tests were executed. + let unrun = source_with_test(&world, "unrun", json!({"steps": [{ + "send": {"type": "echo"}, "expect": {"type": "must-not-run"} + }]})); + std::fs::write(unrun.join("new-source"), "version two").unwrap(); + std::fs::write(world.public_dir("stopped-upgrade").join("keep"), "public state").unwrap(); + + let mut ids = Vec::new(); + for n in 0..sausage::broker::MAILBOX_SIZE { + ids.push(host.broker.send("console", Some("stopped-upgrade"), None, None, json!({"type": "ordered", "n": n}))); + } + let upgrade_id = host.broker.send("console", Some("broker"), None, None, json!({ + "type": "upgrade", "actor": "stopped-upgrade", "wasm": new_wasm.display().to_string(), + "source": unrun.display().to_string(), "preserve_stopped": true + })); + // Do not silently discard unexpected actor mail while awaiting the reply. + let upgraded = receive_message(&mut mailbox, Duration::from_secs(10)).await; + assert_eq!(upgraded.reply_to.as_deref(), Some(upgrade_id.as_str())); + assert_eq!(upgraded.body["type"], "upgraded"); + assert_eq!(upgraded.body["version"], 2); + assert_eq!(upgraded.body["previous"], 1); + assert_eq!(upgraded.body["tests"], json!({"skipped": "preserve_stopped"})); + let meta = world.read_meta("stopped-upgrade"); + assert_eq!(meta.version, Some(2)); + assert!(!meta.running); + assert_eq!(std::fs::read(world.wasm_path("stopped-upgrade")).unwrap(), bytes); + assert_eq!(std::fs::read(world.version_dir("stopped-upgrade", 2).join("actor.wasm")).unwrap(), bytes); + assert_ne!(std::fs::read(world.version_dir("stopped-upgrade", 1).join("actor.wasm")).unwrap(), bytes); + assert_eq!(std::fs::read_to_string(world.source_dir("stopped-upgrade").join("new-source")).unwrap(), "version two"); + assert_eq!(std::fs::read_to_string(world.data_dir("stopped-upgrade").join("main-version")).unwrap(), "1"); + assert_eq!(std::fs::read_to_string(world.public_dir("stopped-upgrade").join("keep")).unwrap(), "public state"); + assert!(mailbox.try_recv().is_err()); + + // The full mailbox is still full: nothing was consumed or added by the upgrade. + let overflow_id = host.broker.send("console", Some("stopped-upgrade"), None, None, json!({"type": "echo"})); + let overflow = receive_message(&mut mailbox, Duration::from_secs(10)).await; + assert_eq!(overflow.reply_to.as_deref(), Some(overflow_id.as_str())); + assert_eq!(overflow.body["type"], "undeliverable"); + assert!(overflow.body["reason"].as_str().unwrap().contains("full")); + + // Only an explicit start executes v2. Every original envelope survives in order. + let start_id = host.broker.send("console", Some("broker"), None, None, json!({"type": "start", "actor": "stopped-upgrade"})); + let mut started = false; + let mut count = 0; + tokio::time::timeout(Duration::from_secs(10), async { + while !started || count < ids.len() { + let reply = mailbox.recv().await.unwrap(); + if reply.reply_to.as_deref() == Some(start_id.as_str()) { + assert_eq!(reply.body["type"], "started"); + started = true; + } else { + assert_eq!(reply.from, "stopped-upgrade"); + assert_eq!(reply.reply_to.as_deref(), Some(ids[count].as_str())); + assert_eq!(reply.body, json!({"type": "ordered", "n": count, "version": "2"})); + count += 1; + } + } + }).await.expect("queued mail was not delivered after start"); + assert_eq!(std::fs::read_to_string(world.data_dir("stopped-upgrade").join("main-version")).unwrap(), "2"); + host.stop().await; +} + +#[tokio::test] +async fn stopped_preserving_option_keeps_running_upgrades_tested_and_default_upgrades_start() { + let world = create_world("preserve-running"); + 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"}, "expect": {"type": "wrong"} + }]})); + spawn_type(&host, &mut mailbox, "running", &wasm, &passing).await; + for (option, source, expected) in [(json!("true"), &passing, "error"), (json!(true), &failing, "error"), (json!(true), &passing, "upgraded")] { + let reply = request(&host, &mut mailbox, json!({ + "type": "upgrade", "actor": "running", "wasm": wasm.display().to_string(), + "source": source.display().to_string(), "preserve_stopped": option + })).await; + assert_eq!(reply.body["type"], expected); + } + assert!(world.read_meta("running").running); + let version = ask_type(&host, &mut mailbox, "running", json!({"type": "version"}), "version").await; + assert_eq!(version.body["version"], "2"); + request(&host, &mut mailbox, json!({"type": "stop", "actor": "running"})).await; + let default = upgrade_actor(&host, &mut mailbox, "running", &wasm, &passing).await; + assert_eq!(default.body["type"], "upgraded"); + assert!(world.read_meta("running").running); + let version = ask_type(&host, &mut mailbox, "running", json!({"type": "version"}), "version").await; + assert_eq!(version.body["version"], "3"); + 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");