Broker exit request and nix run .#loop for self-updating hosts

- broker: `{"type":"exit"}` answers `exiting`; `sausage run` then stops all
  actors and exits with code 75.
- flake: apps `loop` / `loop-internal` run `nix run --refresh git+ssh://.../surma/sausage -- run <world>`
  in a loop; exit code 0 (/quit) ends the loop, anything else restarts.
- PROTOCOL.md, README.md, test in cli/tests/host.rs.

Not compiled: written by the Sausage agent, which cannot build the native host.
This commit is contained in:
Sausage Agent
2026-10-16 00:37:25 +00:00
parent 567646ba87
commit 3435b556fb
6 changed files with 75 additions and 4 deletions
+18 -2
View File
@@ -1,6 +1,6 @@
//! The broker: message routing, mailboxes, subscriptions, the version store,
//! and the requests to `broker` (spawn, upgrade, rollback, stop, start,
//! subscribe, unsubscribe, scan). `supervisor.rs` runs each WASM actor.
//! subscribe, unsubscribe, scan, exit). `supervisor.rs` runs each WASM actor.
//!
//! The broker accepts only `.wasm` files. It does not know any source language.
//! The world folder is the source of truth: `scan` boots every
@@ -58,6 +58,8 @@ pub struct Broker {
metas: Mutex<BTreeMap<String, ActorMeta>>,
op_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
seq: AtomicU64,
/// Set by the `exit` request: the host process should stop.
exit: tokio::sync::Notify,
}
impl Broker {
@@ -71,9 +73,17 @@ impl Broker {
metas: Mutex::new(metas),
op_locks: Mutex::new(HashMap::new()),
seq: AtomicU64::new(1),
exit: tokio::sync::Notify::new(),
}))
}
/// Wait until an actor sends `{"type": "exit"}` to `broker`. The `sausage`
/// binary then stops all actors and exits, so that a wrapper such as
/// `nix run .#loop` can start a fresh (updated) host.
pub async fn exit_requested(&self) {
self.exit.notified().await;
}
pub fn meta(&self, id: &str) -> Option<ActorMeta> {
self.metas.lock().unwrap().get(id).cloned()
}
@@ -427,7 +437,13 @@ impl Broker {
Ok(json!({"type": format!("{ty}d"), "actor": publisher}))
}
"scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})),
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, scan"),
"exit" => {
self.log("broker", &format!("exit requested by {}", env.from));
// notify_one keeps the permit if nobody waits yet.
self.exit.notify_one();
Ok(json!({"type": "exiting"}))
}
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, scan, exit"),
}
}
+12 -1
View File
@@ -92,9 +92,20 @@ async fn init(dir: PathBuf) -> Result<()> {
async fn run(dir: PathBuf) -> Result<()> {
let host = Host::start(&dir).await?;
tokio::spawn(console::run_mailbox(host.mailbox(console::ID)));
console::run_input(host.broker.clone()).await;
let broker = host.broker.clone();
let exit = tokio::select! {
_ = console::run_input(host.broker.clone()) => false,
_ = broker.exit_requested() => true,
};
println!("Stopping actors …");
host.stop().await;
if exit {
// The console still blocks on a stdin read that cannot be cancelled,
// so the runtime would not shut down on its own. Exit code 75 tells
// `nix run .#loop` to start the host again (0 = /quit = stop the loop).
println!("Exiting, as an actor asked the broker to.");
std::process::exit(75);
}
Ok(())
}
+12
View File
@@ -127,6 +127,18 @@ async fn broker_does_not_answer_messages_with_reply_to() {
host.stop().await;
}
#[tokio::test]
async fn exit_request_is_answered_and_signals_the_host() {
let world = create_world("broker-exit");
let (host, mut mailbox) = start(&world).await;
let exiting = request(&host, &mut mailbox, json!({"type": "exit"})).await;
assert_eq!(exiting.body["type"], "exiting");
tokio::time::timeout(Duration::from_secs(1), host.broker.exit_requested())
.await
.expect("exit request did not reach the host");
host.stop().await;
}
#[tokio::test]
async fn filesystem_permissions_and_public_sharing() {
let (_, host, mut mailbox) = filesystem_world("filesystem-permissions").await;