Compare commits

..
Author SHA1 Message Date
Sausage Agent 0bd046ed9f broker: add stopped-preserving upgrade without actor execution
ci / check (push) Failing after 1m47s
2026-10-22 06:49:58 +00:00
6 changed files with 187 additions and 137 deletions
+23 -12
View File
@@ -59,17 +59,13 @@ it exits. After 5 seconds, it kills the instance.
## Filesystem
Each instance sees four folders:
Each instance sees three folders:
| Path | Access | Contents |
| --- | --- | --- |
| `/data` | read-write by its actor; readable by every actor | the actor's state |
| `/public` | read-write | the files that the actor publishes |
| `/actor` | read-only | the `actor/` folder of the world |
| `/config` | read-write by every actor | the world's shared `config/` folder; persists across restarts |
`/config` is the same world folder for every instance, including candidates;
writers coordinate by convention, not permissions. Do not put secrets there.
An actor can write only its own `/data`, but can read every actor's `/data` through `/actor/<id>/data/`. `/data` is not private: any unprotected secret file there is readable by every actor.
@@ -129,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`? | 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` |
| `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/<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 |
@@ -140,7 +136,7 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
**Headless.** `sausage run --headless [dir]` (or `sausage --headless`, or `SAUSAGE_HEADLESS=1`) turns the console input off: the host does not read stdin, so end of input (`/dev/null`) does not stop the world. `nix run .#loop -- --headless <world>` passes it through the loop. It runs until SIGTERM or SIGINT (stops every actor, exit code 0) or a broker `exit` request (exit code 75, as above). Without the flag, EOF on stdin or `/quit` ends the host.
Paths are `/actor/...`, `/config/...`, or the sender's `/public/...` or `/data/...`.
Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`.
**Taps.** After `tap`, the sender gets a copy of every message that the
broker routes (direct messages, replies, and publishes, between any actors,
@@ -186,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/<n>/`
(`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.
@@ -216,8 +213,22 @@ 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 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.
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
+34 -19
View File
@@ -70,7 +70,6 @@ pub struct Broker {
impl Broker {
pub fn new(world: World, runtime: Arc<Runtime>) -> Result<Arc<Self>> {
std::fs::create_dir_all(world.config_dir())?;
let metas = world.actor_ids()?.into_iter().map(|id| (id.clone(), world.read_meta(&id))).collect();
Ok(Arc::new_cyclic(|me| Self {
world,
@@ -259,7 +258,6 @@ impl Broker {
Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: false },
Mount { guest: "/public".into(), host: own.join("public"), writable: true },
Mount { guest: "/data".into(), host: own.join("data"), writable: true },
Mount { guest: "/config".into(), host: self.world.config_dir(), writable: true },
],
network: scratch.is_none(),
stderr_log: own.join("stderr.log"),
@@ -315,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));
}
@@ -333,8 +335,7 @@ impl Broker {
Ok((component, tests))
}
/// Make `version` the current `actor.wasm`, `source/`, and `actor.json`,
/// and record whether the actor runs.
/// Make `version` the current `actor.wasm`, `source/`, and `actor.json`.
pub fn install(&self, id: &str, version: u64, running: bool) -> Result<()> {
self.world.install_files(id, version)?;
self.update_meta(id, |m| {
@@ -362,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?;
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)
}
@@ -520,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:#}")})))
}
@@ -567,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()?;
@@ -576,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()?;
+17 -17
View File
@@ -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 },
@@ -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. A stopped actor stays stopped.
async fn install(&mut self, version: u64, component: Option<Component>) -> Result<()> {
/// they are.
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.running)?;
self.broker.install(&self.id, version, 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) -> 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?;
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}));
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?;
bail!("version {version} did not start ({e:#}); rolled back to {previous}");
}
Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous, "running": true}))
Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous}))
}
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;
if self.running {
self.start().await?;
}
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version, "running": self.running}))
self.start().await?;
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version}))
}
/// 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);
+2 -9
View File
@@ -4,7 +4,6 @@
//!
//! ```text
//! <world>/
//! config/ shared configuration, writable by every actor
//! actor/<id>/
//! actor.wasm the current version
//! actor.json version, running, subscriptions, tapped, watches (broker)
@@ -90,9 +89,6 @@ impl World {
Self { root: root.into() }
}
pub fn config_dir(&self) -> PathBuf {
self.root.join("config")
}
pub fn actors_root(&self) -> PathBuf {
self.root.join("actor")
}
@@ -195,7 +191,7 @@ impl World {
/// Map a path in an actor's namespace to a host path.
///
/// - `/actor/<id>/...` and `/config/...` are world paths.
/// - `/actor/<id>/...` is the global filesystem.
/// - `/public/...` and `/data/...` belong to `sender`.
/// - Native senders (the console) can also give absolute host paths.
pub fn resolve(&self, sender: &str, path: &str) -> Result<PathBuf> {
@@ -216,9 +212,6 @@ impl World {
if let Some(rest) = prefixed("/actor") {
return under(self.actors_root(), rest);
}
if let Some(rest) = prefixed("/config") {
return under(self.config_dir(), rest);
}
if let Some(rest) = prefixed("/public") {
return under(self.public_dir(sender), rest);
}
@@ -228,7 +221,7 @@ impl World {
if NATIVE.contains(&sender) && Path::new(path).is_absolute() {
return Ok(PathBuf::from(path));
}
bail!("cannot resolve path {path} for {sender}: use /actor/<id>/..., /config/..., /public/... or /data/...")
bail!("cannot resolve path {path} for {sender}: use /actor/<id>/..., /public/... or /data/...")
}
/// The path of a host file in the global filesystem, if it has one.
+4
View File
@@ -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();
+107 -80
View File
@@ -175,42 +175,6 @@ async fn filesystem_permissions_and_public_sharing() {
host.stop().await;
}
#[tokio::test]
async fn config_is_shared_writable_and_survives_host_restart() {
let (world, host, mut mailbox) = filesystem_world("world-config").await;
let path = "/config/global.json";
let first = r#"{"model":"first"}"#;
let second = r#"{"model":"second"}"#;
ask_type(
&host, &mut mailbox, "writer",
json!({"type": "write", "path": path, "text": first}), "ok",
).await;
let read = ask_type(
&host, &mut mailbox, "reader",
json!({"type": "read", "path": path}), "content",
).await;
assert_eq!(read.body["text"], first);
// The second actor can replace the same file too.
ask_type(
&host, &mut mailbox, "reader",
json!({"type": "write", "path": path, "text": second}), "ok",
).await;
assert_eq!(std::fs::read_to_string(world.config_dir().join("global.json")).unwrap(), second);
assert_eq!(world.resolve("writer", path).unwrap(), world.config_dir().join("global.json"));
assert_eq!(world.resolve("console", path).unwrap(), world.config_dir().join("global.json"));
host.stop().await;
let (host, mut mailbox) = start(&world).await;
for actor in ["writer", "reader"] {
let read = ask_type(
&host, &mut mailbox, actor,
json!({"type": "read", "path": path}), "content",
).await;
assert_eq!(read.body["text"], second);
}
host.stop().await;
}
#[tokio::test]
async fn actors_can_read_but_not_write_other_actor_data() {
let (_, host, mut mailbox) = filesystem_world("actor-data-sharing").await;
@@ -472,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");
@@ -542,50 +613,6 @@ 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");