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
8 changed files with 226 additions and 198 deletions
+29 -29
View File
@@ -59,27 +59,15 @@ it exits. After 5 seconds, it kills the instance.
## Filesystem
The world folder is the foundation of every instance's filesystem:
Each instance sees three folders:
| Path | Access | Contents |
| --- | --- | --- |
| `/` | read-write by every actor | the whole world folder |
| `/self` | read-write | alias for the actor's own folder (`public/`, `data/`, `source/`, etc.) |
| `/data` | read-write | alias for the actor's state folder |
| `/public` | read-write | alias for the actor's published files |
| `/actor` | read-write by every actor | the `actor/` folder of the world |
| `/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 |
Every actor can read and write every file in the world, including other actors'
`/actor/<id>/data/` and `/actor/<id>/public/`. Nothing there is private.
Creating a folder in the world makes it visible to running actors immediately:
`/config`, `/docs`, `/www`, and `/skills` need no new mounts or host restart.
New worlds include an empty `/config` for global configuration by convention;
existing worlds can simply create it. Keep secrets out of global configuration.
Candidate instances see the same writable world root. Their `/self` alias
points to the scratch actor folder; `/public` and `/data` still point to its
corresponding children, and network access remains off;
absolute world paths are not isolated from the live 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/`,
@@ -137,21 +125,18 @@ 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 |
| `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 absolute world path |
| `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 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 absolute world paths, or the sender's `/self/...`, `/public/...` or
`/data/...` aliases. Native senders (such as the console) can also supply
absolute host paths; apart from `/actor` and these aliases,
these retain their host meaning.
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,
@@ -191,16 +176,17 @@ change of `path` or a path under it (`/actor/a` covers `/actor/a/x`, not
- Events come from `fs`. Like tap copies, they are not routed, a full
mailbox drops them, and they never block the writer. Never answer them.
- If the OS drops events, every watcher gets one
`{"type": "fs_changed", "path": "/", "op": "rescan"}`: anything may
`{"type": "fs_changed", "path": "/actor", "op": "rescan"}`: anything may
have changed.
- A watch is saved in `actor.json` (`"watches": [...]`) and survives
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.
@@ -227,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
+39 -24
View File
@@ -255,9 +255,7 @@ impl Broker {
}
InstanceConfig {
mounts: vec![
Mount { guest: "/".into(), host: self.world.root.clone(), writable: true },
Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: true },
Mount { guest: "/self".into(), host: own.clone(), writable: true },
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 },
],
@@ -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)
}
@@ -447,7 +451,7 @@ impl Broker {
self.host_error("the file watcher dropped events; watchers get a rescan, actors a full pass");
let watchers: Vec<String> =
self.metas.lock().unwrap().iter().filter(|(_, m)| !m.watches.is_empty()).map(|(id, _)| id.clone()).collect();
self.inject(&watchers, vfs::ID, json!({"type": "fs_changed", "path": "/", "op": "rescan"}));
self.inject(&watchers, vfs::ID, json!({"type": "fs_changed", "path": "/actor", "op": "rescan"}));
let report = self.scan().await;
for change in &report {
eprintln!("sausage: {change}");
@@ -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()?;
@@ -601,9 +616,9 @@ impl Broker {
let on = ty == "watch";
let path_s = b["path"].as_str().ok_or_else(|| anyhow!("missing \"path\""))?;
let global = self.world.global_path(&self.world.resolve(&env.from, path_s)?);
let path = global.ok_or_else(|| anyhow!("{path_s} is not under the world root"))?;
let path = global.ok_or_else(|| anyhow!("{path_s} is not under /actor"))?;
let path = match path.trim_end_matches('/') {
"" => "/".to_string(),
"" => "/actor".to_string(),
p => p.to_string(),
};
self.update_meta(&env.from, |m| {
@@ -651,7 +666,7 @@ fn same_file(a: &Path, b: &Path) -> bool {
/// Whether the namespace path `path` is `prefix` or inside it.
fn under(path: &str, prefix: &str) -> bool {
prefix == "/" || path.strip_prefix(prefix).is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
path.strip_prefix(prefix).is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
}
pub fn truncate(s: &str, max: usize) -> 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 -2
View File
@@ -390,8 +390,8 @@ impl preopens::Host for VfsView<'_> {
fn get_directories(&mut self) -> wasmtime::Result<Vec<(Resource<types::Descriptor>, String)>> {
let dirs = preopens::Host::get_directories(&mut self.inner)?;
for (fd, name) in &dirs {
if name == "/actor" || name == "/" {
self.state.dirs.insert(fd.rep(), name.clone());
if name == "/actor" {
self.state.dirs.insert(fd.rep(), "/actor".to_string());
}
}
Ok(dirs)
+8 -10
View File
@@ -5,7 +5,7 @@
//! Events are coalesced per path: once a path is quiet for `QUIET` (or has
//! kept changing for `MAX_DELAY`), it becomes one change
//! `{path, op: write | create | remove | rename, to?}` with a namespace path
//! (relative to the world root, such as `/actor/...` or `/config/...`). Then the broker
//! (`/actor/...`). Then the broker
//!
//! - sends `fs_changed` to the actors that watch the path (`watch` request),
//! - starts or upgrades an actor whose `actor/<id>/actor.wasm` (or folder
@@ -75,7 +75,7 @@ impl Pending {
}
async fn run(broker: Arc<Broker>, mut events: mpsc::UnboundedReceiver<notify::Result<notify::Event>>) {
let root = broker.world.root.clone();
let root = broker.world.actors_root();
let mut pending: HashMap<String, Pending> = HashMap::new();
// Actor checks and full passes. Dropping the set (when the host stops) aborts them.
let mut tasks = tokio::task::JoinSet::new();
@@ -122,11 +122,11 @@ async fn sleep_until(t: Option<Instant>) {
}
}
/// The world namespace path of a host path, unless it is outside the
/// world or a temporary file of the host.
/// The namespace path (`/actor/...`) of a host path, unless it is outside
/// `actor/` or a temporary file of the host.
fn namespace(root: &Path, path: &Path) -> Option<String> {
let rel = path.strip_prefix(root).ok()?;
let mut out = String::new();
let mut out = String::from("/actor");
for part in rel.components() {
let name = part.as_os_str().to_str()?;
if name.starts_with(TMP_PREFIX) {
@@ -135,7 +135,7 @@ fn namespace(root: &Path, path: &Path) -> Option<String> {
out.push('/');
out.push_str(name);
}
Some(if out.is_empty() { "/".to_string() } else { out })
Some(out)
}
/// One change: namespace path, op, and the new path of a rename.
@@ -219,7 +219,7 @@ mod tests {
#[test]
fn a_rename_is_one_change_and_temp_files_are_skipped() {
let root = Path::new("/w");
let root = Path::new("/w/actor");
let now = Instant::now();
let mut pending = HashMap::new();
let ev = |kind, paths: &[&str]| {
@@ -236,8 +236,7 @@ mod tests {
ev(name(RenameMode::Both), &["/w/actor/a/public/x", "/w/actor/a/public/y"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/actor/a/.sausage-tmp-host-1-actor.json"]),
ev(name(RenameMode::Both), &["/w/actor/a/.sausage-tmp-host-1-actor.json", "/w/actor/a/actor.json"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/outside/.lock"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/config/x"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/.lock"]),
] {
for (path, op, to) in changes(root, &e) {
add(&mut pending, path, op, to, now);
@@ -251,7 +250,6 @@ mod tests {
vec![
("/actor/a/actor.json".to_string(), "write", None),
("/actor/a/public/x".to_string(), "rename", Some("/actor/a/public/y".to_string())),
("/config/x".to_string(), "create", None),
]
);
}
+9 -19
View File
@@ -1,17 +1,15 @@
//! On-disk layout of a Sausage world. The folder is the source of truth:
//! `sausage run` boots every actor that it finds in `actor/`, and then
//! watches the folder (`watch.rs`). `/self` aliases the running actor's folder;
//! `/public` and `/data` remain aliases for its `public/` and `data/`.
//! watches the folder (`watch.rs`).
//!
//! ```text
//! <world>/ mounted at /, readable and writable by every actor
//! config/ global configuration by convention (no secrets)
//! <world>/
//! actor/<id>/
//! actor.wasm the current version
//! actor.json version, running, subscriptions, tapped, watches (broker)
//! source/ source of the current version
//! public/ published files, readable and writable by everyone
//! data/ state, readable and writable by everyone
//! public/ written by the actor, read-only for others
//! data/ state, writable only by the actor
//! stderr.log stderr of the actor and notes of the broker
//! versions/<n>/ every version that was installed: actor.wasm, source/, meta.json
//! ```
@@ -133,7 +131,6 @@ impl World {
for id in NATIVE {
world.ensure_actor_dirs(id)?;
}
std::fs::create_dir_all(world.root.join("config"))?;
Ok(world)
}
@@ -194,9 +191,8 @@ impl World {
/// Map a path in an actor's namespace to a host path.
///
/// - Absolute paths are relative to the world root.
/// - `/self/...` aliases `sender`'s folder; `/public/...` and `/data/...`
/// alias its corresponding children.
/// - `/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> {
let under = |base: PathBuf, rest: &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("/self") {
return under(self.actor_dir(sender), rest);
}
if let Some(rest) = prefixed("/public") {
return under(self.public_dir(sender), rest);
}
@@ -228,16 +221,13 @@ impl World {
if NATIVE.contains(&sender) && Path::new(path).is_absolute() {
return Ok(PathBuf::from(path));
}
if Path::new(path).is_absolute() {
return under(self.root.clone(), path);
}
bail!("cannot resolve path {path} for {sender}: use an absolute world path")
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.
pub fn global_path(&self, host: &Path) -> Option<String> {
let rel = host.strip_prefix(&self.root).ok()?;
Some(format!("/{}", rel.display()))
let rel = host.strip_prefix(self.actors_root()).ok()?;
Some(format!("/actor/{}", rel.display()))
}
}
+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();
+118 -97
View File
@@ -140,20 +140,20 @@ async fn exit_request_is_answered_and_signals_the_host() {
}
#[tokio::test]
async fn filesystem_is_writable_and_public_aliases_share_files() {
async fn filesystem_permissions_and_public_sharing() {
let (_, host, mut mailbox) = filesystem_world("filesystem-permissions").await;
for path in [
"/actor/reader/public/shared.txt",
"/actor/writer/public/shared.txt",
"/actor/reader/public/denied.txt",
"/actor/writer/public/denied.txt",
] {
let written = ask(
let denied = ask(
&host,
&mut mailbox,
"writer",
json!({"type": "write", "path": path, "text": "shared"}),
json!({"type": "write", "path": path, "text": "denied"}),
)
.await;
assert_eq!(written.body["type"], "ok", "write to {path} failed");
assert_eq!(denied.body["type"], "error", "write to {path} succeeded");
}
let written = ask(
@@ -176,7 +176,7 @@ async fn filesystem_is_writable_and_public_aliases_share_files() {
}
#[tokio::test]
async fn actors_can_read_and_write_other_actor_data() {
async fn actors_can_read_but_not_write_other_actor_data() {
let (_, host, mut mailbox) = filesystem_world("actor-data-sharing").await;
ask_type(
&host,
@@ -201,10 +201,10 @@ async fn actors_can_read_and_write_other_actor_data() {
&mut mailbox,
"reader",
json!({"type": "write", "path": "/actor/writer/data/x", "text": "changed"}),
"ok",
"error",
)
.await;
let changed = ask_type(
let unchanged = ask_type(
&host,
&mut mailbox,
"writer",
@@ -212,49 +212,7 @@ async fn actors_can_read_and_write_other_actor_data() {
"content",
)
.await;
assert_eq!(changed.body["text"], "changed");
host.stop().await;
}
#[tokio::test]
async fn world_root_is_shared_writable_and_survives_restart() {
let (world, host, mut mailbox) = filesystem_world("world-root").await;
assert!(world.root.join("config").is_dir());
ask_type(&host, &mut mailbox, "writer", json!({"type": "write", "path": "/self/data/x", "text": "self-alias"}), "ok").await;
assert_eq!(ask_type(&host, &mut mailbox, "writer", json!({"type": "read", "path": "/data/x"}), "content").await.body["text"], "self-alias");
let file = "/config/x";
ask_type(&host, &mut mailbox, "writer", json!({"type": "write", "path": file, "text": "first"}), "ok").await;
let read = json!({"type": "read", "path": file});
assert_eq!(ask_type(&host, &mut mailbox, "reader", read.clone(), "content").await.body["text"], "first");
ask_type(&host, &mut mailbox, "reader", json!({"type": "write", "path": file, "text": "second"}), "ok").await;
assert_eq!(std::fs::read_to_string(world.root.join("config/x")).unwrap(), "second");
for actor in ["writer", "reader"] {
assert_eq!(world.resolve(actor, file).unwrap(), world.root.join("config/x"));
}
assert_eq!(world.global_path(&world.root.join("config/x")).as_deref(), Some(file));
// Watches use the same world namespace, not only /actor paths.
let watching = request(&host, &mut mailbox, json!({"type": "watch", "path": world.root.join("config").display().to_string()})).await;
assert_eq!(watching.body, json!({"type": "watching", "path": "/config"}));
drain(&mut mailbox, Duration::from_millis(300)).await;
std::fs::write(world.root.join("config/event"), "watched").unwrap();
assert_eq!(fs_events(drain(&mut mailbox, Duration::from_millis(600)).await),
vec![json!({"type": "fs_changed", "path": "/config/event", "op": "create"})]);
request(&host, &mut mailbox, json!({"type": "unwatch", "path": world.root.join("config").display().to_string()})).await;
// A new top-level folder appears in already running instances, without a mount change.
std::fs::create_dir(world.root.join("docs")).unwrap();
std::fs::write(world.root.join("docs/new"), "host-created").unwrap();
let docs = json!({"type": "read", "path": "/docs/new"});
assert_eq!(ask_type(&host, &mut mailbox, "reader", docs, "content").await.body["text"], "host-created");
ask_type(&host, &mut mailbox, "reader", json!({"type": "write", "path": "/docs/new", "text": "actor-updated"}), "ok").await;
assert_eq!(std::fs::read_to_string(world.root.join("docs/new")).unwrap(), "actor-updated");
host.stop().await;
let (host, mut mailbox) = start(&world).await;
for actor in ["writer", "reader"] {
assert_eq!(ask_type(&host, &mut mailbox, actor, read.clone(), "content").await.body["text"], "second");
}
assert_eq!(unchanged.body["text"], "shared");
host.stop().await;
}
@@ -478,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");
@@ -548,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");
@@ -1229,7 +1250,7 @@ async fn a_watcher_overflow_sends_one_rescan_and_runs_a_full_pass() {
let report = host.broker.clone().overflowed().await;
assert!(report.iter().any(|r| r["type"] == "spawned" && r["actor"] == "planted"), "{report:?}");
let rescan = json!({"type": "fs_changed", "path": "/", "op": "rescan"});
let rescan = json!({"type": "fs_changed", "path": "/actor", "op": "rescan"});
assert_eq!(fs_events(drain(&mut mailbox, Duration::from_millis(600)).await), vec![rescan]);
assert!(bystander.try_recv().is_err(), "an actor without watches got the rescan");
host.stop().await;