Merge pull request 'broker: upgrade and rollback keep a stopped actor stopped' (#3 ) from upgrade-keeps-stopped into main

Reviewed-on: #3
broker: upgrade and rollback keep a stopped actor stopped
2026-10-22 10:50:15 +01:00 · 2026-10-22 10:11:36 +01:00
5 changed files with 84 additions and 183 deletions
@@ -125,8 +125,8 @@ 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`?, `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 |
+| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version: a running actor is stopped (normal shutdown) and starts the new version; a stopped actor gets the new version and stays stopped, with its mailbox untouched, until `start`. The answer has `running` |
+| `rollback` | `actor`, `version` | go back to an older version; like `upgrade`, a stopped actor stays stopped. The answer has `running` |
 | `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 |
@@ -182,11 +182,10 @@ 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
-(except the stopped-preserving upgrade described below).
+`actor.wasm` on disk) is tested first, while the old version keeps running.
 Only when its tests pass does the broker copy it to `versions/<n>/`
 (`actor.wasm`, `source/`, `meta.json`) and switch to it. So `versions/` holds
-exactly the versions that were installed (including stopped installations), and instances always run from
+exactly the versions that were installed, 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.
@@ -213,22 +212,8 @@ instance. A version without test files passes.

 **Swap.** The broker holds the mailbox, stops the old instance, makes the
 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.
+it. A stopped actor is not started: it keeps its mailbox and runs the new
+version at the next `start`. A rollback does the same with the older version.

 **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
@@ -313,16 +313,12 @@ impl Broker {
        Ok(component)
    }

-    /// 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)> {
+    /// 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)> {
        let component = self.load(id, wasm)?;
-        let tests = if run_tests {
-            candidate::test(self, id, meta.version, &component, source).await
-        } else {
-            Ok(json!({"skipped": "preserve_stopped"}))
-        };
+        let tests = candidate::test(self, id, meta.version, &component, source).await;
        if let Err(e) = &tests {
            self.log(id, &format!("version {} from {} failed its tests: {e:#}", meta.version, meta.wasm));
        }
@@ -335,7 +331,8 @@ impl Broker {
        Ok((component, tests))
    }

-    /// Make `version` the current `actor.wasm`, `source/`, and `actor.json`.
+    /// Make `version` the current `actor.wasm`, `source/`, and `actor.json`,
+    /// and record whether the actor runs.
    pub fn install(&self, id: &str, version: u64, running: bool) -> Result<()> {
        self.world.install_files(id, version)?;
        self.update_meta(id, |m| {
@@ -363,29 +360,26 @@ 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, true).await?;
+        let (_, tests) = self.try_version(id, meta, wasm, source).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, preserve_stopped: bool) -> Result<Value> {
+    pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> 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, preserve_stopped).await
+        self.upgrade_version(id, &VersionMeta { version, ..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?;
+    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?;
        let (version, requester) = (meta.version, meta.requested_by.clone());
-        let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, preserve_stopped, reply }).await?;
+        let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, reply }).await?;
        result["tests"] = tests;
        Ok(result)
    }
@@ -524,7 +518,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, false).await,
+            Some(_) => self.upgrade_version(id, &vmeta, &wasm, source).await,
        };
        Some(result.unwrap_or_else(|e| json!({"type": "error", "actor": id, "reason": format!("{e:#}")})))
    }
@@ -571,11 +565,7 @@ impl Broker {
            }
            "upgrade" => {
                let (wasm, source, meta) = version_meta()?;
-                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
+                self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta).await
            }
            "rollback" => {
                let id = actor()?;
@@ -584,15 +574,8 @@ impl Broker {
                let _guard = lock.lock().await;
                self.control(&id, |reply| Control::Rollback { version, 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
-            },
+            "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()?;
@@ -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, preserve_stopped: bool, reply: Reply },
+    Upgrade { version: u64, component: Component, requester: String, reply: Reply },
    Rollback { version: u64, reply: Reply },
    Stop { reply: Reply },
    Start { reply: Reply },
@@ -220,51 +220,51 @@ 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>, running: bool) -> Result<()> {
+    /// they are. A stopped actor stays stopped.
+    async fn install(&mut self, version: u64, component: Option<Component>) -> 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, running)?;
+        self.broker.install(&self.id, version, self.running)?;
        self.version = version;
        self.component = Some(component);
-        self.running = running;
        self.restarts = 0;
        self.restart_at = None;
        Ok(())
    }

-    async fn upgrade(&mut self, version: u64, component: Component, requester: String, preserve_stopped: bool) -> Result<Value> {
+    async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result<Value> {
        let previous = self.version;
-        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.install(version, Some(component)).await?;
+        if !self.running {
+            // Installed, not started: main runs and mail is read at the next `start`.
+            return Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous, "running": false}));
        }
        self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester });
        if let Err(e) = self.start().await {
            self.rollback(previous).await?;
            bail!("version {version} did not start ({e:#}); rolled back to {previous}");
        }
-        Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous}))
+        Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous, "running": true}))
    }

    async fn rollback(&mut self, version: u64) -> Result<Value> {
        let from = self.version;
-        self.install(version, None, true).await?;
+        self.install(version, None).await?;
        self.probation = None;
-        self.start().await?;
-        Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version}))
+        if self.running {
+            self.start().await?;
+        }
+        Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version, "running": self.running}))
    }

    /// 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, preserve_stopped, reply } => {
-                let _ = reply.send(self.upgrade(version, component, requester, preserve_stopped).await);
+            Control::Upgrade { version, component, requester, reply } => {
+                let _ = reply.send(self.upgrade(version, component, requester).await);
            }
            Control::Rollback { version, reply } => {
                let _ = reply.send(self.rollback(version).await);
@@ -38,10 +38,6 @@ 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();
@@ -436,113 +436,6 @@ 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");
@@ -613,6 +506,50 @@ async fn rolls_back_a_crash_loop_to_the_previous_version_without_restoring_data(
    host.stop().await;
 }

+#[tokio::test]
+async fn upgrade_and_rollback_keep_a_stopped_actor_stopped() {
+    let world = create_world("upgrade-stopped");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "source");
+    spawn_type(&host, &mut mailbox, "sleeper", &wasm, &source).await;
+    let stopped = request(&host, &mut mailbox, json!({"type": "stop", "actor": "sleeper"})).await;
+    assert_eq!(stopped.body["type"], "stopped");
+
+    let upgraded = upgrade_actor(&host, &mut mailbox, "sleeper", &wasm, &source).await;
+    assert_eq!(upgraded.body["type"], "upgraded", "{upgraded:?}");
+    assert_eq!(upgraded.body["version"], 2);
+    assert_eq!(upgraded.body["running"], false);
+    let meta = world.read_meta("sleeper");
+    assert_eq!((meta.version, meta.running), (Some(2), false));
+
+    // Mail waits in the mailbox: the stopped actor reads nothing.
+    let waiting = host.broker.send("console", Some("sleeper"), None, None, json!({"type": "version"}));
+    assert!(
+        tokio::time::timeout(Duration::from_millis(500), receive_reply(&mut mailbox, &waiting, Duration::from_secs(10)))
+            .await
+            .is_err(),
+        "a stopped actor must not answer"
+    );
+
+    let rolled_back = request(&host, &mut mailbox, json!({"type": "rollback", "actor": "sleeper", "version": 1})).await;
+    assert_eq!(rolled_back.body["type"], "rolled_back", "{rolled_back:?}");
+    assert_eq!(rolled_back.body["running"], false);
+    assert!(!world.read_meta("sleeper").running);
+
+    // `start` runs the installed version and delivers the waiting message.
+    let started = request(&host, &mut mailbox, json!({"type": "start", "actor": "sleeper"})).await;
+    assert_eq!(started.body["type"], "started");
+    let answer = receive_reply(&mut mailbox, &waiting, Duration::from_secs(10)).await;
+    assert_eq!(answer.body["version"], "1");
+
+    // A running actor still restarts on the new version.
+    let running = upgrade_actor(&host, &mut mailbox, "sleeper", &wasm, &source).await;
+    assert_eq!(running.body["running"], true);
+    assert_eq!(ask(&host, &mut mailbox, "sleeper", json!({"type": "version"})).await.body["version"], "3");
+    host.stop().await;
+}
+
 #[tokio::test]
 async fn explicit_rollback_keeps_current_data() {
    let world = create_world("explicit-rollback");
Author	SHA1	Message	Date
surma	3d3f6f35fb	Merge pull request 'broker: upgrade and rollback keep a stopped actor stopped' (#3 ) from upgrade-keeps-stopped into main ci / check (push) Successful in 13s Details Reviewed-on: #3	2026-10-22 10:50:15 +01:00
surma	f1af07506f	broker: upgrade and rollback keep a stopped actor stopped ci / check (push) Successful in 8m42s Details ci / check (pull_request) Successful in 4m38s Details A stopped actor gets the new version installed, keeps its mailbox and does not run main until the next start. Running actors still restart on the new version. The upgraded and rolled_back answers say whether the actor runs.	2026-10-22 10:11:36 +01:00