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
Sausage Agent a72e42a8c9 headless: register signal handlers synchronously, split tests, drop deployment examples and wording
ci / check (pull_request) Successful in 1m36s
ci / check (push) Successful in 11s
2026-10-18 15:31:41 +00:00
Sausage Agent 01a821ab72 Add --headless mode (no stdin read, runs until SIGTERM/exit) + systemd/NixOS examples
ci / check (push) Successful in 2m44s
2026-10-18 15:24:20 +00:00
Sausage Agent c204e1ef0c docs: clarify actor data is world-readable
ci / check (pull_request) Successful in 12s
ci / check (push) Successful in 6s
2026-10-17 17:22:50 +00:00
Sausage Agent 6923ca3e60 preserve the packaged runtime Rust toolchain
ci / check (push) Successful in 6s
2026-10-17 16:36:08 +00:00
Sausage Agent c504532513 drive packaged host runtime check over pipes
ci / check (push) Failing after 1m16s
2026-10-17 16:33:31 +00:00
Sausage Agent eb35a79698 trace the packaged host runtime check
ci / check (push) Failing after 1m47s
2026-10-17 16:29:02 +00:00
Sausage Agent bfcd71e2f4 test packaged host cargo runtime with an empty PATH
ci / check (push) Failing after 1m33s
2026-10-17 16:25:14 +00:00
11 changed files with 360 additions and 45 deletions
+25 -5
View File
@@ -63,10 +63,12 @@ Each instance sees three folders:
| Path | Access | Contents |
| --- | --- | --- |
| `/data` | read-write | the actor's state |
| `/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 |
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.
`/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json`
(installed version, running, subscriptions, tapped, watches), `source/`, `public/`, `data/`,
`stderr.log`, and `versions/<n>/`. `ls /actor` lists all actors.
@@ -123,14 +125,16 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
| `type` | Fields | Effect |
| --- | --- | --- |
| `spawn` | `actor`, `wasm`, `source`?, `built_by`? | test, then start a new actor |
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version |
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`?, `preserve_stopped`? (boolean, default `false`) | test, then swap the version; with `preserve_stopped: true`, a stopped actor is compiled and installed without execution and stays stopped |
| `rollback` | `actor`, `version` | go back to an older version |
| `stop`, `start` | `actor` | stop or start an actor |
| `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<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 |
| `tap`, `untap` | | receive (or not) a copy of every message the broker routes |
| `watch`, `unwatch` | `path` | receive (or not) `fs_changed` for every change of `path` or anything under it; answer `{"type": "watching" \| "unwatched", "path"}` with the path as an `/actor/...` path |
| `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) |
| `exit` | | answer `exiting`, stop every actor, and end the `sausage` process with exit code 75 (a wrapper like `nix run .#loop` starts it again, updated) |
**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/...`, or the sender's `/public/...` or `/data/...`.
@@ -178,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.
@@ -210,6 +215,21 @@ instance. A version without test files passes.
top-level `actor.wasm` and `source/` a copy of the new version, and starts
it. A rollback does the same with the older version.
**Stopped-preserving upgrade.** Send `upgrade` with `"preserve_stopped": true`
to install a stopped actor's new program without starting it, running `main`
(even as a candidate), or reading or sending any of its mail. The host compiles
the WASM and copies the new version, source and metadata as usual, increments
the version, and keeps `actor.json`'s `running: false`. Its mailbox, `data/`
and `public/` stay untouched. Because candidate tests execute `main`, they are
not run in this case: the `upgraded` reply contains
`"tests": {"skipped": "preserve_stopped"}`. Test the build separately before
installing it this way. An explicit `start` later runs the installed version
and delivers the queued mail. No startup/crash probation runs for this stopped
installation. Running actors still use the normal tested swap, even with this
option. Omitting the option (or `false`) retains the existing behavior,
including starting a stopped actor. Explicit stop/start requests serialize
with upgrades of the same actor.
**Crashes.** The broker restarts a crashed actor after 0.5 s, then doubles
the delay up to 30 s. A new version that crashes 3 times within 30 seconds
after an upgrade goes back to the previous version. The actor that asked for
+13 -1
View File
@@ -48,6 +48,18 @@ From a clone (`git clone https://gitea.surma.technology/surma/sausage`), the
same commands are `nix run`, `nix run -- init`, `nix run -- run` and
`nix run .#loop`.
### Headless
`--headless` (or `SAUSAGE_HEADLESS=1`) turns the console input off: the host
never reads stdin, so end of input does not stop the world. It runs until
SIGTERM/SIGINT (clean stop, exit code 0) or a broker `exit` request (exit
code 75, so the loop starts the newest version). Create the world first
(`init`), then pass the flag through the loop:
```sh
nix run git+https://gitea.surma.technology/surma/sausage#loop -- --headless my-world
```
In the console:
```text
@@ -66,7 +78,7 @@ actor.wasm the current version
actor.json installed version, running, subscriptions
source/ the source of the current version
public/ files that the actor publishes
data/ the actor's state
data/ actor state; owner-writable, world-readable via /actor/<id>/data/ (not private)
stderr.log what the actor and the broker write about it
versions/<n>/ every installed version: actor.wasm, source/, meta.json
```
+2
View File
@@ -3,3 +3,5 @@ People talk to you through other actors. Your text goes to your subscribers.
Before writing an actor, read `/actor/broker/public/PROTOCOL.md` for messages, the broker, `cargo`, and tests.
Read the SDK at `/actor/cargo/public/sdk/` before you write an actor.
You can change your own source at `/actor/agent/source`.
The global filesystem is readable by every actor, including every actor's `/data` at `/actor/<id>/data/`. An actor can write only its own `/data` and `/public`. `/data` is not private: any unprotected secret file there is readable by every actor.
+36 -18
View File
@@ -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()?;
+2 -1
View File
@@ -3,7 +3,8 @@
//! - `@<actor> <json>` sends a JSON body. `@<actor> <text>` sends
//! `{"type": "text", "text": ...}`.
//! - A line without `@` goes to `agent` as text.
//! - `/help`, `/quit`. EOF on stdin also quits.
//! - `/help`, `/quit`. EOF on stdin also quits (not with `--headless`, which
//! never reads stdin).
use std::sync::Arc;
+43 -7
View File
@@ -5,6 +5,7 @@ use anyhow::{Context, Result, bail};
use clap::{Parser, Subcommand};
use include_dir::{Dir, include_dir};
use serde_json::json;
use tokio::signal::unix::Signal;
use sausage::world::{ActorMeta, World};
use sausage::{Host, cargo_actor, console};
@@ -17,6 +18,11 @@ static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
#[derive(Parser)]
#[command(version, about)]
struct Cli {
/// Do not read stdin, so end of input does not stop the world. It runs
/// until SIGTERM (or SIGINT), or until an actor sends
/// `@broker {"type":"exit"}`. Also set by SAUSAGE_HEADLESS=1.
#[arg(long, global = true)]
headless: bool,
#[command(subcommand)]
command: Option<Cmd>,
}
@@ -91,13 +97,41 @@ async fn init(dir: PathBuf) -> Result<()> {
Ok(())
}
async fn run(dir: PathBuf) -> Result<()> {
/// Install the SIGTERM and SIGINT handlers now (not when first polled).
fn termination_signals() -> Result<(Signal, Signal)> {
use tokio::signal::unix::{SignalKind, signal};
let term = signal(SignalKind::terminate()).context("install SIGTERM handler")?;
let int = signal(SignalKind::interrupt()).context("install SIGINT handler")?;
Ok((term, int))
}
/// Resolves on SIGTERM or SIGINT.
async fn termination((mut term, mut int): (Signal, Signal)) {
tokio::select! {
_ = term.recv() => {}
_ = int.recv() => {}
}
}
async fn run(dir: PathBuf, headless: bool) -> Result<()> {
// Register the signal handlers before the host starts, so a signal that
// arrives early is never the default (kill) action.
let signals = if headless { Some(termination_signals()?) } else { None };
let host = Host::start(&dir).await?;
tokio::spawn(console::run_mailbox(host.mailbox(console::ID)));
let broker = host.broker.clone();
let exit = tokio::select! {
_ = console::run_input(host.broker.clone()) => false,
_ = broker.exit_requested() => true,
let exit = match signals {
Some(signals) => {
println!("Sausage headless: running until SIGTERM or an exit request.");
tokio::select! {
_ = termination(signals) => false,
_ = broker.exit_requested() => true,
}
}
None => tokio::select! {
_ = console::run_input(host.broker.clone()) => false,
_ = broker.exit_requested() => true,
},
};
println!("Stopping actors …");
host.stop().await;
@@ -113,16 +147,18 @@ async fn run(dir: PathBuf) -> Result<()> {
#[tokio::main]
async fn main() -> Result<()> {
match Cli::parse().command {
let cli = Cli::parse();
let headless = cli.headless || std::env::var("SAUSAGE_HEADLESS").is_ok_and(|v| !v.is_empty() && v != "0");
match cli.command {
Some(Cmd::Init { dir }) => init(dir).await,
Some(Cmd::Run { dir }) => run(dir).await,
Some(Cmd::Run { dir }) => run(dir, headless).await,
None => {
let dir = PathBuf::from("world");
if !dir.join("actor").is_dir() {
println!("No world in ./world yet, creating it (same as `sausage init`).");
init(dir.clone()).await?;
}
run(dir).await
run(dir, headless).await
}
}
}
+14 -9
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 },
@@ -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);
+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();
+44
View File
@@ -39,3 +39,47 @@ fn bare_command_without_a_world_starts_init() {
// It asks before it creates anything.
assert!(!dir.join("world").exists());
}
/// Start a headless host on a fresh world with stdin at EOF; returns once it
/// printed its ready line (printed after the signal handlers are installed).
fn start_headless(label: &str) -> (std::process::Child, std::io::Lines<std::io::BufReader<std::process::ChildStdout>>) {
use std::io::{BufRead, BufReader};
let dir = empty_dir(label);
let world = dir.join("world");
sausage::World::create(&world).unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_sausage"))
.args(["run", "--headless"])
.arg(&world)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.unwrap();
let mut lines = BufReader::new(child.stdout.take().unwrap()).lines();
let line = lines.next().expect("host exited before it started").unwrap();
assert!(line.contains("headless"), "unexpected first line: {line}");
(child, lines)
}
/// With stdin at EOF (as soon as the host starts), a headless host stays up.
#[test]
fn headless_ignores_stdin_eof() {
let (mut child, _lines) = start_headless("headless-eof");
std::thread::sleep(std::time::Duration::from_millis(500));
let alive = child.try_wait().unwrap().is_none();
let _ = child.kill();
let _ = child.wait();
assert!(alive, "headless host exited on stdin EOF");
}
/// SIGTERM right after the ready line stops the actors and exits with 0.
#[test]
fn headless_stops_cleanly_on_sigterm_right_after_ready() {
let (mut child, lines) = start_headless("headless-term");
let pid = child.id().to_string();
assert!(Command::new("kill").args(["-TERM", &pid]).status().unwrap().success());
let rest: Vec<String> = lines.map(|l| l.unwrap()).collect();
let status = child.wait().unwrap();
assert!(status.success(), "exit status {status:?}, output {rest:?}");
assert!(rest.iter().any(|l| l.contains("Stopping actors")), "output {rest:?}");
}
+107
View File
@@ -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");
+70 -4
View File
@@ -21,9 +21,11 @@
targets = [ "wasm32-wasip2" ];
extensions = [ "rust-src" ];
};
# Crane builds the dependencies once (`cargoArtifacts`, keyed only on
# Cargo.toml/Cargo.lock), so a change to our own crates rebuilds
# just them.
# The runtime cargo actor needs Cargo, rustc (with wasm32-wasip2), and
# the native linker, even when the host itself is started with a bare
# PATH. Keep this same runtime path in the packaged host and exercise
# the installed binary below with PATH deliberately empty.
runtimePath = pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ];
craneLib = (crane.mkLib pkgs).overrideToolchain rust;
common = {
pname = "sausage";
@@ -42,12 +44,13 @@
});
sausage = craneLib.buildPackage (common // {
inherit cargoArtifacts;
doNotRemoveReferencesToRustToolchain = true;
cargoExtraArgs = "--locked -p sausage";
doCheck = false;
nativeBuildInputs = [ pkgs.makeWrapper ];
# The `cargo` actor runs cargo at runtime. Build scripts need a C linker.
postInstall = ''
wrapProgram $out/bin/sausage --prefix PATH : ${pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ]}
wrapProgram $out/bin/sausage --prefix PATH : ${runtimePath}
'';
meta.mainProgram = "sausage";
});
@@ -91,6 +94,69 @@
rm -f cli/tests/actors/toolbox/Cargo.lock
'';
});
# Exercise the installed host exactly as the loop launches it: the
# caller has no Cargo on PATH, so only the packaged runtime toolchain
# can run the native tests and the wasm32-wasip2 build.
checks.runtime-toolchain = pkgs.runCommand "sausage-runtime-toolchain" { } ''
set -eu
root="$TMPDIR/runtime-toolchain"
world="$root/world"
emptyPath="$root/empty-bin"
mkdir -p "$emptyPath" "$root/home" "$root/cargo-home"
for actor in broker console cargo fs; do
mkdir -p "$world/actor/$actor/public" "$world/actor/$actor/data"
done
source="$world/actor/console/public/probe"
mkdir -p "$source/src" "$source/tests"
cat > "$source/Cargo.toml" <<'TOML'
[package]
name = "runtime-probe"
version = "0.1.0"
edition = "2024"
TOML
printf 'fn main() {}\n' > "$source/src/main.rs"
cat > "$source/tests/probe.rs" <<'RUST'
#[test]
fn native_test_harness_runs() {
assert_eq!(2 + 2, 4);
}
RUST
echo "runtime-check: run packaged host with an empty PATH"
coproc host {
${pkgs.coreutils}/bin/env -i PATH="$emptyPath" HOME="$root/home" TMPDIR="$root" \
CARGO_HOME="$root/cargo-home" ${sausage}/bin/sausage run "$world"
}
hostInput=''${host[1]}
hostOutput=''${host[0]}
hostPid=$host_PID
trap 'kill "$hostPid" 2>/dev/null || true' EXIT
printf '%s\n' '@cargo {"type":"build","source":"/actor/console/public/probe"}' >&"$hostInput"
built=0
while IFS= read -r -t 300 line <&"$hostOutput"; do
echo "$line"
case "$line" in
*'"type":"build_failed"'*)
echo "packaged cargo actor build failed with an empty caller PATH" >&2
exit 1
;;
*'"type":"built"'*)
case "$line" in
*'"passed":1'*) built=1 ;;
*) echo "native test did not pass" >&2; exit 1 ;;
esac
break
;;
esac
done
if [ "$built" -ne 1 ]; then
echo "packaged host did not build the probe" >&2
exit 1
fi
printf '%s\n' '/quit' >&"$hostInput"
wait "$hostPid"
mkdir -p "$out"
touch "$out/passed"
'';
apps.default = { type = "app"; program = "${sausage}/bin/sausage"; };
apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };