From f9260faa5b2a3971f79c55c5c7a30ec3d748c18f Mon Sep 17 00:00:00 2001 From: Surma Date: Thu, 15 Oct 2026 13:44:06 +0100 Subject: [PATCH] Host: one place for the instance config; message ids are strings --- PROTOCOL.md | 2 +- cli/src/broker.rs | 36 ++++++++++++++++++++++-------------- cli/src/candidate.rs | 18 ++---------------- cli/src/supervisor.rs | 13 ++----------- 4 files changed, 27 insertions(+), 42 deletions(-) diff --git a/PROTOCOL.md b/PROTOCOL.md index bb2ae53..67bd63b 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -37,7 +37,7 @@ A line on **stdout**: Text for people is `{"type": "text", "text": "..."}`. - `to` is the recipient. Without `to`, the broker publishes the message to every subscriber of the sender. -- `id` is optional. If the actor sets it, the message ID is `"/"`. +- `id` is an optional string. If the actor sets it, the message ID is `"/"`. Otherwise, the broker assigns `m-`. The broker always sets `from`. - `reply_to` marks a reply: its value is the ID of the message it answers. There is no request/response. A reply is a new message. diff --git a/cli/src/broker.rs b/cli/src/broker.rs index fd0ee30..61e6b50 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -19,7 +19,7 @@ use wasmtime::component::Component; use crate::candidate; use crate::fs::Mount; -use crate::runtime::Runtime; +use crate::runtime::{InstanceConfig, Runtime}; use crate::supervisor::{Control, Reply, Supervisor}; use crate::world::{self, ActorMeta, NATIVE, VersionMeta, World}; @@ -139,14 +139,10 @@ impl Broker { self.log(from, &format!("message without \"body\", ignored: {}", truncate(line, 500))); return; }; - let own_id = match obj.get("id") { - Some(Value::String(s)) => Some(s.clone()), - Some(Value::Number(n)) => Some(n.to_string()), - _ => None, - }; + let own_id = obj.get("id").and_then(Value::as_str); let to = obj.get("to").and_then(Value::as_str); let reply_to = obj.get("reply_to").and_then(Value::as_str); - self.send(from, to, own_id.as_deref(), reply_to, body); + self.send(from, to, own_id, reply_to, body); } fn deliver(&self, env: Envelope) { @@ -178,13 +174,25 @@ impl Broker { self.op_locks.lock().unwrap().entry(id.to_string()).or_default().clone() } - /// Mounts of a running actor. `/actor` hides every `data/` folder. - pub fn mounts(&self, id: &str) -> Vec { - vec![ - Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: false }, - Mount { guest: "/public".into(), host: self.world.public_dir(id), writable: true }, - Mount { guest: "/data".into(), host: self.world.data_dir(id), writable: true }, - ] + /// What an instance of `id` gets (see PROTOCOL.md). A candidate runs in + /// `scratch`, which holds a copy of the actor's `data/` and `public/`, + /// and has no HTTP. + pub fn instance_config(&self, id: &str, version: u64, scratch: Option<&Path>) -> InstanceConfig { + let own = scratch.map_or_else(|| self.world.actor_dir(id), Path::to_path_buf); + let mut env = vec![("SAUSAGE_ACTOR".into(), id.to_string()), ("SAUSAGE_VERSION".into(), version.to_string())]; + if scratch.is_some() { + env.push(("SAUSAGE_CANDIDATE".into(), "1".into())); + } + InstanceConfig { + mounts: vec![ + 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 }, + ], + allow_http: scratch.is_none(), + stderr_log: own.join("stderr.log"), + env, + } } // --------------------------------------------------------------------- diff --git a/cli/src/candidate.rs b/cli/src/candidate.rs index 1a5cd08..34e5b6d 100644 --- a/cli/src/candidate.rs +++ b/cli/src/candidate.rs @@ -29,8 +29,7 @@ use serde_json::{Value, json}; use wasmtime::component::Component; use crate::broker::{Broker, SHUTDOWN_DEADLINE, shutdown_line, truncate}; -use crate::fs::Mount; -use crate::runtime::{Instance, InstanceConfig}; +use crate::runtime::Instance; use crate::world; const EXPECT_TIMEOUT: Duration = Duration::from_secs(10); @@ -104,20 +103,7 @@ impl Candidate<'_> { world::copy_tree(&from, &dir.join(name))?; } } - let cfg = InstanceConfig { - mounts: vec![ - Mount { guest: "/actor".into(), host: self.broker.world.actors_root(), writable: false }, - Mount { guest: "/public".into(), host: dir.join("public"), writable: true }, - Mount { guest: "/data".into(), host: dir.join("data"), writable: true }, - ], - allow_http: false, - stderr_log: dir.join("stderr.log"), - env: vec![ - ("SAUSAGE_ACTOR".into(), self.id.to_string()), - ("SAUSAGE_VERSION".into(), self.version.to_string()), - ("SAUSAGE_CANDIDATE".into(), "1".into()), - ], - }; + let cfg = self.broker.instance_config(self.id, self.version, Some(&dir)); Ok((self.broker.runtime.start(self.component, cfg)?, scratch)) } diff --git a/cli/src/supervisor.rs b/cli/src/supervisor.rs index d05a628..ed76e99 100644 --- a/cli/src/supervisor.rs +++ b/cli/src/supervisor.rs @@ -12,7 +12,7 @@ use tokio::sync::{mpsc, oneshot}; use wasmtime::component::Component; use crate::broker::{Broker, Envelope, SHUTDOWN_DEADLINE, shutdown_line}; -use crate::runtime::{Exit, Instance, InstanceConfig}; +use crate::runtime::{Exit, Instance}; /// A new version that crashes this often within `CRASH_WINDOW` after an /// upgrade is rolled back automatically. @@ -178,16 +178,7 @@ impl Supervisor { if self.component.is_none() { self.component = Some(self.broker.load_version(&self.id, self.version)?); } - let world = &self.broker.world; - let cfg = InstanceConfig { - mounts: self.broker.mounts(&self.id), - allow_http: true, - stderr_log: world.actor_dir(&self.id).join("stderr.log"), - env: vec![ - ("SAUSAGE_ACTOR".into(), self.id.clone()), - ("SAUSAGE_VERSION".into(), self.version.to_string()), - ], - }; + let cfg = self.broker.instance_config(&self.id, self.version, None); self.instance = Some(self.broker.runtime.start(self.component.as_ref().unwrap(), cfg)?); Ok(()) }