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
10 changed files with 186 additions and 234 deletions
+22 -31
View File
@@ -59,15 +59,13 @@ it exits. After 5 seconds, it kills the instance.
## Filesystem
Each instance sees these mounts:
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-only | one non-secret global JSON file, owned by `shared` |
| `/global` | read-write for `shared`, read-only for all other actors | backing directory containing that same `config` file |
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.
@@ -79,28 +77,6 @@ Files are plain POSIX files: a write is visible at once. To publish a file
so that readers never see half of it, write a temporary file and `rename` it
(the Rust SDK has `write_atomic`).
`/config` is global configuration, never credentials (in particular, never the
LLM proxy token). The host initializes it to `{}` without reading or copying
legacy actor settings. Configuration schema and typed edits belong to `shared`,
not the host. Ordinary actors cannot write either `/config` or its alias
`/global/config`: WASI mounts are read-only, with no message sender checks.
WASI preopens directories rather than individual files. The host mounts the
isolated world `global/` directory at `/` read-only and at `/global` with the
owner's write access. `shared` publishes atomically by writing a sibling such
as `/global/config.tmp`, closing it, then renaming it to `/global/config`.
New opens of `/config` see the replacement, including in already-running
actors; an already-open file retains its old inode. The root mount does not
expose the rest of the host world. Candidate tests get a scratch copy of
`global/`, so even `shared` tests cannot modify the live file.
On disk, the one JSON file is `<world>/global/config`. `<world>/config` is a
relative symlink to it, for native/world filesystem readers (including an
export of the world directory). Host-side atomic writers must replace
`<world>/global/config`, not replace the symlink. Starting an older world
adds the empty file and alias without changing actor state; an existing
regular `<world>/config` is moved into the backing directory on first use.
Send large or binary data as a path under `/actor/<id>/public/`, not inside a
message.
@@ -149,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 |
@@ -206,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.
@@ -236,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 -21
View File
@@ -255,10 +255,6 @@ impl Broker {
}
InstanceConfig {
mounts: vec![
// WASI preopens directories, not files. This isolated directory
// supplies /config without exposing the rest of the world.
Mount { guest: "/".into(), host: scratch.map_or_else(|| self.world.global_dir(), |p| p.join("global")), writable: false },
Mount { guest: "/global".into(), host: scratch.map_or_else(|| self.world.global_dir(), |p| p.join("global")), writable: id == "shared" },
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 },
@@ -317,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));
}
@@ -335,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| {
@@ -364,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)
}
@@ -522,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:#}")})))
}
@@ -569,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()?;
@@ -578,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()?;
+1 -1
View File
@@ -97,7 +97,7 @@ impl Candidate<'_> {
let dir = self.broker.world.scratch_dir().join(format!("{}-{}-{n}", self.id, std::process::id()));
let scratch = Scratch { dir: dir.clone() };
let world = &self.broker.world;
for (from, name) in [(world.data_dir(self.id), "data"), (world.public_dir(self.id), "public"), (world.global_dir(), "global")] {
for (from, name) in [(world.data_dir(self.id), "data"), (world.public_dir(self.id), "public")] {
std::fs::create_dir_all(dir.join(name))?;
if from.is_dir() {
world::copy_tree(&from, &dir.join(name))?;
-1
View File
@@ -48,7 +48,6 @@ impl Host {
if lock.try_lock().is_err() {
bail!("another sausage process runs {}", root.display());
}
world.ensure_config()?;
let broker = Broker::new(world.clone(), runtime::Runtime::new()?)?;
world::write_atomic(&world.public_dir("broker").join("PROTOCOL.md"), PROTOCOL.as_bytes())?;
+1 -3
View File
@@ -23,9 +23,7 @@ use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpHooks, WasiHttpVi
use crate::broker::Broker;
use crate::vfs::{Vfs, VfsState, VfsView};
/// A host folder that the guest sees at `guest`. File-like root paths such
/// as `/config` use an isolated backing directory mounted at `/`; owners
/// get a separate writable directory alias for atomic replacement.
/// A host folder that the guest sees at `guest`.
#[derive(Clone, Debug)]
pub struct Mount {
pub guest: String,
+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
View File
@@ -6,8 +6,6 @@
//! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for
//! the reply. Other paths go to `wasmtime-wasi` unchanged except that an
//! actor directory listing advertises `fs/` when its source docs declare it.
//! `/config` and `/global/config` are native mount paths: they deliberately
//! use WASI mount permissions rather than actor-served write requests.
//!
//! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}`
//! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`,
-37
View File
@@ -4,8 +4,6 @@
//!
//! ```text
//! <world>/
//! global/config non-secret global JSON (shared owns writes)
//! config -> global/config world-facing read alias
//! actor/<id>/
//! actor.wasm the current version
//! actor.json version, running, subscriptions, tapped, watches (broker)
@@ -91,37 +89,6 @@ impl World {
Self { root: root.into() }
}
/// Isolated backing directory for the root /config mount and the
/// owner's writable /global alias. Never copy legacy actor config here.
pub fn global_dir(&self) -> PathBuf {
self.root.join("global")
}
pub fn config_path(&self) -> PathBuf {
self.global_dir().join("config")
}
/// Add the non-secret file to new and pre-existing worlds. The symlink
/// is for native world readers; actor mounts use the backing directory.
pub fn ensure_config(&self) -> Result<()> {
std::fs::create_dir_all(self.global_dir())?;
let alias = self.root.join("config");
if !self.config_path().exists() {
if alias.is_file() && !alias.is_symlink() {
std::fs::rename(&alias, self.config_path())?;
} else {
write_atomic(&self.config_path(), b"{}\n")?;
}
}
match std::fs::read_link(&alias) {
Ok(target) if target == Path::new("global/config") => {}
Ok(_) => bail!("{} must point to global/config", alias.display()),
Err(_) if alias.try_exists()? => bail!("{} already exists", alias.display()),
Err(_) => std::os::unix::fs::symlink("global/config", alias)?,
}
Ok(())
}
pub fn actors_root(&self) -> PathBuf {
self.root.join("actor")
}
@@ -164,7 +131,6 @@ impl World {
for id in NATIVE {
world.ensure_actor_dirs(id)?;
}
world.ensure_config()?;
Ok(world)
}
@@ -243,9 +209,6 @@ impl World {
let prefixed = |prefix: &str| {
path.strip_prefix(prefix).filter(|rest| rest.is_empty() || rest.starts_with('/'))
};
if path == "/config" || path == "/global/config" {
return Ok(self.config_path());
}
if let Some(rest) = prefixed("/actor") {
return under(self.actors_root(), rest);
}
+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 -121
View File
@@ -47,83 +47,6 @@ async fn receive_shutdown_messages(
.expect("shutdown output or lifecycle reply did not arrive");
}
#[tokio::test]
async fn global_config_is_readable_owner_written_and_atomically_replaced() {
let world = create_world("global-config");
let (host, mut mailbox) = start(&world).await;
let wasm = toolbox_wasm().clone();
for actor in ["shared", "reader"] {
let source = passing_source(&world, actor);
spawn_type(&host, &mut mailbox, actor, &wasm, &source).await;
}
let initial = ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": "/config"}), "content").await;
assert_eq!(initial.body["text"], "{}\n");
for path in ["/config", "/global/config", "/global/config.tmp", "/actor/shared/data/../../../../global/config"] {
ask_type(&host, &mut mailbox, "reader", json!({"type": "write", "path": path, "text": "not allowed"}), "error").await;
}
ask_type(&host, &mut mailbox, "reader", json!({"type": "rename", "path": "/config", "to": "/global/stolen"}), "error").await;
// The root view stays read-only even for the owner; publication uses its
// writable directory alias, with a sibling temporary file and rename.
ask_type(&host, &mut mailbox, "shared", json!({"type": "write", "path": "/config", "text": "not allowed"}), "error").await;
let next = "{\"llm\":{\"thinking\":\"medium\"}}\n";
ask_type(&host, &mut mailbox, "shared", json!({"type": "write", "path": "/global/config.tmp", "text": next}), "ok").await;
let before = ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": "/config"}), "content").await;
assert_eq!(before.body["text"], "{}\n");
// A native/world reader holding the old inode sees the complete old file.
let mut old = std::fs::File::open(world.root.join("config")).unwrap();
ask_type(&host, &mut mailbox, "shared", json!({"type": "rename", "path": "/global/config.tmp", "to": "/global/config"}), "ok").await;
let after = ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": "/config"}), "content").await;
assert_eq!(after.body["text"], next);
assert_eq!(std::fs::read_to_string(world.root.join("config")).unwrap(), next);
assert_eq!(std::fs::read_to_string(world.resolve("reader", "/config").unwrap()).unwrap(), next);
let mut previous = String::new();
std::io::Read::read_to_string(&mut old, &mut previous).unwrap();
assert_eq!(previous, "{}\n");
assert!(!world.global_dir().join("config.tmp").exists());
// The mount must not give shared writable access to any other world data.
ask_type(&host, &mut mailbox, "shared", json!({"type": "write", "path": "/global/../actor/reader/data/escape", "text": "no"}), "error").await;
host.stop().await;
let (host, mut mailbox) = start(&world).await;
let persisted = ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": "/config"}), "content").await;
assert_eq!(persisted.body["text"], next);
host.stop().await;
}
#[tokio::test]
async fn global_config_candidates_cannot_change_the_live_file() {
let world = create_world("candidate-global-config");
let (host, mut mailbox) = start(&world).await;
let source = source_with_test(&world, "shared", json!({"steps": [
{"send": {"type": "write", "path": "/global/next", "text": "{\"test\":true}"}, "expect": {"type": "ok"}},
{"send": {"type": "rename", "path": "/global/next", "to": "/global/config"}, "expect": {"type": "ok"}},
{"send": {"type": "read", "path": "/config"}, "expect": {"type": "content", "text": "{\"test\":true}"}}
]}));
spawn_type(&host, &mut mailbox, "shared", toolbox_wasm(), &source).await;
assert_eq!(std::fs::read_to_string(world.config_path()).unwrap(), "{}\n");
host.stop().await;
}
#[tokio::test]
async fn global_config_is_added_to_existing_worlds_without_copying_actor_settings() {
let world = create_world("old-world-config");
std::fs::remove_file(world.root.join("config")).unwrap();
std::fs::remove_dir_all(world.global_dir()).unwrap();
world.ensure_actor_dirs("shared").unwrap();
// Synthetic marker only; no actual credentials are read in this test.
std::fs::write(world.data_dir("shared").join("config.json"), "legacy-private-marker").unwrap();
let (host, _) = start(&world).await;
assert_eq!(std::fs::read_to_string(world.root.join("config")).unwrap(), "{}\n");
assert_eq!(std::fs::read_to_string(world.data_dir("shared").join("config.json")).unwrap(), "legacy-private-marker");
host.stop().await;
// Preserve an operator-provided global config when adding the mounts.
std::fs::remove_file(world.root.join("config")).unwrap();
std::fs::remove_dir_all(world.global_dir()).unwrap();
std::fs::write(world.root.join("config"), "{\"operator\":true}").unwrap();
let (host, _) = start(&world).await;
assert_eq!(std::fs::read_to_string(world.root.join("config")).unwrap(), "{\"operator\":true}");
host.stop().await;
}
#[tokio::test]
async fn routes_replies_publishes_and_reports_undeliverable() {
let world = create_world("routing");
@@ -513,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");
@@ -583,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");