Sausage: host, broker, commit-on-close filesystem, cargo actor, console, agent
Imported from the spike branch, with: - a library crate (Host API) for the CLI and the tests - JSON test scenarios for candidates - commit on close only for new or truncated files - a Nix flake: nix run -- init, nix run -- run
This commit is contained in:
@@ -0,0 +1,856 @@
|
||||
//! The broker: message routing, mailboxes, subscriptions, and the lifecycle of
|
||||
//! WASM actors (spawn, upgrade, rollback, stop, start, remove).
|
||||
//!
|
||||
//! The broker accepts only `.wasm` files. It does not know any source language.
|
||||
|
||||
use std::collections::{BTreeSet, HashMap};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use anyhow::{Context, Result, anyhow, bail};
|
||||
use serde_json::{Value, json};
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
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};
|
||||
|
||||
pub const MAILBOX_SIZE: usize = 1000;
|
||||
pub const SHUTDOWN_DEADLINE: Duration = Duration::from_secs(5);
|
||||
/// A new version that crashes this often within `CRASH_WINDOW` after a swap
|
||||
/// is rolled back automatically.
|
||||
const CRASH_LIMIT: usize = 3;
|
||||
const CRASH_WINDOW: Duration = Duration::from_secs(30);
|
||||
const STDIN_WRITE_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Envelope {
|
||||
pub id: String,
|
||||
pub from: String,
|
||||
pub to: String,
|
||||
pub reply_to: Option<String>,
|
||||
pub body: Value,
|
||||
}
|
||||
|
||||
impl Envelope {
|
||||
/// The NDJSON line that the recipient reads on stdin.
|
||||
pub fn to_line(&self) -> String {
|
||||
let mut v = json!({"id": self.id, "from": self.from, "body": self.body});
|
||||
if let Some(r) = &self.reply_to {
|
||||
v["reply_to"] = json!(r);
|
||||
}
|
||||
v.to_string()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn shutdown_line() -> String {
|
||||
json!({"id": "broker/shutdown", "from": "broker", "body": {"type": "shutdown"}}).to_string()
|
||||
}
|
||||
|
||||
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>> },
|
||||
}
|
||||
|
||||
pub struct Broker {
|
||||
pub world: World,
|
||||
pub runtime: Arc<Runtime>,
|
||||
mailboxes: Mutex<HashMap<String, mpsc::Sender<Envelope>>>,
|
||||
controls: Mutex<HashMap<String, mpsc::Sender<Control>>>,
|
||||
registry: Mutex<Registry>,
|
||||
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()?;
|
||||
Ok(Arc::new(Self {
|
||||
world,
|
||||
runtime,
|
||||
mailboxes: Mutex::new(HashMap::new()),
|
||||
controls: Mutex::new(HashMap::new()),
|
||||
registry: Mutex::new(registry),
|
||||
op_locks: Mutex::new(HashMap::new()),
|
||||
seq: AtomicU64::new(1),
|
||||
}))
|
||||
}
|
||||
|
||||
pub fn registry(&self) -> Registry {
|
||||
self.registry.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
fn update_registry(&self, f: impl FnOnce(&mut Registry)) -> Result<()> {
|
||||
let mut reg = self.registry.lock().unwrap();
|
||||
f(&mut reg);
|
||||
self.world.save_registry(®)
|
||||
}
|
||||
|
||||
/// Create the mailbox of an actor. Messages wait here while the actor is
|
||||
/// down, during a swap, or while it restarts.
|
||||
pub fn open_mailbox(&self, id: &str) -> mpsc::Receiver<Envelope> {
|
||||
let (tx, rx) = mpsc::channel(MAILBOX_SIZE);
|
||||
self.mailboxes.lock().unwrap().insert(id.to_string(), tx);
|
||||
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 {
|
||||
Some(own) => format!("{from}/{own}"),
|
||||
None => self.next_id(),
|
||||
};
|
||||
let recipients: Vec<String> = match to {
|
||||
Some(to) => vec![to.to_string()],
|
||||
None => self
|
||||
.registry
|
||||
.lock()
|
||||
.unwrap()
|
||||
.subscriptions
|
||||
.get(from)
|
||||
.map(|s| s.iter().cloned().collect())
|
||||
.unwrap_or_default(),
|
||||
};
|
||||
for to in recipients {
|
||||
self.deliver(Envelope {
|
||||
id: id.clone(),
|
||||
from: from.to_string(),
|
||||
to,
|
||||
reply_to: reply_to.map(str::to_string),
|
||||
body: body.clone(),
|
||||
});
|
||||
}
|
||||
id
|
||||
}
|
||||
|
||||
/// Route one stdout line of a WASM actor.
|
||||
pub fn route_line(&self, from: &str, line: &str) {
|
||||
let line = line.trim();
|
||||
if line.is_empty() {
|
||||
return;
|
||||
}
|
||||
let parsed: Option<Value> = serde_json::from_str(line).ok();
|
||||
let Some(Value::Object(obj)) = parsed else {
|
||||
self.log(from, &format!("stdout line is not a JSON object, ignored: {}", truncate(line, 500)));
|
||||
return;
|
||||
};
|
||||
let Some(body) = obj.get("body").cloned() else {
|
||||
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()),
|
||||
_ => None,
|
||||
});
|
||||
let reply_to = obj.get("reply_to").and_then(Value::as_str);
|
||||
self.send(from, to, own_id.as_deref(), reply_to, body);
|
||||
}
|
||||
|
||||
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)),
|
||||
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)),
|
||||
},
|
||||
};
|
||||
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}),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Append a host note to the stderr log of an actor.
|
||||
pub fn log(&self, id: &str, text: &str) {
|
||||
use std::io::Write;
|
||||
let path = self.world.actor_dir(id).join("stderr.log");
|
||||
if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(path) {
|
||||
let _ = writeln!(f, "[sausage {}] {text}", world::now());
|
||||
}
|
||||
}
|
||||
|
||||
fn op_lock(&self, id: &str) -> Arc<tokio::sync::Mutex<()>> {
|
||||
self.op_locks.lock().unwrap().entry(id.to_string()).or_default().clone()
|
||||
}
|
||||
|
||||
/// Mounts of a running actor.
|
||||
pub fn mounts(&self, id: &str) -> Vec<Mount> {
|
||||
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 },
|
||||
]
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// 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());
|
||||
}
|
||||
|
||||
async fn control(&self, id: &str, make: impl FnOnce(oneshot::Sender<Result<Value>>) -> 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) {
|
||||
let ids: Vec<String> = self.controls.lock().unwrap().keys().cloned().collect();
|
||||
let mut waits = Vec::new();
|
||||
for id in ids {
|
||||
waits.push(async move {
|
||||
let _ = self.control(&id, |reply| Control::Shutdown { reply }).await;
|
||||
});
|
||||
}
|
||||
futures_join_all(waits).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);
|
||||
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::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))
|
||||
}
|
||||
|
||||
/// 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> {
|
||||
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) {
|
||||
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 {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
let _ = std::fs::remove_dir_all(self.world.versions_dir(id));
|
||||
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"}));
|
||||
if start {
|
||||
self.start_supervisor(id, 1, true);
|
||||
}
|
||||
Ok(json!({"type": "spawned", "actor": id, "version": 1, "tests": report}))
|
||||
}
|
||||
|
||||
pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, mut 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);
|
||||
}
|
||||
let mut result = self
|
||||
.control(id, |reply| Control::Upgrade { version, requester: meta.requested_by.clone(), reply })
|
||||
.await?;
|
||||
result["tests"] = report.unwrap();
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
/// 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:#}")}),
|
||||
};
|
||||
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 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 = source_s.map(|s| self.world.resolve(&env.from, s)).transpose()?;
|
||||
if let Some(s) = &source
|
||||
&& !s.is_dir()
|
||||
{
|
||||
bail!("{} is not a folder", source_s.unwrap());
|
||||
}
|
||||
let m = VersionMeta {
|
||||
version,
|
||||
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),
|
||||
requested_by: env.from.clone(),
|
||||
created_at: world::now(),
|
||||
};
|
||||
Ok((wasm, source, m))
|
||||
};
|
||||
match ty {
|
||||
"spawn" => {
|
||||
let id = actor()?;
|
||||
let (wasm, source, m) = meta(1)?;
|
||||
self.create_actor(&id, &wasm, source.as_deref(), m, true).await
|
||||
}
|
||||
"upgrade" => {
|
||||
let id = actor()?;
|
||||
let (wasm, source, m) = meta(0)?;
|
||||
self.upgrade_actor(&id, &wasm, source.as_deref(), m).await
|
||||
}
|
||||
"rollback" => {
|
||||
let id = actor()?;
|
||||
let target = b.get("version").and_then(Value::as_u64);
|
||||
let lock = self.op_lock(&id);
|
||||
let _g = lock.lock().await;
|
||||
self.control(&id, |reply| Control::Rollback { target, reply }).await
|
||||
}
|
||||
"stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await,
|
||||
"start" => self.control(&actor()?, |reply| Control::Start { reply }).await,
|
||||
"remove" => {
|
||||
let id = actor()?;
|
||||
let lock = self.op_lock(&id);
|
||||
let _g = 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);
|
||||
}
|
||||
})?;
|
||||
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);
|
||||
if on {
|
||||
set.insert(sub.clone());
|
||||
} else {
|
||||
set.remove(&sub);
|
||||
}
|
||||
})?;
|
||||
Ok(json!({"type": format!("{ty}d"), "actor": publisher}))
|
||||
}
|
||||
"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}))
|
||||
}
|
||||
"shutdown" => Ok(json!({"type": "ignored"})),
|
||||
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, list"),
|
||||
}
|
||||
}
|
||||
|
||||
/// The native `broker` actor.
|
||||
pub async fn run_actor(self: Arc<Self>, mut mailbox: mpsc::Receiver<Envelope>) {
|
||||
while let Some(env) = mailbox.recv().await {
|
||||
tokio::spawn(self.clone().handle(env));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
pub fn truncate(s: &str, max: usize) -> String {
|
||||
if s.len() <= max {
|
||||
return s.to_string();
|
||||
}
|
||||
let mut end = max;
|
||||
while !s.is_char_boundary(end) {
|
||||
end -= 1;
|
||||
}
|
||||
format!("{}… ({} bytes)", &s[..end], s.len())
|
||||
}
|
||||
|
||||
fn write_status(world: &World, id: &str, status: &Value) {
|
||||
let mut s = status.clone();
|
||||
s["updated_at"] = json!(world::now());
|
||||
let _ = world::write_atomic(
|
||||
&world.actor_dir(id).join("status.json"),
|
||||
serde_json::to_string_pretty(&s).unwrap_or_default().as_bytes(),
|
||||
);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Supervisor: owns one WASM actor and its instance.
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
struct Supervisor {
|
||||
broker: Arc<Broker>,
|
||||
id: String,
|
||||
version: u64,
|
||||
desired_running: bool,
|
||||
mailbox: mpsc::Receiver<Envelope>,
|
||||
control: mpsc::Receiver<Control>,
|
||||
component: Option<Component>,
|
||||
instance: Option<Instance>,
|
||||
state: &'static str,
|
||||
started_at: Instant,
|
||||
restarts: u32,
|
||||
restart_at: Option<Instant>,
|
||||
last_exit: Option<String>,
|
||||
last_delivered: Option<Envelope>,
|
||||
/// The message that the actor got just before it crashed. The broker
|
||||
/// delivers it again after the next upgrade.
|
||||
parked: Option<Envelope>,
|
||||
redeliver: Option<Envelope>,
|
||||
previous: Option<u64>,
|
||||
swapped_at: Option<Instant>,
|
||||
crashes_since_swap: usize,
|
||||
upgrade_requester: Option<String>,
|
||||
}
|
||||
|
||||
async fn recv_line(instance: &mut Option<Instance>) -> Option<String> {
|
||||
match instance {
|
||||
Some(i) => i.stdout.recv().await,
|
||||
None => std::future::pending().await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn sleep_until_opt(t: Option<Instant>) {
|
||||
match t {
|
||||
Some(t) => tokio::time::sleep_until(t.into()).await,
|
||||
None => std::future::pending().await,
|
||||
}
|
||||
}
|
||||
|
||||
impl Supervisor {
|
||||
fn new(
|
||||
broker: Arc<Broker>,
|
||||
id: String,
|
||||
version: u64,
|
||||
running: bool,
|
||||
mailbox: mpsc::Receiver<Envelope>,
|
||||
control: mpsc::Receiver<Control>,
|
||||
) -> Self {
|
||||
Self {
|
||||
broker,
|
||||
id,
|
||||
version,
|
||||
desired_running: running,
|
||||
mailbox,
|
||||
control,
|
||||
component: None,
|
||||
instance: None,
|
||||
state: if running { "starting" } else { "stopped" },
|
||||
started_at: Instant::now(),
|
||||
restarts: 0,
|
||||
restart_at: None,
|
||||
last_exit: None,
|
||||
last_delivered: None,
|
||||
parked: None,
|
||||
redeliver: None,
|
||||
previous: None,
|
||||
swapped_at: None,
|
||||
crashes_since_swap: 0,
|
||||
upgrade_requester: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn world(&self) -> &World {
|
||||
&self.broker.world
|
||||
}
|
||||
|
||||
fn status(&self) {
|
||||
write_status(
|
||||
self.world(),
|
||||
&self.id,
|
||||
&json!({
|
||||
"id": self.id,
|
||||
"version": self.version,
|
||||
"state": self.state,
|
||||
"restarts": self.restarts,
|
||||
"last_exit": self.last_exit,
|
||||
"parked_message": self.parked.as_ref().map(|m| json!({"id": m.id, "from": m.from, "body": m.body})),
|
||||
"previous_version": self.previous,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async fn run(mut self) {
|
||||
self.status();
|
||||
loop {
|
||||
if self.desired_running && self.instance.is_none() && self.restart_at.is_none_or(|t| Instant::now() >= t) {
|
||||
self.restart_at = None;
|
||||
self.start_instance().await;
|
||||
}
|
||||
let has_instance = self.instance.is_some();
|
||||
let waiting = !has_instance && self.desired_running;
|
||||
tokio::select! {
|
||||
ctl = self.control.recv() => {
|
||||
let Some(ctl) = ctl else { break };
|
||||
if self.handle_control(ctl).await {
|
||||
break;
|
||||
}
|
||||
}
|
||||
msg = self.mailbox.recv(), if has_instance => {
|
||||
let Some(msg) = msg else { break };
|
||||
self.deliver(msg).await;
|
||||
}
|
||||
line = recv_line(&mut self.instance) => {
|
||||
match line {
|
||||
Some(line) => self.broker.route_line(&self.id, &line),
|
||||
None => {
|
||||
let exit = self.instance.as_mut().unwrap().wait().await;
|
||||
self.instance = None;
|
||||
self.on_exit(exit).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = sleep_until_opt(self.restart_at), if waiting => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn deliver(&mut self, msg: Envelope) {
|
||||
let Some(inst) = self.instance.as_mut() else { return };
|
||||
let line = msg.to_line();
|
||||
match tokio::time::timeout(STDIN_WRITE_TIMEOUT, inst.send_line(&line)).await {
|
||||
Ok(Ok(())) => self.last_delivered = Some(msg),
|
||||
Ok(Err(_)) => {} // The instance is ending. The exit path handles it.
|
||||
Err(_) => {
|
||||
let note = format!("actor did not read stdin for {STDIN_WRITE_TIMEOUT:?}; message {} dropped", msg.id);
|
||||
self.broker.log(&self.id, ¬e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn start_instance(&mut self) {
|
||||
if self.component.is_none() {
|
||||
let path = self.world().version_dir(&self.id, self.version).join("actor.wasm");
|
||||
let t = Instant::now();
|
||||
match self.broker.runtime.load(&path) {
|
||||
Ok(c) => {
|
||||
self.broker.log(&self.id, &format!("compiled version {} in {} ms", self.version, t.elapsed().as_millis()));
|
||||
self.component = Some(c);
|
||||
}
|
||||
Err(e) => {
|
||||
self.broker.log(&self.id, &format!("cannot load version {}: {e:#}", self.version));
|
||||
self.state = "broken";
|
||||
self.desired_running = false;
|
||||
self.status();
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
let cfg = InstanceConfig {
|
||||
mounts: self.broker.mounts(&self.id),
|
||||
allow_http: true,
|
||||
stderr_log: self.world().actor_dir(&self.id).join("stderr.log"),
|
||||
env: vec![
|
||||
("SAUSAGE_ACTOR".into(), self.id.clone()),
|
||||
("SAUSAGE_VERSION".into(), self.version.to_string()),
|
||||
],
|
||||
};
|
||||
match self.broker.runtime.start(self.component.as_ref().unwrap(), cfg) {
|
||||
Ok(inst) => {
|
||||
self.instance = Some(inst);
|
||||
self.state = "running";
|
||||
self.started_at = Instant::now();
|
||||
self.status();
|
||||
if let Some(msg) = self.redeliver.take() {
|
||||
self.broker.log(&self.id, &format!("delivering parked message {} again", msg.id));
|
||||
self.deliver(msg).await;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
self.broker.log(&self.id, &format!("cannot start: {e:#}"));
|
||||
self.state = "broken";
|
||||
self.desired_running = false;
|
||||
self.status();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn stop_instance(&mut self) -> Option<Exit> {
|
||||
let inst = self.instance.take()?;
|
||||
let exit = inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE).await;
|
||||
self.broker.log(&self.id, &format!("stopped: {exit}"));
|
||||
self.last_exit = Some(exit.to_string());
|
||||
Some(exit)
|
||||
}
|
||||
|
||||
async fn on_exit(&mut self, exit: Exit) {
|
||||
self.broker.log(&self.id, &format!("instance ended unexpectedly: {exit}"));
|
||||
self.last_exit = Some(exit.to_string());
|
||||
if !exit.is_clean()
|
||||
&& let Some(m) = self.last_delivered.take()
|
||||
{
|
||||
self.parked = Some(m);
|
||||
}
|
||||
if self.started_at.elapsed() > Duration::from_secs(60) {
|
||||
self.restarts = 0;
|
||||
}
|
||||
if let (Some(prev), Some(at)) = (self.previous, self.swapped_at)
|
||||
&& at.elapsed() < CRASH_WINDOW
|
||||
{
|
||||
self.crashes_since_swap += 1;
|
||||
if self.crashes_since_swap >= CRASH_LIMIT {
|
||||
let from = self.version;
|
||||
self.broker.log(&self.id, &format!("version {from} crashed {CRASH_LIMIT} times after the upgrade; rolling back to {prev}"));
|
||||
let result = self.rollback(Some(prev)).await;
|
||||
let body = match result {
|
||||
Ok(_) => json!({"type": "rolled_back", "actor": self.id, "from": from, "to": prev, "reason": "crash loop after upgrade", "last_exit": exit.to_string()}),
|
||||
Err(e) => json!({"type": "error", "actor": self.id, "reason": format!("automatic rollback failed: {e:#}")}),
|
||||
};
|
||||
if let Some(r) = self.upgrade_requester.clone() {
|
||||
self.broker.send("broker", Some(&r), None, None, body);
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
let delay = Duration::from_millis(500 * 2u64.pow(self.restarts.min(6))).min(Duration::from_secs(30));
|
||||
self.restarts += 1;
|
||||
self.restart_at = Some(Instant::now() + delay);
|
||||
self.state = "restarting";
|
||||
self.status();
|
||||
}
|
||||
|
||||
fn snapshot_current(&self) -> Result<()> {
|
||||
let snap = self.world().version_dir(&self.id, self.version).join("snapshot");
|
||||
world::replace_tree(&self.world().data_dir(&self.id), &snap.join("data"))?;
|
||||
world::replace_tree(&self.world().public_dir(&self.id), &snap.join("public"))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
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))?;
|
||||
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;
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
async fn upgrade(&mut self, version: u64, requester: String) -> Result<Value> {
|
||||
self.stop_instance().await;
|
||||
self.snapshot_current()?;
|
||||
let previous = self.version;
|
||||
self.install(version)?;
|
||||
self.previous = Some(previous);
|
||||
self.swapped_at = Some(Instant::now());
|
||||
self.crashes_since_swap = 0;
|
||||
self.restarts = 0;
|
||||
self.restart_at = None;
|
||||
self.upgrade_requester = Some(requester);
|
||||
self.redeliver = self.parked.take();
|
||||
self.desired_running = true;
|
||||
self.start_instance().await;
|
||||
Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous}))
|
||||
}
|
||||
|
||||
async fn rollback(&mut self, target: Option<u64>) -> Result<Value> {
|
||||
let target = match target {
|
||||
Some(t) => t,
|
||||
None => (1..self.version)
|
||||
.rev()
|
||||
.find(|v| self.world().version_dir(&self.id, *v).join("actor.wasm").is_file())
|
||||
.ok_or_else(|| anyhow!("{} has no version before {}", self.id, self.version))?,
|
||||
};
|
||||
let dir = self.world().version_dir(&self.id, target);
|
||||
if !dir.join("actor.wasm").is_file() {
|
||||
bail!("{} has no version {target}", self.id);
|
||||
}
|
||||
self.stop_instance().await;
|
||||
self.snapshot_current()?;
|
||||
let from = self.version;
|
||||
let snap = dir.join("snapshot");
|
||||
if snap.is_dir() {
|
||||
world::replace_tree(&snap.join("data"), &self.world().data_dir(&self.id))?;
|
||||
world::replace_tree(&snap.join("public"), &self.world().public_dir(&self.id))?;
|
||||
}
|
||||
self.install(target)?;
|
||||
self.previous = None;
|
||||
self.swapped_at = None;
|
||||
self.restarts = 0;
|
||||
self.restart_at = None;
|
||||
self.desired_running = true;
|
||||
self.start_instance().await;
|
||||
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": target, "restored_state": snap.is_dir()}))
|
||||
}
|
||||
|
||||
/// Returns `true` when the supervisor must end.
|
||||
async fn handle_control(&mut self, ctl: Control) -> bool {
|
||||
match ctl {
|
||||
Control::Upgrade { version, requester, reply } => {
|
||||
let r = self.upgrade(version, requester).await;
|
||||
let _ = reply.send(r);
|
||||
}
|
||||
Control::Rollback { target, reply } => {
|
||||
let r = self.rollback(target).await;
|
||||
let _ = reply.send(r);
|
||||
}
|
||||
Control::Stop { reply } => {
|
||||
self.desired_running = false;
|
||||
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 _ = 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 _ = reply.send(r.map(|_| json!({"type": "started", "actor": self.id})));
|
||||
}
|
||||
Control::Remove { reply } => {
|
||||
self.desired_running = false;
|
||||
self.stop_instance().await;
|
||||
let r = self.archive();
|
||||
let _ = reply.send(r);
|
||||
return true;
|
||||
}
|
||||
Control::Shutdown { reply } => {
|
||||
self.stop_instance().await;
|
||||
self.state = "host stopped";
|
||||
self.status();
|
||||
let _ = reply.send(Ok(json!({"type": "shut_down"})));
|
||||
return true;
|
||||
}
|
||||
}
|
||||
self.status();
|
||||
false
|
||||
}
|
||||
|
||||
/// Move every 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))?;
|
||||
}
|
||||
}
|
||||
Ok(json!({"type": "removed", "actor": self.id, "archive": dest.display().to_string()}))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user