Host: one place for the instance config; message ids are strings

This commit is contained in:
2026-10-15 13:44:06 +01:00
parent 11144b947e
commit f9260faa5b
4 changed files with 27 additions and 42 deletions
+1 -1
View File
@@ -37,7 +37,7 @@ A line on **stdout**:
Text for people is `{"type": "text", "text": "..."}`. Text for people is `{"type": "text", "text": "..."}`.
- `to` is the recipient. Without `to`, the broker publishes the message to - `to` is the recipient. Without `to`, the broker publishes the message to
every subscriber of the sender. every subscriber of the sender.
- `id` is optional. If the actor sets it, the message ID is `"<sender>/<id>"`. - `id` is an optional string. If the actor sets it, the message ID is `"<sender>/<id>"`.
Otherwise, the broker assigns `m-<n>`. The broker always sets `from`. Otherwise, the broker assigns `m-<n>`. The broker always sets `from`.
- `reply_to` marks a reply: its value is the ID of the message it answers. - `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. There is no request/response. A reply is a new message.
+22 -14
View File
@@ -19,7 +19,7 @@ use wasmtime::component::Component;
use crate::candidate; use crate::candidate;
use crate::fs::Mount; use crate::fs::Mount;
use crate::runtime::Runtime; use crate::runtime::{InstanceConfig, Runtime};
use crate::supervisor::{Control, Reply, Supervisor}; use crate::supervisor::{Control, Reply, Supervisor};
use crate::world::{self, ActorMeta, NATIVE, VersionMeta, World}; 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))); self.log(from, &format!("message without \"body\", ignored: {}", truncate(line, 500)));
return; return;
}; };
let own_id = match obj.get("id") { let own_id = obj.get("id").and_then(Value::as_str);
Some(Value::String(s)) => Some(s.clone()),
Some(Value::Number(n)) => Some(n.to_string()),
_ => None,
};
let to = obj.get("to").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); 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) { fn deliver(&self, env: Envelope) {
@@ -178,13 +174,25 @@ impl Broker {
self.op_locks.lock().unwrap().entry(id.to_string()).or_default().clone() self.op_locks.lock().unwrap().entry(id.to_string()).or_default().clone()
} }
/// Mounts of a running actor. `/actor` hides every `data/` folder. /// What an instance of `id` gets (see PROTOCOL.md). A candidate runs in
pub fn mounts(&self, id: &str) -> Vec<Mount> { /// `scratch`, which holds a copy of the actor's `data/` and `public/`,
vec![ /// and has no HTTP.
Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: false }, pub fn instance_config(&self, id: &str, version: u64, scratch: Option<&Path>) -> InstanceConfig {
Mount { guest: "/public".into(), host: self.world.public_dir(id), writable: true }, let own = scratch.map_or_else(|| self.world.actor_dir(id), Path::to_path_buf);
Mount { guest: "/data".into(), host: self.world.data_dir(id), writable: true }, 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,
}
} }
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
+2 -16
View File
@@ -29,8 +29,7 @@ use serde_json::{Value, json};
use wasmtime::component::Component; use wasmtime::component::Component;
use crate::broker::{Broker, SHUTDOWN_DEADLINE, shutdown_line, truncate}; use crate::broker::{Broker, SHUTDOWN_DEADLINE, shutdown_line, truncate};
use crate::fs::Mount; use crate::runtime::Instance;
use crate::runtime::{Instance, InstanceConfig};
use crate::world; use crate::world;
const EXPECT_TIMEOUT: Duration = Duration::from_secs(10); const EXPECT_TIMEOUT: Duration = Duration::from_secs(10);
@@ -104,20 +103,7 @@ impl Candidate<'_> {
world::copy_tree(&from, &dir.join(name))?; world::copy_tree(&from, &dir.join(name))?;
} }
} }
let cfg = InstanceConfig { let cfg = self.broker.instance_config(self.id, self.version, Some(&dir));
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()),
],
};
Ok((self.broker.runtime.start(self.component, cfg)?, scratch)) Ok((self.broker.runtime.start(self.component, cfg)?, scratch))
} }
+2 -11
View File
@@ -12,7 +12,7 @@ use tokio::sync::{mpsc, oneshot};
use wasmtime::component::Component; use wasmtime::component::Component;
use crate::broker::{Broker, Envelope, SHUTDOWN_DEADLINE, shutdown_line}; 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 /// A new version that crashes this often within `CRASH_WINDOW` after an
/// upgrade is rolled back automatically. /// upgrade is rolled back automatically.
@@ -178,16 +178,7 @@ impl Supervisor {
if self.component.is_none() { if self.component.is_none() {
self.component = Some(self.broker.load_version(&self.id, self.version)?); self.component = Some(self.broker.load_version(&self.id, self.version)?);
} }
let world = &self.broker.world; let cfg = self.broker.instance_config(&self.id, self.version, None);
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()),
],
};
self.instance = Some(self.broker.runtime.start(self.component.as_ref().unwrap(), cfg)?); self.instance = Some(self.broker.runtime.start(self.component.as_ref().unwrap(), cfg)?);
Ok(()) Ok(())
} }