Compare commits

...
Author SHA1 Message Date
Sausage Agent 34e2f35272 Add writable /self alias alongside /public and /data
ci / check (push) Successful in 15s
2026-10-22 13:35:13 +00:00
Sausage Agent f932ab9719 Mount the writable world root for every actor
ci / check (push) Successful in 3m9s
2026-10-22 13:29:38 +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
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
13 changed files with 356 additions and 69 deletions
+31 -11
View File
@@ -59,13 +59,27 @@ it exits. After 5 seconds, it kills the instance.
## Filesystem ## Filesystem
Each instance sees three folders: The world folder is the foundation of every instance's filesystem:
| Path | Access | Contents | | Path | Access | Contents |
| --- | --- | --- | | --- | --- | --- |
| `/data` | read-write | the actor's state | | `/` | read-write by every actor | the whole world folder |
| `/public` | read-write | the files that the actor publishes | | `/self` | read-write | alias for the actor's own folder (`public/`, `data/`, `source/`, etc.) |
| `/actor` | read-only | the `actor/` folder of the world | | `/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 |
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.
`/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json` `/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json`
(installed version, running, subscriptions, tapped, watches), `source/`, `public/`, `data/`, (installed version, running, subscriptions, tapped, watches), `source/`, `public/`, `data/`,
@@ -123,16 +137,21 @@ 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 |
| `tap`, `untap` | | receive (or not) a copy of every message the broker routes | | `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 | | `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 |
| `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) |
Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`. **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.
**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,
@@ -172,7 +191,7 @@ 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 - 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. mailbox drops them, and they never block the writer. Never answer them.
- If the OS drops events, every watcher gets one - If the OS drops events, every watcher gets one
`{"type": "fs_changed", "path": "/actor", "op": "rescan"}`: anything may `{"type": "fs_changed", "path": "/", "op": "rescan"}`: anything may
have changed. have changed.
- A watch is saved in `actor.json` (`"watches": [...]`) and survives - A watch is saved in `actor.json` (`"watches": [...]`) and survives
restarts and upgrades. `unwatch` with the same path ends it. restarts and upgrades. `unwatch` with the same path ends it.
@@ -208,7 +227,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
+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 same commands are `nix run`, `nix run -- init`, `nix run -- run` and
`nix run .#loop`. `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: In the console:
```text ```text
@@ -66,7 +78,7 @@ actor.wasm the current version
actor.json installed version, running, subscriptions actor.json installed version, running, subscriptions
source/ the source of the current version source/ the source of the current version
public/ files that the actor publishes 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 stderr.log what the actor and the broker write about it
versions/<n>/ every installed version: actor.wasm, source/, meta.json 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. 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. Read the SDK at `/actor/cargo/public/sdk/` before you write an actor.
You can change your own source at `/actor/agent/source`. 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.
+12 -9
View File
@@ -255,7 +255,9 @@ impl Broker {
} }
InstanceConfig { InstanceConfig {
mounts: vec![ mounts: vec![
Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: false }, 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: "/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 },
], ],
@@ -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}))
} }
@@ -444,7 +447,7 @@ impl Broker {
self.host_error("the file watcher dropped events; watchers get a rescan, actors a full pass"); self.host_error("the file watcher dropped events; watchers get a rescan, actors a full pass");
let watchers: Vec<String> = let watchers: Vec<String> =
self.metas.lock().unwrap().iter().filter(|(_, m)| !m.watches.is_empty()).map(|(id, _)| id.clone()).collect(); 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": "/actor", "op": "rescan"})); self.inject(&watchers, vfs::ID, json!({"type": "fs_changed", "path": "/", "op": "rescan"}));
let report = self.scan().await; let report = self.scan().await;
for change in &report { for change in &report {
eprintln!("sausage: {change}"); eprintln!("sausage: {change}");
@@ -598,9 +601,9 @@ impl Broker {
let on = ty == "watch"; let on = ty == "watch";
let path_s = b["path"].as_str().ok_or_else(|| anyhow!("missing \"path\""))?; 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 global = self.world.global_path(&self.world.resolve(&env.from, path_s)?);
let path = global.ok_or_else(|| anyhow!("{path_s} is not under /actor"))?; let path = global.ok_or_else(|| anyhow!("{path_s} is not under the world root"))?;
let path = match path.trim_end_matches('/') { let path = match path.trim_end_matches('/') {
"" => "/actor".to_string(), "" => "/".to_string(),
p => p.to_string(), p => p.to_string(),
}; };
self.update_meta(&env.from, |m| { self.update_meta(&env.from, |m| {
@@ -648,7 +651,7 @@ fn same_file(a: &Path, b: &Path) -> bool {
/// Whether the namespace path `path` is `prefix` or inside it. /// Whether the namespace path `path` is `prefix` or inside it.
fn under(path: &str, prefix: &str) -> bool { fn under(path: &str, prefix: &str) -> bool {
path.strip_prefix(prefix).is_some_and(|rest| rest.is_empty() || rest.starts_with('/')) prefix == "/" || path.strip_prefix(prefix).is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
} }
pub fn truncate(s: &str, max: usize) -> String { pub fn truncate(s: &str, max: usize) -> String {
+2 -1
View File
@@ -3,7 +3,8 @@
//! - `@<actor> <json>` sends a JSON body. `@<actor> <text>` sends //! - `@<actor> <json>` sends a JSON body. `@<actor> <text>` sends
//! `{"type": "text", "text": ...}`. //! `{"type": "text", "text": ...}`.
//! - A line without `@` goes to `agent` as 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; use std::sync::Arc;
+43 -7
View File
@@ -5,6 +5,7 @@ use anyhow::{Context, Result, bail};
use clap::{Parser, Subcommand}; use clap::{Parser, Subcommand};
use include_dir::{Dir, include_dir}; use include_dir::{Dir, include_dir};
use serde_json::json; use serde_json::json;
use tokio::signal::unix::Signal;
use sausage::world::{ActorMeta, World}; use sausage::world::{ActorMeta, World};
use sausage::{Host, cargo_actor, console}; use sausage::{Host, cargo_actor, console};
@@ -17,6 +18,11 @@ static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
#[derive(Parser)] #[derive(Parser)]
#[command(version, about)] #[command(version, about)]
struct Cli { 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(subcommand)]
command: Option<Cmd>, command: Option<Cmd>,
} }
@@ -91,13 +97,41 @@ async fn init(dir: PathBuf) -> Result<()> {
Ok(()) 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?; let host = Host::start(&dir).await?;
tokio::spawn(console::run_mailbox(host.mailbox(console::ID))); tokio::spawn(console::run_mailbox(host.mailbox(console::ID)));
let broker = host.broker.clone(); let broker = host.broker.clone();
let exit = tokio::select! { let exit = match signals {
_ = console::run_input(host.broker.clone()) => false, Some(signals) => {
_ = broker.exit_requested() => true, 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 …"); println!("Stopping actors …");
host.stop().await; host.stop().await;
@@ -113,16 +147,18 @@ async fn run(dir: PathBuf) -> Result<()> {
#[tokio::main] #[tokio::main]
async fn main() -> Result<()> { 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::Init { dir }) => init(dir).await,
Some(Cmd::Run { dir }) => run(dir).await, Some(Cmd::Run { dir }) => run(dir, headless).await,
None => { None => {
let dir = PathBuf::from("world"); let dir = PathBuf::from("world");
if !dir.join("actor").is_dir() { if !dir.join("actor").is_dir() {
println!("No world in ./world yet, creating it (same as `sausage init`)."); println!("No world in ./world yet, creating it (same as `sausage init`).");
init(dir.clone()).await?; init(dir.clone()).await?;
} }
run(dir).await run(dir, headless).await
} }
} }
} }
+11 -6
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;
self.start().await?; if self.running {
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version})) self.start().await?;
}
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.
+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)>> { fn get_directories(&mut self) -> wasmtime::Result<Vec<(Resource<types::Descriptor>, String)>> {
let dirs = preopens::Host::get_directories(&mut self.inner)?; let dirs = preopens::Host::get_directories(&mut self.inner)?;
for (fd, name) in &dirs { for (fd, name) in &dirs {
if name == "/actor" { if name == "/actor" || name == "/" {
self.state.dirs.insert(fd.rep(), "/actor".to_string()); self.state.dirs.insert(fd.rep(), name.clone());
} }
} }
Ok(dirs) Ok(dirs)
+10 -8
View File
@@ -5,7 +5,7 @@
//! Events are coalesced per path: once a path is quiet for `QUIET` (or has //! Events are coalesced per path: once a path is quiet for `QUIET` (or has
//! kept changing for `MAX_DELAY`), it becomes one change //! kept changing for `MAX_DELAY`), it becomes one change
//! `{path, op: write | create | remove | rename, to?}` with a namespace path //! `{path, op: write | create | remove | rename, to?}` with a namespace path
//! (`/actor/...`). Then the broker //! (relative to the world root, such as `/actor/...` or `/config/...`). Then the broker
//! //!
//! - sends `fs_changed` to the actors that watch the path (`watch` request), //! - sends `fs_changed` to the actors that watch the path (`watch` request),
//! - starts or upgrades an actor whose `actor/<id>/actor.wasm` (or folder //! - 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>>) { async fn run(broker: Arc<Broker>, mut events: mpsc::UnboundedReceiver<notify::Result<notify::Event>>) {
let root = broker.world.actors_root(); let root = broker.world.root.clone();
let mut pending: HashMap<String, Pending> = HashMap::new(); let mut pending: HashMap<String, Pending> = HashMap::new();
// Actor checks and full passes. Dropping the set (when the host stops) aborts them. // Actor checks and full passes. Dropping the set (when the host stops) aborts them.
let mut tasks = tokio::task::JoinSet::new(); let mut tasks = tokio::task::JoinSet::new();
@@ -122,11 +122,11 @@ async fn sleep_until(t: Option<Instant>) {
} }
} }
/// The namespace path (`/actor/...`) of a host path, unless it is outside /// The world namespace path of a host path, unless it is outside the
/// `actor/` or a temporary file of the host. /// world or a temporary file of the host.
fn namespace(root: &Path, path: &Path) -> Option<String> { fn namespace(root: &Path, path: &Path) -> Option<String> {
let rel = path.strip_prefix(root).ok()?; let rel = path.strip_prefix(root).ok()?;
let mut out = String::from("/actor"); let mut out = String::new();
for part in rel.components() { for part in rel.components() {
let name = part.as_os_str().to_str()?; let name = part.as_os_str().to_str()?;
if name.starts_with(TMP_PREFIX) { if name.starts_with(TMP_PREFIX) {
@@ -135,7 +135,7 @@ fn namespace(root: &Path, path: &Path) -> Option<String> {
out.push('/'); out.push('/');
out.push_str(name); out.push_str(name);
} }
Some(out) Some(if out.is_empty() { "/".to_string() } else { out })
} }
/// One change: namespace path, op, and the new path of a rename. /// One change: namespace path, op, and the new path of a rename.
@@ -219,7 +219,7 @@ mod tests {
#[test] #[test]
fn a_rename_is_one_change_and_temp_files_are_skipped() { fn a_rename_is_one_change_and_temp_files_are_skipped() {
let root = Path::new("/w/actor"); let root = Path::new("/w");
let now = Instant::now(); let now = Instant::now();
let mut pending = HashMap::new(); let mut pending = HashMap::new();
let ev = |kind, paths: &[&str]| { let ev = |kind, paths: &[&str]| {
@@ -236,7 +236,8 @@ mod tests {
ev(name(RenameMode::Both), &["/w/actor/a/public/x", "/w/actor/a/public/y"]), 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(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(name(RenameMode::Both), &["/w/actor/a/.sausage-tmp-host-1-actor.json", "/w/actor/a/actor.json"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/.lock"]), ev(EventKind::Create(notify::event::CreateKind::File), &["/outside/.lock"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/config/x"]),
] { ] {
for (path, op, to) in changes(root, &e) { for (path, op, to) in changes(root, &e) {
add(&mut pending, path, op, to, now); add(&mut pending, path, op, to, now);
@@ -250,6 +251,7 @@ mod tests {
vec![ vec![
("/actor/a/actor.json".to_string(), "write", None), ("/actor/a/actor.json".to_string(), "write", None),
("/actor/a/public/x".to_string(), "rename", Some("/actor/a/public/y".to_string())), ("/actor/a/public/x".to_string(), "rename", Some("/actor/a/public/y".to_string())),
("/config/x".to_string(), "create", None),
] ]
); );
} }
+19 -9
View File
@@ -1,15 +1,17 @@
//! On-disk layout of a Sausage world. The folder is the source of truth: //! 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 //! `sausage run` boots every actor that it finds in `actor/`, and then
//! watches the folder (`watch.rs`). //! watches the folder (`watch.rs`). `/self` aliases the running actor's folder;
//! `/public` and `/data` remain aliases for its `public/` and `data/`.
//! //!
//! ```text //! ```text
//! <world>/ //! <world>/ mounted at /, readable and writable by every actor
//! config/ global configuration by convention (no secrets)
//! 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)
//! source/ source of the current version //! source/ source of the current version
//! public/ written by the actor, read-only for others //! public/ published files, readable and writable by everyone
//! data/ state, writable only by the actor //! data/ state, readable and writable by everyone
//! stderr.log stderr of the actor and notes of the broker //! stderr.log stderr of the actor and notes of the broker
//! versions/<n>/ every version that was installed: actor.wasm, source/, meta.json //! versions/<n>/ every version that was installed: actor.wasm, source/, meta.json
//! ``` //! ```
@@ -131,6 +133,7 @@ impl World {
for id in NATIVE { for id in NATIVE {
world.ensure_actor_dirs(id)?; world.ensure_actor_dirs(id)?;
} }
std::fs::create_dir_all(world.root.join("config"))?;
Ok(world) Ok(world)
} }
@@ -191,8 +194,9 @@ 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. /// - Absolute paths are relative to the world root.
/// - `/public/...` and `/data/...` belong to `sender`. /// - `/self/...` aliases `sender`'s folder; `/public/...` and `/data/...`
/// alias its corresponding children.
/// - 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> {
let under = |base: PathBuf, rest: &str| -> Result<PathBuf> { let under = |base: PathBuf, rest: &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("/self") {
return under(self.actor_dir(sender), 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,13 +228,16 @@ 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/...") if Path::new(path).is_absolute() {
return under(self.root.clone(), path);
}
bail!("cannot resolve path {path} for {sender}: use an absolute world path")
} }
/// 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.
pub fn global_path(&self, host: &Path) -> Option<String> { pub fn global_path(&self, host: &Path) -> Option<String> {
let rel = host.strip_prefix(self.actors_root()).ok()?; let rel = host.strip_prefix(&self.root).ok()?;
Some(format!("/actor/{}", rel.display())) Some(format!("/{}", rel.display()))
} }
} }
+44
View File
@@ -39,3 +39,47 @@ fn bare_command_without_a_world_starts_init() {
// It asks before it creates anything. // It asks before it creates anything.
assert!(!dir.join("world").exists()); 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:?}");
}
+97 -11
View File
@@ -140,20 +140,20 @@ async fn exit_request_is_answered_and_signals_the_host() {
} }
#[tokio::test] #[tokio::test]
async fn filesystem_permissions_and_public_sharing() { async fn filesystem_is_writable_and_public_aliases_share_files() {
let (_, host, mut mailbox) = filesystem_world("filesystem-permissions").await; let (_, host, mut mailbox) = filesystem_world("filesystem-permissions").await;
for path in [ for path in [
"/actor/reader/public/denied.txt", "/actor/reader/public/shared.txt",
"/actor/writer/public/denied.txt", "/actor/writer/public/shared.txt",
] { ] {
let denied = ask( let written = ask(
&host, &host,
&mut mailbox, &mut mailbox,
"writer", "writer",
json!({"type": "write", "path": path, "text": "denied"}), json!({"type": "write", "path": path, "text": "shared"}),
) )
.await; .await;
assert_eq!(denied.body["type"], "error", "write to {path} succeeded"); assert_eq!(written.body["type"], "ok", "write to {path} failed");
} }
let written = ask( let written = ask(
@@ -176,7 +176,7 @@ async fn filesystem_permissions_and_public_sharing() {
} }
#[tokio::test] #[tokio::test]
async fn actors_can_read_but_not_write_other_actor_data() { async fn actors_can_read_and_write_other_actor_data() {
let (_, host, mut mailbox) = filesystem_world("actor-data-sharing").await; let (_, host, mut mailbox) = filesystem_world("actor-data-sharing").await;
ask_type( ask_type(
&host, &host,
@@ -201,10 +201,10 @@ async fn actors_can_read_but_not_write_other_actor_data() {
&mut mailbox, &mut mailbox,
"reader", "reader",
json!({"type": "write", "path": "/actor/writer/data/x", "text": "changed"}), json!({"type": "write", "path": "/actor/writer/data/x", "text": "changed"}),
"error", "ok",
) )
.await; .await;
let unchanged = ask_type( let changed = ask_type(
&host, &host,
&mut mailbox, &mut mailbox,
"writer", "writer",
@@ -212,7 +212,49 @@ async fn actors_can_read_but_not_write_other_actor_data() {
"content", "content",
) )
.await; .await;
assert_eq!(unchanged.body["text"], "shared"); 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");
}
host.stop().await; host.stop().await;
} }
@@ -506,6 +548,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");
@@ -1143,7 +1229,7 @@ async fn a_watcher_overflow_sends_one_rescan_and_runs_a_full_pass() {
let report = host.broker.clone().overflowed().await; let report = host.broker.clone().overflowed().await;
assert!(report.iter().any(|r| r["type"] == "spawned" && r["actor"] == "planted"), "{report:?}"); assert!(report.iter().any(|r| r["type"] == "spawned" && r["actor"] == "planted"), "{report:?}");
let rescan = json!({"type": "fs_changed", "path": "/actor", "op": "rescan"}); let rescan = json!({"type": "fs_changed", "path": "/", "op": "rescan"});
assert_eq!(fs_events(drain(&mut mailbox, Duration::from_millis(600)).await), vec![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"); assert!(bystander.try_recv().is_err(), "an actor without watches got the rescan");
host.stop().await; host.stop().await;
+70 -4
View File
@@ -21,9 +21,11 @@
targets = [ "wasm32-wasip2" ]; targets = [ "wasm32-wasip2" ];
extensions = [ "rust-src" ]; extensions = [ "rust-src" ];
}; };
# Crane builds the dependencies once (`cargoArtifacts`, keyed only on # The runtime cargo actor needs Cargo, rustc (with wasm32-wasip2), and
# Cargo.toml/Cargo.lock), so a change to our own crates rebuilds # the native linker, even when the host itself is started with a bare
# just them. # 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; craneLib = (crane.mkLib pkgs).overrideToolchain rust;
common = { common = {
pname = "sausage"; pname = "sausage";
@@ -42,12 +44,13 @@
}); });
sausage = craneLib.buildPackage (common // { sausage = craneLib.buildPackage (common // {
inherit cargoArtifacts; inherit cargoArtifacts;
doNotRemoveReferencesToRustToolchain = true;
cargoExtraArgs = "--locked -p sausage"; cargoExtraArgs = "--locked -p sausage";
doCheck = false; doCheck = false;
nativeBuildInputs = [ pkgs.makeWrapper ]; nativeBuildInputs = [ pkgs.makeWrapper ];
# The `cargo` actor runs cargo at runtime. Build scripts need a C linker. # The `cargo` actor runs cargo at runtime. Build scripts need a C linker.
postInstall = '' postInstall = ''
wrapProgram $out/bin/sausage --prefix PATH : ${pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ]} wrapProgram $out/bin/sausage --prefix PATH : ${runtimePath}
''; '';
meta.mainProgram = "sausage"; meta.mainProgram = "sausage";
}); });
@@ -91,6 +94,69 @@
rm -f cli/tests/actors/toolbox/Cargo.lock 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.default = { type = "app"; program = "${sausage}/bin/sausage"; };
apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; }; apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };