broker: add stopped-preserving upgrade without actor execution

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/<n>/`: 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/<n>/` 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<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
-        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<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> Result<Value> {
+    pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta, preserve_stopped: bool) -> Result<Value> {
         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<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
-        let (component, tests) = self.try_version(id, meta, wasm, source).await?;
+    async fn upgrade_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, preserve_stopped: bool) -> Result<Value> {
+        // 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()?;

cli/src/supervisor.rs

@@ -23,7 +23,7 @@ pub type Reply = oneshot::Sender<Result<Value>>;
 
 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<Component>) -> Result<()> {
+    async fn install(&mut self, version: u64, component: Option<Component>, 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<Value> {
+    async fn upgrade(&mut self, version: u64, component: Component, requester: String, preserve_stopped: bool) -> Result<Value> {
         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<Value> {
         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);

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::<String, std::fs::File>::new();
     // Files that this actor serves at /actor/<id>/fs/ (path -> text).
     let mut served = std::collections::BTreeMap::<String, String>::new();

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");