Compare commits

...
Author SHA1 Message Date
Sausage Agent 48b4287287 host: add a shared writable /config world mount
ci / check (push) Successful in 6m41s
2026-10-22 13:08:57 +00:00
surma 3d3f6f35fb Merge pull request 'broker: upgrade and rollback keep a stopped actor stopped' (#3) from upgrade-keeps-stopped into main
ci / check (push) Successful in 13s
Reviewed-on: #3
2026-10-22 10:50:15 +01:00
surma f1af07506f broker: upgrade and rollback keep a stopped actor stopped
ci / check (push) Successful in 8m42s
ci / check (pull_request) Successful in 4m38s
A stopped actor gets the new version installed, keeps its mailbox and does
not run main until the next start. Running actors still restart on the new
version. The upgraded and rolled_back answers say whether the actor runs.
2026-10-22 10:11:36 +01:00
5 changed files with 117 additions and 17 deletions
+10 -5
View File
@@ -59,13 +59,17 @@ it exits. After 5 seconds, it kills the instance.
## Filesystem ## Filesystem
Each instance sees three folders: Each instance sees four folders:
| Path | Access | Contents | | Path | Access | Contents |
| --- | --- | --- | | --- | --- | --- |
| `/data` | read-write by its actor; readable by every actor | 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 | | `/public` | read-write | the files that the actor publishes |
| `/actor` | read-only | the `actor/` folder of the world | | `/actor` | read-only | the `actor/` folder of the world |
| `/config` | read-write by every actor | the world's shared `config/` folder; persists across restarts |
`/config` is the same world folder for every instance, including candidates;
writers coordinate by convention, not permissions. Do not put secrets there.
An actor can write only its own `/data`, but can read every actor's `/data` through `/actor/<id>/data/`. `/data` is not private: any unprotected secret file there is readable by every actor. 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.
@@ -125,8 +129,8 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
| `type` | Fields | Effect | | `type` | Fields | Effect |
| --- | --- | --- | | --- | --- | --- |
| `spawn` | `actor`, `wasm`, `source`?, `built_by`? | test, then start a new actor | | `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`? | 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 | | `rollback` | `actor`, `version` | go back to an older version; like `upgrade`, a stopped actor stays stopped. The answer has `running` |
| `stop`, `start` | `actor` | stop or start an actor | | `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` | | `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 | | `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
@@ -136,7 +140,7 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
**Headless.** `sausage run --headless [dir]` (or `sausage --headless`, or `SAUSAGE_HEADLESS=1`) turns the console input off: the host does not read stdin, so end of input (`/dev/null`) does not stop the world. `nix run .#loop -- --headless <world>` passes it through the loop. It runs until SIGTERM or SIGINT (stops every actor, exit code 0) or a broker `exit` request (exit code 75, as above). Without the flag, EOF on stdin or `/quit` ends the host. **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/...`. Paths are `/actor/...`, `/config/...`, or the sender's `/public/...` or `/data/...`.
**Taps.** After `tap`, the sender gets a copy of every message that the **Taps.** After `tap`, the sender gets a copy of every message that the
broker routes (direct messages, replies, and publishes, between any actors, broker routes (direct messages, replies, and publishes, between any actors,
@@ -212,7 +216,8 @@ instance. A version without test files passes.
**Swap.** The broker holds the mailbox, stops the old instance, makes the **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 top-level `actor.wasm` and `source/` a copy of the new version, and starts
it. A rollback does the same with the older version. it. A stopped actor is not started: it keeps its mailbox and runs the new
version at the next `start`. A rollback does the same with the older version.
**Crashes.** The broker restarts a crashed actor after 0.5 s, then doubles **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 the delay up to 30 s. A new version that crashes 3 times within 30 seconds
+7 -4
View File
@@ -70,6 +70,7 @@ pub struct Broker {
impl Broker { impl Broker {
pub fn new(world: World, runtime: Arc<Runtime>) -> Result<Arc<Self>> { pub fn new(world: World, runtime: Arc<Runtime>) -> Result<Arc<Self>> {
std::fs::create_dir_all(world.config_dir())?;
let metas = world.actor_ids()?.into_iter().map(|id| (id.clone(), world.read_meta(&id))).collect(); let metas = world.actor_ids()?.into_iter().map(|id| (id.clone(), world.read_meta(&id))).collect();
Ok(Arc::new_cyclic(|me| Self { Ok(Arc::new_cyclic(|me| Self {
world, world,
@@ -258,6 +259,7 @@ impl Broker {
Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: false }, Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: false },
Mount { guest: "/public".into(), host: own.join("public"), writable: true }, Mount { guest: "/public".into(), host: own.join("public"), writable: true },
Mount { guest: "/data".into(), host: own.join("data"), writable: true }, Mount { guest: "/data".into(), host: own.join("data"), writable: true },
Mount { guest: "/config".into(), host: self.world.config_dir(), writable: true },
], ],
network: scratch.is_none(), network: scratch.is_none(),
stderr_log: own.join("stderr.log"), stderr_log: own.join("stderr.log"),
@@ -331,12 +333,13 @@ impl Broker {
Ok((component, tests)) Ok((component, tests))
} }
/// Make `version` the current `actor.wasm`, `source/`, and `actor.json`. /// Make `version` the current `actor.wasm`, `source/`, and `actor.json`,
pub fn install(&self, id: &str, version: u64) -> Result<()> { /// and record whether the actor runs.
pub fn install(&self, id: &str, version: u64, running: bool) -> Result<()> {
self.world.install_files(id, version)?; self.world.install_files(id, version)?;
self.update_meta(id, |m| { self.update_meta(id, |m| {
m.version = Some(version); m.version = Some(version);
m.running = true; m.running = running;
}) })
} }
@@ -360,7 +363,7 @@ impl Broker {
async fn create_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> { 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).await?;
self.install(id, meta.version)?; self.install(id, meta.version, true)?;
self.start_supervisor(id, meta.version, true); self.start_supervisor(id, meta.version, true);
Ok(json!({"type": "spawned", "actor": id, "version": meta.version, "tests": tests})) Ok(json!({"type": "spawned", "actor": id, "version": meta.version, "tests": tests}))
} }
+10 -5
View File
@@ -220,17 +220,16 @@ impl Supervisor {
/// Make `version` current. It is compiled first (unless `component` is /// Make `version` current. It is compiled first (unless `component` is
/// given): if that fails, nothing changes. `data/` and `public/` stay as /// given): if that fails, nothing changes. `data/` and `public/` stay as
/// they are. /// they are. A stopped actor stays stopped.
async fn install(&mut self, version: u64, component: Option<Component>) -> Result<()> { async fn install(&mut self, version: u64, component: Option<Component>) -> Result<()> {
let component = match component { let component = match component {
Some(c) => c, Some(c) => c,
None => self.broker.load_version(&self.id, version)?, None => self.broker.load_version(&self.id, version)?,
}; };
self.stop().await; self.stop().await;
self.broker.install(&self.id, version)?; self.broker.install(&self.id, version, self.running)?;
self.version = version; self.version = version;
self.component = Some(component); self.component = Some(component);
self.running = true;
self.restarts = 0; self.restarts = 0;
self.restart_at = None; self.restart_at = None;
Ok(()) Ok(())
@@ -239,20 +238,26 @@ impl Supervisor {
async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result<Value> { async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result<Value> {
let previous = self.version; let previous = self.version;
self.install(version, Some(component)).await?; self.install(version, Some(component)).await?;
if !self.running {
// Installed, not started: main runs and mail is read at the next `start`.
return Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous, "running": false}));
}
self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester }); self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester });
if let Err(e) = self.start().await { if let Err(e) = self.start().await {
self.rollback(previous).await?; self.rollback(previous).await?;
bail!("version {version} did not start ({e:#}); rolled back to {previous}"); bail!("version {version} did not start ({e:#}); rolled back to {previous}");
} }
Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous})) Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous, "running": true}))
} }
async fn rollback(&mut self, version: u64) -> Result<Value> { async fn rollback(&mut self, version: u64) -> Result<Value> {
let from = self.version; let from = self.version;
self.install(version, None).await?; self.install(version, None).await?;
self.probation = None; self.probation = None;
if self.running {
self.start().await?; self.start().await?;
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version})) }
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version, "running": self.running}))
} }
/// Handle a control request. Returns `true` when the supervisor ends. /// Handle a control request. Returns `true` when the supervisor ends.
+9 -2
View File
@@ -4,6 +4,7 @@
//! //!
//! ```text //! ```text
//! <world>/ //! <world>/
//! config/ shared configuration, writable by every actor
//! actor/<id>/ //! actor/<id>/
//! actor.wasm the current version //! actor.wasm the current version
//! actor.json version, running, subscriptions, tapped, watches (broker) //! actor.json version, running, subscriptions, tapped, watches (broker)
@@ -89,6 +90,9 @@ impl World {
Self { root: root.into() } Self { root: root.into() }
} }
pub fn config_dir(&self) -> PathBuf {
self.root.join("config")
}
pub fn actors_root(&self) -> PathBuf { pub fn actors_root(&self) -> PathBuf {
self.root.join("actor") self.root.join("actor")
} }
@@ -191,7 +195,7 @@ impl World {
/// Map a path in an actor's namespace to a host path. /// Map a path in an actor's namespace to a host path.
/// ///
/// - `/actor/<id>/...` is the global filesystem. /// - `/actor/<id>/...` and `/config/...` are world paths.
/// - `/public/...` and `/data/...` belong to `sender`. /// - `/public/...` and `/data/...` belong to `sender`.
/// - Native senders (the console) can also give absolute host paths. /// - Native senders (the console) can also give absolute host paths.
pub fn resolve(&self, sender: &str, path: &str) -> Result<PathBuf> { pub fn resolve(&self, sender: &str, path: &str) -> Result<PathBuf> {
@@ -212,6 +216,9 @@ impl World {
if let Some(rest) = prefixed("/actor") { if let Some(rest) = prefixed("/actor") {
return under(self.actors_root(), rest); return under(self.actors_root(), rest);
} }
if let Some(rest) = prefixed("/config") {
return under(self.config_dir(), rest);
}
if let Some(rest) = prefixed("/public") { if let Some(rest) = prefixed("/public") {
return under(self.public_dir(sender), rest); return under(self.public_dir(sender), rest);
} }
@@ -221,7 +228,7 @@ impl World {
if NATIVE.contains(&sender) && Path::new(path).is_absolute() { if NATIVE.contains(&sender) && Path::new(path).is_absolute() {
return Ok(PathBuf::from(path)); return Ok(PathBuf::from(path));
} }
bail!("cannot resolve path {path} for {sender}: use /actor/<id>/..., /public/... or /data/...") bail!("cannot resolve path {path} for {sender}: use /actor/<id>/..., /config/..., /public/... or /data/...")
} }
/// The path of a host file in the global filesystem, if it has one. /// The path of a host file in the global filesystem, if it has one.
+80
View File
@@ -175,6 +175,42 @@ async fn filesystem_permissions_and_public_sharing() {
host.stop().await; host.stop().await;
} }
#[tokio::test]
async fn config_is_shared_writable_and_survives_host_restart() {
let (world, host, mut mailbox) = filesystem_world("world-config").await;
let path = "/config/global.json";
let first = r#"{"model":"first"}"#;
let second = r#"{"model":"second"}"#;
ask_type(
&host, &mut mailbox, "writer",
json!({"type": "write", "path": path, "text": first}), "ok",
).await;
let read = ask_type(
&host, &mut mailbox, "reader",
json!({"type": "read", "path": path}), "content",
).await;
assert_eq!(read.body["text"], first);
// The second actor can replace the same file too.
ask_type(
&host, &mut mailbox, "reader",
json!({"type": "write", "path": path, "text": second}), "ok",
).await;
assert_eq!(std::fs::read_to_string(world.config_dir().join("global.json")).unwrap(), second);
assert_eq!(world.resolve("writer", path).unwrap(), world.config_dir().join("global.json"));
assert_eq!(world.resolve("console", path).unwrap(), world.config_dir().join("global.json"));
host.stop().await;
let (host, mut mailbox) = start(&world).await;
for actor in ["writer", "reader"] {
let read = ask_type(
&host, &mut mailbox, actor,
json!({"type": "read", "path": path}), "content",
).await;
assert_eq!(read.body["text"], second);
}
host.stop().await;
}
#[tokio::test] #[tokio::test]
async fn actors_can_read_but_not_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; let (_, host, mut mailbox) = filesystem_world("actor-data-sharing").await;
@@ -506,6 +542,50 @@ async fn rolls_back_a_crash_loop_to_the_previous_version_without_restoring_data(
host.stop().await; 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] #[tokio::test]
async fn explicit_rollback_keeps_current_data() { async fn explicit_rollback_keeps_current_data() {
let world = create_world("explicit-rollback"); let world = create_world("explicit-rollback");