surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

World folder is the source of truth

Browse Source 
- actor/<id>/ holds actor.wasm, actor.json, source/, public/, data/, versions/
- the broker boots by scanning actor/; a changed actor.wasm is an upgrade,
  a rejected one stays in versions/ with a reason and the files come back
- /actor hides every data/ folder
- subscriptions live in the subscriber's actor.json
- cargo installs the SDK and points sausage-actor at it; no build limits
- PROTOCOL.md
This commit is contained in:
surma
2026-10-15 11:14:11 +01:00
parent c9d9864743
commit 0a8e8c2e3c
13 changed files with 786 additions and 335 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/broker.rs
+293 -229
View File
@@ -1,10 +1,12 @@
//! The broker: message routing, mailboxes, subscriptions, and the lifecycle of
//! WASM actors (spawn, upgrade, rollback, stop, start, remove).
//! WASM actors (spawn, upgrade, rollback, stop, start, remove, scan).
//!
//! The broker accepts only `.wasm` files. It does not know any source language.
//! The world folder is the source of truth: `scan` boots every
//! `actor/<id>/actor.wasm` and treats a changed file as an upgrade.
use std::collections::{BTreeSet, HashMap};
use std::path::{Path, PathBuf};
use std::collections::{BTreeMap, HashMap};
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
@@ -17,7 +19,7 @@ use wasmtime::component::Component;
use crate::candidate;
use crate::fs::Mount;
use crate::runtime::{Exit, Instance, InstanceConfig, Runtime};
use crate::world::{self, ActorRecord, NATIVE, Registry, VersionMeta, World};
use crate::world::{self, ActorMeta, NATIVE, VersionMeta, World};
pub const MAILBOX_SIZE: usize = 1000;
pub const SHUTDOWN_DEADLINE: Duration = Duration::from_secs(5);
@@ -51,13 +53,15 @@ pub fn shutdown_line() -> String {
json!({"id": "broker/shutdown", "from": "broker", "body": {"type": "shutdown"}}).to_string()
}
type Reply = oneshot::Sender<Result<Value>>;
enum Control {
Upgrade { version: u64, requester: String, reply: oneshot::Sender<Result<Value>> },
Rollback { target: Option<u64>, reply: oneshot::Sender<Result<Value>> },
Stop { reply: oneshot::Sender<Result<Value>> },
Start { reply: oneshot::Sender<Result<Value>> },
Remove { reply: oneshot::Sender<Result<Value>> },
Shutdown { reply: oneshot::Sender<Result<Value>> },
Upgrade { version: u64, requester: String, reply: Reply },
Rollback { target: Option<u64>, reply: Reply },
Stop { reply: Reply },
Start { reply: Reply },
Remove { reply: Reply },
Shutdown { reply: Reply },
}
pub struct Broker {
@@ -65,33 +69,36 @@ pub struct Broker {
pub runtime: Arc<Runtime>,
mailboxes: Mutex<HashMap<String, mpsc::Sender<Envelope>>>,
controls: Mutex<HashMap<String, mpsc::Sender<Control>>>,
registry: Mutex<Registry>,
/// `actor.json` of every actor, native and WASM.
metas: Mutex<BTreeMap<String, ActorMeta>>,
op_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
seq: AtomicU64,
}
impl Broker {
pub fn new(world: World, runtime: Arc<Runtime>) -> Result<Arc<Self>> {
let registry = world.load_registry()?;
let metas = world.actor_ids()?.into_iter().map(|id| (id.clone(), world.read_meta(&id))).collect();
Ok(Arc::new(Self {
world,
runtime,
mailboxes: Mutex::new(HashMap::new()),
controls: Mutex::new(HashMap::new()),
registry: Mutex::new(registry),
metas: Mutex::new(metas),
op_locks: Mutex::new(HashMap::new()),
seq: AtomicU64::new(1),
}))
}
pub fn registry(&self) -> Registry {
self.registry.lock().unwrap().clone()
pub fn meta(&self, id: &str) -> Option<ActorMeta> {
self.metas.lock().unwrap().get(id).cloned()
}
fn update_registry(&self, f: impl FnOnce(&mut Registry)) -> Result<()> {
let mut reg = self.registry.lock().unwrap();
f(&mut reg);
self.world.save_registry(&reg)
/// Change the `actor.json` of `id` (creating it if needed) and save it.
fn update_meta(&self, id: &str, f: impl FnOnce(&mut ActorMeta)) -> Result<()> {
let mut metas = self.metas.lock().unwrap();
let meta = metas.entry(id.to_string()).or_insert_with(|| ActorMeta { running: true, ..Default::default() });
f(meta);
self.world.write_meta(id, meta)
}
/// Create the mailbox of an actor. Messages wait here while the actor is
@@ -102,27 +109,23 @@ impl Broker {
rx
}
fn next_id(&self) -> String {
format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed))
}
/// Send a message from a native actor. `to: None` publishes to the
/// subscribers of `from`. Returns the message ID.
pub fn send(&self, from: &str, to: Option<&str>, id: Option<&str>, reply_to: Option<&str>, body: Value) -> String {
let id = match id {
/// Send a message. `to: None` publishes to the subscribers of `from`.
/// With `own_id`, the message ID is `"<from>/<own_id>"`. Returns the ID.
pub fn send(&self, from: &str, to: Option<&str>, own_id: Option<&str>, reply_to: Option<&str>, body: Value) -> String {
let id = match own_id {
Some(own) => format!("{from}/{own}"),
None => self.next_id(),
None => format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
};
let recipients: Vec<String> = match to {
Some(to) => vec![to.to_string()],
None => self
.registry
.metas
.lock()
.unwrap()
.subscriptions
.get(from)
.map(|s| s.iter().cloned().collect())
.unwrap_or_default(),
.iter()
.filter(|(_, m)| m.subscriptions.contains(from))
.map(|(id, _)| id.clone())
.collect(),
};
for to in recipients {
self.deliver(Envelope {
@@ -142,8 +145,7 @@ impl Broker {
if line.is_empty() {
return;
}
let parsed: Option<Value> = serde_json::from_str(line).ok();
let Some(Value::Object(obj)) = parsed else {
let Ok(Value::Object(obj)) = serde_json::from_str::<Value>(line) else {
self.log(from, &format!("stdout line is not a JSON object, ignored: {}", truncate(line, 500)));
return;
};
@@ -151,12 +153,12 @@ impl Broker {
self.log(from, &format!("message without \"body\", ignored: {}", truncate(line, 500)));
return;
};
let to = obj.get("to").and_then(Value::as_str);
let own_id = obj.get("id").and_then(|v| match v {
Value::String(s) => Some(s.clone()),
Value::Number(n) => Some(n.to_string()),
let own_id = match obj.get("id") {
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 reply_to = obj.get("reply_to").and_then(Value::as_str);
self.send(from, to, own_id.as_deref(), reply_to, body);
}
@@ -164,24 +166,16 @@ impl Broker {
fn deliver(&self, env: Envelope) {
let tx = self.mailboxes.lock().unwrap().get(&env.to).cloned();
let error = match tx {
None => Some(format!("no actor with ID {}", env.to)),
None => format!("no actor with ID {}", env.to),
Some(tx) => match tx.try_send(env.clone()) {
Ok(()) => None,
Err(mpsc::error::TrySendError::Full(_)) => Some(format!("mailbox of {} is full", env.to)),
Err(mpsc::error::TrySendError::Closed(_)) => Some(format!("actor {} is gone", env.to)),
Ok(()) => return,
Err(mpsc::error::TrySendError::Full(_)) => format!("mailbox of {} is full", env.to),
Err(mpsc::error::TrySendError::Closed(_)) => format!("actor {} is gone", env.to),
},
};
if let Some(reason) = error
&& env.from != "broker"
&& env.from != "test"
{
self.send(
"broker",
Some(&env.from),
None,
Some(&env.id),
json!({"type": "undeliverable", "reason": reason, "to": env.to}),
);
if env.from != "broker" {
let body = json!({"type": "undeliverable", "reason": error, "to": env.to});
self.send("broker", Some(&env.from), None, Some(&env.id), body);
}
}
@@ -198,7 +192,7 @@ impl Broker {
self.op_locks.lock().unwrap().entry(id.to_string()).or_default().clone()
}
/// Mounts of a running actor.
/// Mounts of a running actor. `/actor` hides every `data/` folder.
pub fn mounts(&self, id: &str) -> Vec<Mount> {
vec![
Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: false },
@@ -211,185 +205,280 @@ impl Broker {
// Lifecycle
// ---------------------------------------------------------------------
/// Start the supervisors of all registered WASM actors.
pub fn start_all(self: &Arc<Self>) {
let reg = self.registry();
for (id, rec) in reg.actors {
self.start_supervisor(&id, rec.version, rec.running);
}
}
fn start_supervisor(self: &Arc<Self>, id: &str, version: u64, running: bool) {
let mailbox = self.open_mailbox(id);
let (ctl_tx, ctl_rx) = mpsc::channel(16);
self.controls.lock().unwrap().insert(id.to_string(), ctl_tx);
let sv = Supervisor::new(self.clone(), id.to_string(), version, running, mailbox, ctl_rx);
tokio::spawn(sv.run());
tokio::spawn(Supervisor::new(self.clone(), id.to_string(), version, running, mailbox, ctl_rx).run());
}
async fn control(&self, id: &str, make: impl FnOnce(oneshot::Sender<Result<Value>>) -> Control) -> Result<Value> {
fn is_supervised(&self, id: &str) -> bool {
self.controls.lock().unwrap().contains_key(id)
}
async fn control(&self, id: &str, make: impl FnOnce(Reply) -> Control) -> Result<Value> {
let tx = self.controls.lock().unwrap().get(id).cloned().ok_or_else(|| anyhow!("no WASM actor with ID {id}"))?;
let (reply, rx) = oneshot::channel();
tx.send(make(reply)).await.map_err(|_| anyhow!("supervisor of {id} is gone"))?;
rx.await.map_err(|_| anyhow!("supervisor of {id} did not answer"))?
}
/// Stop every actor gracefully (host shutdown). The registry keeps their
/// `running` flags, so the next `sausage run` starts them again.
pub async fn shutdown_all(&self) {
/// Stop every actor gracefully, in parallel (host shutdown). The
/// `running` flags stay, so the next `sausage run` starts them again.
pub async fn shutdown_all(self: &Arc<Self>) {
let ids: Vec<String> = self.controls.lock().unwrap().keys().cloned().collect();
let mut waits = Vec::new();
let mut tasks = tokio::task::JoinSet::new();
for id in ids {
waits.push(async move {
let _ = self.control(&id, |reply| Control::Shutdown { reply }).await;
});
let this = self.clone();
tasks.spawn(async move { this.control(&id, |reply| Control::Shutdown { reply }).await });
}
futures_join_all(waits).await;
tasks.join_all().await;
}
/// Store a new version: copy the `.wasm` file and the source folder into
/// the version store. Returns the version directory and the component.
fn store_version(&self, id: &str, version: u64, wasm: &Path, source: Option<&Path>, meta: &VersionMeta) -> Result<(PathBuf, Component)> {
let dir = self.world.version_dir(id, version);
let _ = std::fs::remove_dir_all(&dir);
/// Copy a `.wasm` file and its source folder into `versions/<n>/`.
/// Every version that the broker sees stays there, also rejected ones.
fn store_version(&self, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Component> {
let dir = self.world.version_dir(id, meta.version);
std::fs::create_dir_all(&dir)?;
std::fs::copy(wasm, dir.join("actor.wasm")).with_context(|| format!("copy {}", wasm.display()))?;
match source {
Some(src) => world::copy_tree(src, &dir.join("source"))?,
None => std::fs::create_dir_all(dir.join("source"))?,
}
world::replace_tree(source.unwrap_or(&dir.join("no-source")), &dir.join("source"))?;
world::write_atomic(&dir.join("meta.json"), serde_json::to_string_pretty(meta)?.as_bytes())?;
let component = self.runtime.load(&dir.join("actor.wasm"))?;
Ok((dir, component))
self.runtime.load(&dir.join("actor.wasm"))
}
/// Record in `versions/<n>/meta.json` why the broker refused a version.
fn reject(&self, id: &str, version: u64, reason: &anyhow::Error) {
let path = self.world.version_dir(id, version).join("meta.json");
let mut meta: VersionMeta = std::fs::read_to_string(&path)
.ok()
.and_then(|s| serde_json::from_str(&s).ok())
.unwrap_or_default();
meta.rejected = Some(format!("{reason:#}"));
let _ = world::write_atomic(&path, serde_json::to_string_pretty(&meta).unwrap_or_default().as_bytes());
}
fn version_rejected(&self, id: &str, version: u64) -> bool {
std::fs::read_to_string(self.world.version_dir(id, version).join("meta.json"))
.ok()
.and_then(|s| serde_json::from_str::<VersionMeta>(&s).ok())
.is_some_and(|m| m.rejected.is_some())
}
/// Install `version` as the current files of a new actor.
fn install_files(&self, id: &str, version: u64) -> Result<()> {
let dir = self.world.version_dir(id, version);
self.world.ensure_actor_dirs(id)?;
world::write_atomic(&self.world.wasm_path(id), &std::fs::read(dir.join("actor.wasm"))?)?;
world::replace_tree(&dir.join("source"), &self.world.source_dir(id))?;
self.update_meta(id, |m| {
m.version = Some(version);
m.running = true;
})
}
/// Create a new actor from a `.wasm` file. With `start`, also start its
/// supervisor (in a running host). Without `start` (during `init`), only
/// register it.
pub async fn create_actor(
self: &Arc<Self>,
id: &str,
wasm: &Path,
source: Option<&Path>,
meta: VersionMeta,
start: bool,
) -> Result<Value> {
/// supervisor (in a running host). Without `start` (`init`), only install it.
pub async fn create_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta, start: bool) -> Result<Value> {
if !world::valid_id(id) {
bail!("invalid actor ID {id:?}: use a-z, 0-9, '-' and '_'");
}
if NATIVE.contains(&id) || self.registry().actors.contains_key(id) {
if NATIVE.contains(&id) || self.is_supervised(id) || self.world.wasm_path(id).exists() {
bail!("an actor with ID {id} already exists");
}
let lock = self.op_lock(id);
let _guard = lock.lock().await;
let (dir, component) = self.store_version(id, 1, wasm, source, &meta)?;
let report = candidate::test(self, id, &component, &dir.join("source"), None, None).await;
let report = match report {
let version = self.world.next_version(id);
let meta = VersionMeta { version, ..meta };
self.create_version(id, version, &meta, wasm, source, start).await
}
async fn create_version(self: &Arc<Self>, id: &str, version: u64, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, start: bool) -> Result<Value> {
let tested = async {
let component = self.store_version(id, meta, wasm, source)?;
candidate::test(self, id, &component, &self.world.version_dir(id, version).join("source"), None, None).await
}
.await;
let report = match tested {
Ok(r) => r,
Err(e) => {
let _ = std::fs::remove_dir_all(self.world.versions_dir(id));
self.reject(id, version, &e);
return Err(e);
}
};
self.world.ensure_actor_dirs(id)?;
world::replace_tree(&dir.join("source"), &self.world.source_dir(id))?;
self.update_registry(|r| {
r.actors.insert(id.to_string(), ActorRecord { version: 1, next_version: 2, running: true });
})?;
write_status(&self.world, id, &json!({"id": id, "version": 1, "state": "created"}));
self.install_files(id, version)?;
write_status(&self.world, id, &json!({"id": id, "version": version, "state": "created"}));
if start {
self.start_supervisor(id, 1, true);
self.start_supervisor(id, version, true);
}
Ok(json!({"type": "spawned", "actor": id, "version": 1, "tests": report}))
Ok(json!({"type": "spawned", "actor": id, "version": version, "tests": report}))
}
pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, mut meta: VersionMeta) -> Result<Value> {
pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> Result<Value> {
let lock = self.op_lock(id);
let _guard = lock.lock().await;
let rec = self.registry().actors.get(id).cloned().ok_or_else(|| anyhow!("no WASM actor with ID {id}"))?;
let version = rec.next_version;
self.update_registry(|r| {
if let Some(a) = r.actors.get_mut(id) {
a.next_version = version + 1;
}
})?;
meta.version = version;
let (dir, component) = self.store_version(id, version, wasm, source, &meta)?;
let report = candidate::test(
self,
id,
&component,
&dir.join("source"),
Some(&self.world.data_dir(id)),
Some(&self.world.public_dir(id)),
)
.await;
if let Err(e) = report {
let _ = std::fs::remove_dir_all(&dir);
return Err(e);
if !self.is_supervised(id) {
bail!("no WASM actor with ID {id}");
}
let mut result = self
.control(id, |reply| Control::Upgrade { version, requester: meta.requested_by.clone(), reply })
.await?;
result["tests"] = report.unwrap();
let version = self.world.next_version(id);
self.upgrade_version(id, &VersionMeta { version, ..meta }, wasm, source).await
}
async fn upgrade_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
let version = meta.version;
let tested = async {
let component = self.store_version(id, meta, wasm, source)?;
let source = self.world.version_dir(id, version).join("source");
candidate::test(self, id, &component, &source, Some(&self.world.data_dir(id)), Some(&self.world.public_dir(id))).await
}
.await;
let report = match tested {
Ok(r) => r,
Err(e) => {
self.reject(id, version, &e);
return Err(e);
}
};
let requester = meta.requested_by.clone();
let mut result = self.control(id, |reply| Control::Upgrade { version, requester, reply }).await?;
result["tests"] = report;
Ok(result)
}
/// Boot the actors in `actor/`. Called at start and on `scan`.
///
/// - A known `actor.wasm` (equal to the installed version) starts as is.
/// - A new or changed `actor.wasm` is first copied to `versions/<n>/`.
/// A new actor is tested and installed. A changed one is an upgrade:
/// if it fails, the installed files come back.
/// - A file that equals an already rejected version is not tried again.
pub async fn scan(self: &Arc<Self>) -> Vec<Value> {
let mut report = Vec::new();
for id in self.world.actor_ids().unwrap_or_default() {
if NATIVE.contains(&id.as_str()) || !self.world.wasm_path(&id).is_file() {
continue;
}
let lock = self.op_lock(&id);
let _guard = lock.lock().await;
if let Some(r) = self.scan_actor(&id).await {
report.push(r);
}
}
report
}
async fn scan_actor(self: &Arc<Self>, id: &str) -> Option<Value> {
let meta = self.world.read_meta(id);
let wasm = self.world.wasm_path(id);
let same = |v: u64| same_file(&wasm, &self.world.version_dir(id, v).join("actor.wasm"));
let installed = meta.version.filter(|v| self.world.version_dir(id, *v).join("actor.wasm").is_file());
self.metas.lock().unwrap().insert(id.to_string(), meta.clone());
if let Some(v) = installed
&& !self.is_supervised(id)
{
self.start_supervisor(id, v, meta.running);
}
if installed.is_some_and(same) {
return None;
}
let latest = self.world.versions(id).last().copied();
if let Some(v) = latest
&& same(v)
&& self.version_rejected(id, v)
{
if let Some(current) = installed {
let _ = self.restore_files(id, current);
}
self.log(id, &format!("actor.wasm equals the rejected version {v}; not installed"));
return Some(json!({"actor": id, "skipped": "rejected version", "version": v}));
}
let version = self.world.next_version(id);
let source = self.world.source_dir(id);
let vmeta = VersionMeta {
version,
wasm: format!("/actor/{id}/actor.wasm"),
source: Some(format!("/actor/{id}/source")),
built_by: None,
requested_by: "scan".into(),
created_at: world::now(),
rejected: None,
};
let source = source.is_dir().then_some(source.as_path());
let result = match installed {
None => self.create_version(id, version, &vmeta, &wasm, source, true).await,
Some(current) => {
let r = self.upgrade_version(id, &vmeta, &wasm, source).await;
if r.is_err() {
// Put the installed version back in place.
let _ = self.restore_files(id, current);
}
r
}
};
Some(match result {
Ok(v) => v,
Err(e) => {
self.log(id, &format!("version {version} from disk rejected: {e:#}"));
json!({"type": "error", "actor": id, "version": version, "reason": format!("{e:#}")})
}
})
}
fn restore_files(&self, id: &str, version: u64) -> Result<()> {
let dir = self.world.version_dir(id, version);
world::write_atomic(&self.world.wasm_path(id), &std::fs::read(dir.join("actor.wasm"))?)?;
world::replace_tree(&dir.join("source"), &self.world.source_dir(id))
}
/// Handle one message to `broker`.
async fn handle(self: Arc<Self>, env: Envelope) {
let result = self.clone().command(&env).await;
let body = match result {
Ok(v) => v,
Err(e) => json!({"type": "error", "request": env.body.get("type"), "reason": format!("{e:#}")}),
};
let body = self.clone().command(&env).await.unwrap_or_else(|e| {
json!({"type": "error", "request": env.body.get("type"), "reason": format!("{e:#}")})
});
self.send("broker", Some(&env.from), None, Some(&env.id), body);
}
async fn command(self: Arc<Self>, env: &Envelope) -> Result<Value> {
let b = &env.body;
let ty = b.get("type").and_then(Value::as_str).unwrap_or("");
let actor = || -> Result<String> {
b.get("actor").and_then(Value::as_str).map(str::to_string).ok_or_else(|| anyhow!("missing \"actor\""))
};
let meta = |version: u64| -> Result<(PathBuf, Option<PathBuf>, VersionMeta)> {
let wasm_s = b.get("wasm").and_then(Value::as_str).ok_or_else(|| anyhow!("missing \"wasm\""))?;
let ty = b["type"].as_str().unwrap_or("");
let actor = || b["actor"].as_str().map(str::to_string).ok_or_else(|| anyhow!("missing \"actor\""));
let version_meta = || -> Result<(std::path::PathBuf, Option<std::path::PathBuf>, VersionMeta)> {
let wasm_s = b["wasm"].as_str().ok_or_else(|| anyhow!("missing \"wasm\""))?;
let wasm = self.world.resolve(&env.from, wasm_s)?;
if !wasm.is_file() {
bail!("{wasm_s} is not a file");
}
let source_s = b.get("source").and_then(Value::as_str);
let source_s = b["source"].as_str();
let source = source_s.map(|s| self.world.resolve(&env.from, s)).transpose()?;
if let Some(s) = &source
&& !s.is_dir()
if let (Some(s), Some(p)) = (source_s, &source)
&& !p.is_dir()
{
bail!("{} is not a folder", source_s.unwrap());
bail!("{s} is not a folder");
}
let m = VersionMeta {
version,
let meta = VersionMeta {
version: 0,
wasm: wasm_s.to_string(),
source: source_s.map(str::to_string),
built_by: b.get("built_by").and_then(Value::as_str).map(str::to_string),
built_by: b["built_by"].as_str().map(str::to_string),
requested_by: env.from.clone(),
created_at: world::now(),
rejected: None,
};
Ok((wasm, source, m))
Ok((wasm, source, meta))
};
match ty {
"spawn" => {
let id = actor()?;
let (wasm, source, m) = meta(1)?;
self.create_actor(&id, &wasm, source.as_deref(), m, true).await
let (wasm, source, meta) = version_meta()?;
self.create_actor(&actor()?, &wasm, source.as_deref(), meta, true).await
}
"upgrade" => {
let id = actor()?;
let (wasm, source, m) = meta(0)?;
self.upgrade_actor(&id, &wasm, source.as_deref(), m).await
let (wasm, source, meta) = version_meta()?;
self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta).await
}
"rollback" => {
let id = actor()?;
let target = b.get("version").and_then(Value::as_u64);
let target = b["version"].as_u64();
let lock = self.op_lock(&id);
let _g = lock.lock().await;
let _guard = lock.lock().await;
self.control(&id, |reply| Control::Rollback { target, reply }).await
}
"stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await,
@@ -397,49 +486,44 @@ impl Broker {
"remove" => {
let id = actor()?;
let lock = self.op_lock(&id);
let _g = lock.lock().await;
let _guard = lock.lock().await;
let r = self.control(&id, |reply| Control::Remove { reply }).await?;
self.mailboxes.lock().unwrap().remove(&id);
self.controls.lock().unwrap().remove(&id);
self.update_registry(|reg| {
reg.actors.remove(&id);
reg.subscriptions.remove(&id);
for subs in reg.subscriptions.values_mut() {
subs.remove(&id);
}
})?;
self.metas.lock().unwrap().remove(&id);
Ok(r)
}
"subscribe" | "unsubscribe" => {
let publisher = actor()?;
let sub = env.from.clone();
let on = ty == "subscribe";
self.update_registry(|reg| {
let set = reg.subscriptions.entry(publisher.clone()).or_insert_with(BTreeSet::new);
self.update_meta(&env.from, |m| {
if on {
set.insert(sub.clone());
m.subscriptions.insert(publisher.clone());
} else {
set.remove(&sub);
m.subscriptions.remove(&publisher);
}
})?;
Ok(json!({"type": format!("{ty}d"), "actor": publisher}))
}
"scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})),
"list" => {
let reg = self.registry();
let mut actors: Vec<Value> = NATIVE.iter().map(|id| json!({"id": id, "kind": "native"})).collect();
for (id, rec) in reg.actors {
let status = std::fs::read_to_string(self.world.actor_dir(&id).join("status.json"))
.ok()
.and_then(|s| serde_json::from_str::<Value>(&s).ok());
actors.push(json!({
"id": id, "kind": "wasm", "version": rec.version, "running": rec.running,
"state": status.as_ref().and_then(|s| s.get("state")).cloned(),
}));
}
Ok(json!({"type": "actors", "actors": actors, "subscriptions": reg.subscriptions}))
let metas = self.metas.lock().unwrap().clone();
let actors: Vec<Value> = metas
.iter()
.map(|(id, m)| {
let status = std::fs::read_to_string(self.world.actor_dir(id).join("status.json"))
.ok()
.and_then(|s| serde_json::from_str::<Value>(&s).ok())
.unwrap_or_default();
json!({"id": id, "kind": if NATIVE.contains(&id.as_str()) { "native" } else { "wasm" },
"version": m.version, "running": m.running, "state": status["state"],
"subscriptions": m.subscriptions})
})
.collect();
Ok(json!({"type": "actors", "actors": actors}))
}
"shutdown" => Ok(json!({"type": "ignored"})),
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, list"),
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, scan, list"),
}
}
@@ -451,14 +535,8 @@ impl Broker {
}
}
async fn futures_join_all<F: Future<Output = ()>>(futs: Vec<F>) {
let mut set = Vec::new();
for f in futs {
set.push(Box::pin(f));
}
for f in set {
f.await;
}
fn same_file(a: &Path, b: &Path) -> bool {
matches!((std::fs::read(a), std::fs::read(b)), (Ok(x), Ok(y)) if x == y)
}
pub fn truncate(s: &str, max: usize) -> String {
@@ -723,20 +801,23 @@ impl Supervisor {
Ok(())
}
/// Make `version` the current files: `actor.wasm`, `source/`, `actor.json`.
fn install(&mut self, version: u64) -> Result<()> {
let src = self.world().version_dir(&self.id, version).join("source");
world::replace_tree(&src, &self.world().source_dir(&self.id))?;
let dir = self.world().version_dir(&self.id, version);
world::write_atomic(&self.world().wasm_path(&self.id), &std::fs::read(dir.join("actor.wasm"))?)?;
world::replace_tree(&dir.join("source"), &self.world().source_dir(&self.id))?;
self.version = version;
self.component = None;
let id = self.id.clone();
self.broker.update_registry(|r| {
if let Some(a) = r.actors.get_mut(&id) {
a.version = version;
a.running = true;
}
self.broker.update_meta(&self.id, |m| {
m.version = Some(version);
m.running = true;
})
}
fn set_running(&self, running: bool) -> Result<()> {
self.broker.update_meta(&self.id, |m| m.running = running)
}
async fn upgrade(&mut self, version: u64, requester: String) -> Result<Value> {
self.stop_instance().await;
self.snapshot_current()?;
@@ -759,7 +840,7 @@ impl Supervisor {
Some(t) => t,
None => (1..self.version)
.rev()
.find(|v| self.world().version_dir(&self.id, *v).join("actor.wasm").is_file())
.find(|v| self.world().version_dir(&self.id, *v).join("actor.wasm").is_file() && !self.broker.version_rejected(&self.id, *v))
.ok_or_else(|| anyhow!("{} has no version before {}", self.id, self.version))?,
};
let dir = self.world().version_dir(&self.id, target);
@@ -800,23 +881,13 @@ impl Supervisor {
self.restart_at = None;
self.stop_instance().await;
self.state = "stopped";
let id = self.id.clone();
let r = self.broker.update_registry(|r| {
if let Some(a) = r.actors.get_mut(&id) {
a.running = false;
}
});
let r = self.set_running(false);
let _ = reply.send(r.map(|_| json!({"type": "stopped", "actor": self.id})));
}
Control::Start { reply } => {
self.desired_running = true;
self.restart_at = None;
let id = self.id.clone();
let r = self.broker.update_registry(|r| {
if let Some(a) = r.actors.get_mut(&id) {
a.running = true;
}
});
let r = self.set_running(true);
let _ = reply.send(r.map(|_| json!({"type": "started", "actor": self.id})));
}
Control::Remove { reply } => {
@@ -838,19 +909,12 @@ impl Supervisor {
false
}
/// Move every folder of the actor into `archive/`. Nothing is deleted.
/// Move the folder of the actor into `archive/`. Nothing is deleted.
fn archive(&self) -> Result<Value> {
let dest = self.world().root.join("archive").join(format!("{}-{}", self.id, world::now()));
std::fs::create_dir_all(&dest)?;
for (from, name) in [
(self.world().actor_dir(&self.id), "actor"),
(self.world().root.join("private").join(&self.id), "private"),
(self.world().versions_dir(&self.id), "versions"),
] {
if from.exists() {
std::fs::rename(&from, dest.join(name))?;
}
}
let archive = self.world().root.join("archive");
std::fs::create_dir_all(&archive)?;
let dest = archive.join(format!("{}-{}", self.id, world::now()));
std::fs::rename(self.world().actor_dir(&self.id), &dest)?;
Ok(json!({"type": "removed", "actor": self.id, "archive": dest.display().to_string()}))
}
}