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

Trim the harness to what it needs to keep going

Browse Source 
Host:
- test first: a new version is tested where it lies; only an installed
  version goes to versions/, and instances run from there
- data/ and public/ are shared by all versions; no snapshots; rollback
  restores code only
- no re-delivery, status.json, list, remove/archive, smoke test, data/
  hiding, or stdin timeout; rollback needs a version
- supervisor.rs split out; one place for installing a version
- init only writes the agent folder; run installs it like any actor
- console: @actor, text to the agent, /help, /quit

Agent:
- every text goes out as a broadcast; 5-line prompt; 5 tools that describe
  themselves; turn.json only marks an active turn

Docs: PROTOCOL.md, README.md, DECISIONS.md, AGENTS.md rewritten.
Examples: only notes and notes-v2.
This commit is contained in:
surma
2026-10-15 13:35:45 +01:00
parent cee862275c
commit 11144b947e
35 changed files with 909 additions and 2038 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/broker.rs
+67 -586
View File
@@ -1,5 +1,6 @@
//! The broker: message routing, mailboxes, subscriptions, and the lifecycle of
//! WASM actors (spawn, upgrade, rollback, stop, start, remove, scan).
//! 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.
//!
//! The broker accepts only `.wasm` files. It does not know any source language.
//! The world folder is the source of truth: `scan` boots every
@@ -18,16 +19,12 @@ use wasmtime::component::Component;
use crate::candidate;
use crate::fs::Mount;
use crate::runtime::{Exit, Instance, InstanceConfig, Runtime};
use crate::runtime::Runtime;
use crate::supervisor::{Control, Reply, Supervisor};
use crate::world::{self, ActorMeta, NATIVE, 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 {
@@ -53,17 +50,6 @@ 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: Reply },
Rollback { target: Option<u64>, reply: Reply },
Stop { reply: Reply },
Start { reply: Reply },
Remove { reply: Reply },
Shutdown { reply: Reply },
}
pub struct Broker {
pub world: World,
pub runtime: Arc<Runtime>,
@@ -235,58 +221,51 @@ impl Broker {
tasks.join_all().await;
}
/// 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> {
/// Compile `versions/<version>/actor.wasm`.
pub fn load_version(&self, id: &str, version: u64) -> Result<Component> {
self.load(id, &self.world.version_dir(id, version).join("actor.wasm"))
}
fn load(&self, id: &str, wasm: &Path) -> Result<Component> {
let started = Instant::now();
let component = self.runtime.load(wasm)?;
self.log(id, &format!("compiled {} in {} ms", wasm.display(), started.elapsed().as_millis()));
Ok(component)
}
/// Test a new version where it lies. Only when the tests pass, copy it
/// to `versions/<n>/`: that folder holds only versions that run(ran).
/// Returns the component and the test report.
async fn try_version(&self, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<(Component, Value)> {
let component = self.load(id, wasm)?;
let tests = candidate::test(self, id, meta.version, &component, source).await;
if let Err(e) = &tests {
self.log(id, &format!("version {} from {} failed its tests: {e:#}", meta.version, meta.wasm));
}
let tests = tests?;
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()))?;
world::replace_source(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())?;
self.load_version(id, meta.version)
Ok((component, tests))
}
/// Compile `versions/<version>/actor.wasm`.
pub fn load_version(&self, id: &str, version: u64) -> Result<Component> {
let started = Instant::now();
let component = self.runtime.load(&self.world.version_dir(id, version).join("actor.wasm"))?;
self.log(id, &format!("compiled version {version} in {} ms", started.elapsed().as_millis()));
Ok(component)
}
/// 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))?;
/// Make `version` the current `actor.wasm`, `source/`, and `actor.json`.
pub fn install(&self, id: &str, version: u64) -> Result<()> {
self.world.install_files(id, version)?;
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` (`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> {
pub fn set_running(&self, id: &str, running: bool) -> Result<()> {
self.update_meta(id, |m| m.running = running)
}
/// Create a new actor from a `.wasm` file: test it, install it, start it.
pub async fn create_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> Result<Value> {
if !world::valid_id(id) {
bail!("invalid actor ID {id:?}: use a-z, 0-9, '-' and '_'");
}
@@ -295,31 +274,15 @@ impl Broker {
if NATIVE.contains(&id) || self.is_supervised(id) || self.world.wasm_path(id).exists() {
bail!("an actor with ID {id} already exists");
}
let version = self.world.next_version(id);
let meta = VersionMeta { version, ..meta };
self.create_version(id, version, &meta, wasm, source, start).await
let meta = VersionMeta { version: self.world.next_version(id), ..meta };
self.create_version(id, &meta, wasm, source).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)?;
let source = self.world.version_dir(id, version).join("source");
candidate::test(self, id, version, &component, &source, &self.world.data_dir(id), &self.world.public_dir(id)).await
}
.await;
let report = match tested {
Ok(r) => r,
Err(e) => {
self.reject(id, version, &e);
return Err(e);
}
};
self.install_files(id, version)?;
write_status(&self.world, id, &json!({"id": id, "version": version, "state": "created"}));
if start {
self.start_supervisor(id, version, true);
}
Ok(json!({"type": "spawned", "actor": id, "version": version, "tests": report}))
async fn create_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
let (_, tests) = self.try_version(id, meta, wasm, source).await?;
self.install(id, meta.version)?;
self.start_supervisor(id, meta.version, true);
Ok(json!({"type": "spawned", "actor": id, "version": meta.version, "tests": tests}))
}
pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> Result<Value> {
@@ -333,33 +296,19 @@ impl Broker {
}
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, version, &component, &source, &self.world.data_dir(id), &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;
let (component, tests) = self.try_version(id, meta, wasm, source).await?;
let (version, requester) = (meta.version, meta.requested_by.clone());
let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, reply }).await?;
result["tests"] = tests;
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.
/// - An `actor.wasm` equal to the installed version starts as is.
/// - A new or changed `actor.wasm` is tested where it lies. A new actor
/// is installed, a changed one is an upgrade. If its tests fail,
/// nothing changes, and the file stays for the next scan.
pub async fn scan(self: &Arc<Self>) -> Vec<Value> {
let mut report = Vec::new();
for id in self.world.actor_ids().unwrap_or_default() {
@@ -377,8 +326,6 @@ impl Broker {
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
@@ -386,59 +333,25 @@ impl Broker {
{
self.start_supervisor(id, v, meta.running);
}
if installed.is_some_and(same) {
let wasm = self.world.wasm_path(id);
if installed.is_some_and(|v| same_file(&wasm, &self.world.version_dir(id, v).join("actor.wasm"))) {
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,
version: self.world.next_version(id),
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(_) => {
let r = self.upgrade_version(id, &vmeta, &wasm, source).await;
// Put the installed version back in place. Read it again: an
// automatic rollback can have changed it meanwhile.
if r.is_err()
&& let Some(current) = self.meta(id).and_then(|m| m.version)
{
let _ = self.restore_files(id, current);
}
r
}
None => self.create_version(id, &vmeta, &wasm, source).await,
Some(_) => self.upgrade_version(id, &vmeta, &wasm, source).await,
};
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))
Some(result.unwrap_or_else(|e| json!({"type": "error", "actor": id, "reason": format!("{e:#}")})))
}
/// Handle one message to `broker`.
@@ -473,14 +386,13 @@ impl Broker {
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, meta))
};
match ty {
"spawn" => {
let (wasm, source, meta) = version_meta()?;
self.create_actor(&actor()?, &wasm, source.as_deref(), meta, true).await
self.create_actor(&actor()?, &wasm, source.as_deref(), meta).await
}
"upgrade" => {
let (wasm, source, meta) = version_meta()?;
@@ -488,23 +400,13 @@ impl Broker {
}
"rollback" => {
let id = actor()?;
let target = b["version"].as_u64();
let version = b["version"].as_u64().ok_or_else(|| anyhow!("missing \"version\""))?;
let lock = self.op_lock(&id);
let _guard = lock.lock().await;
self.control(&id, |reply| Control::Rollback { target, reply }).await
self.control(&id, |reply| Control::Rollback { version, 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 _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.metas.lock().unwrap().remove(&id);
Ok(r)
}
"subscribe" | "unsubscribe" => {
let publisher = actor()?;
let on = ty == "subscribe";
@@ -518,35 +420,21 @@ impl Broker {
Ok(json!({"type": format!("{ty}d"), "actor": publisher}))
}
"scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})),
"list" => {
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, scan, list"),
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, scan"),
}
}
/// The native `broker` actor. Requests run concurrently. Aborting this
/// task aborts the requests that still run.
/// task aborts the requests that still run. Replies (messages with
/// `reply_to`) get no answer, so that two actors never answer each other
/// forever.
pub async fn run_actor(self: Arc<Self>, mut mailbox: mpsc::Receiver<Envelope>) {
let mut requests = tokio::task::JoinSet::new();
loop {
tokio::select! {
env = mailbox.recv() => match env {
Some(env) => { requests.spawn(self.clone().handle(env)); }
Some(env) if env.reply_to.is_none() => { requests.spawn(self.clone().handle(env)); }
Some(_) => {}
None => break,
},
Some(_) = requests.join_next() => {}
@@ -570,410 +458,3 @@ pub fn truncate(s: &str, max: usize) -> String {
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>,
/// A control request that arrived during a blocked delivery.
next_control: Option<Control>,
}
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,
next_control: 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 let Some(ctl) = self.next_control.take() {
if self.handle_control(ctl).await {
break;
}
continue;
}
if self.desired_running && self.instance.is_none() && self.restart_at.is_none_or(|t| Instant::now() >= t) {
self.restart_at = None;
let _ = 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 => {}
}
}
}
/// Write a message to the instance's stdin, and route its output
/// meanwhile. An instance that does not read its stdin for
/// `STDIN_WRITE_TIMEOUT`, or while a control request waits, is killed and
/// restarted: a partly written line would corrupt the next message.
async fn deliver(&mut self, msg: Envelope) {
let Some(inst) = self.instance.as_mut() else { return };
let (broker, id, line) = (self.broker.clone(), self.id.clone(), msg.to_line());
let write = tokio::time::timeout(STDIN_WRITE_TIMEOUT, inst.send_line(&line, |l| broker.route_line(&id, &l)));
let failure = tokio::select! {
written = write => match written {
Ok(Ok(())) => {
self.last_delivered = Some(msg);
return;
}
// The instance is ending. The main loop sees its closed stdout.
Ok(Err(_)) => None,
Err(_) => Some(format!("the actor did not read stdin for {STDIN_WRITE_TIMEOUT:?}")),
},
Some(ctl) = self.control.recv() => {
self.next_control = Some(ctl);
Some("the actor did not read stdin when the broker needed it".to_string())
}
};
if msg.from != "broker" {
let reason = failure.as_deref().unwrap_or("the actor stopped");
let body = json!({"type": "undeliverable", "reason": reason, "to": self.id});
self.broker.send("broker", Some(&msg.from), None, Some(&msg.id), body);
}
if let (Some(failure), Some(inst)) = (failure, &self.instance) {
// The main loop sees the end of the instance and restarts it.
self.broker.log(&self.id, &format!("killed: {failure}"));
inst.kill();
}
}
/// Start an instance of the current version. On failure, the actor is
/// "broken" until the next start, upgrade, or rollback.
async fn start_instance(&mut self) -> Result<()> {
let started = self.try_start();
if let Err(e) = &started {
self.broker.log(&self.id, &format!("cannot start version {}: {e:#}", self.version));
self.state = "broken";
self.desired_running = false;
self.status();
return started;
}
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;
}
Ok(())
}
fn try_start(&mut self) -> Result<()> {
if self.component.is_none() {
self.component = Some(self.broker.load_version(&self.id, self.version)?);
}
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()),
],
};
self.instance = Some(self.broker.runtime.start(self.component.as_ref().unwrap(), cfg)?);
Ok(())
}
/// Stop the instance gracefully. Its last messages still get routed.
async fn stop_instance(&mut self) -> Option<Exit> {
let inst = self.instance.take()?;
let (broker, id) = (self.broker.clone(), self.id.clone());
let exit = inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |line| broker.route_line(&id, &line)).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(())
}
/// Make `version` current: compile it first, then write `actor.wasm`,
/// `source/`, and `actor.json`. If compiling fails, nothing changes.
fn install(&mut self, version: u64) -> Result<()> {
let component = self.broker.load_version(&self.id, version)?;
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.broker.update_meta(&self.id, |m| {
m.version = Some(version);
m.running = true;
})?;
self.version = version;
self.component = Some(component);
Ok(())
}
fn set_running(&self, running: bool) -> Result<()> {
self.broker.update_meta(&self.id, |m| m.running = running)
}
/// Clear the restart and crash-loop state before a swap.
fn reset_restarts(&mut self) {
self.restarts = 0;
self.restart_at = None;
self.crashes_since_swap = 0;
self.desired_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.reset_restarts();
self.previous = Some(previous);
self.swapped_at = Some(Instant::now());
self.upgrade_requester = Some(requester);
self.redeliver = self.parked.take();
if let Err(e) = self.start_instance().await {
self.rollback(Some(previous)).await?;
bail!("version {version} did not start ({e:#}); rolled back to {previous}");
}
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() && !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);
if !dir.join("actor.wasm").is_file() {
bail!("{} has no version {target}", self.id);
}
self.broker.load_version(&self.id, target)?;
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.reset_restarts();
self.previous = None;
self.swapped_at = None;
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 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 r = self.set_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 removed = r.is_ok();
let _ = reply.send(r);
if removed {
return true;
}
self.state = "stopped";
}
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 the folder of the actor into `archive/`. Nothing is deleted.
fn archive(&self) -> Result<Value> {
let archive = self.world().root.join("archive");
std::fs::create_dir_all(&archive)?;
let nanos = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos();
let dest = archive.join(format!("{}-{nanos}", self.id));
std::fs::rename(self.world().actor_dir(&self.id), &dest)?;
Ok(json!({"type": "removed", "actor": self.id, "archive": dest.display().to_string()}))
}
}
cli/src/candidate.rs
+11 -39
View File
@@ -18,8 +18,8 @@
//! also go to that recipient.
//!
//! Each file runs in a fresh instance on a copy of the actor's `/data` and
//! `/public` (empty folders for a new actor). The candidate has no HTTP access. Without test files, the
//! candidate must start and stay up for `SMOKE_TIME`.
//! `/public` (empty folders for a new actor), without HTTP. A version
//! without test files passes.
use std::path::Path;
use std::time::Duration;
@@ -33,7 +33,6 @@ use crate::fs::Mount;
use crate::runtime::{Instance, InstanceConfig};
use crate::world;
const SMOKE_TIME: Duration = Duration::from_secs(2);
const EXPECT_TIMEOUT: Duration = Duration::from_secs(10);
/// `expected` matches `actual` if every key of an expected object exists in
@@ -66,50 +65,22 @@ fn stderr_tail(path: &Path) -> String {
text[s..].to_string()
}
/// The version under test and the state that its tests start from.
/// The version under test, which starts from a copy of the actor's state.
struct Candidate<'a> {
broker: &'a Broker,
id: &'a str,
version: u64,
component: &'a Component,
data: &'a Path,
public: &'a Path,
}
pub async fn test(
broker: &Broker,
id: &str,
version: u64,
component: &Component,
source: &Path,
data: &Path,
public: &Path,
) -> Result<Value> {
let candidate = Candidate { broker, id, version, component, data, public };
/// Run every test file of `<source>/tests/`.
pub async fn test(broker: &Broker, id: &str, version: u64, component: &Component, source: Option<&Path>) -> Result<Value> {
let candidate = Candidate { broker, id, version, component };
let Some(source) = source else { return Ok(json!({"passed": []})) };
let mut files: Vec<_> = std::fs::read_dir(source.join("tests"))
.map(|rd| {
rd.filter_map(|e| e.ok())
.map(|e| e.path())
.filter(|p| p.extension().is_some_and(|x| x == "json"))
.collect()
})
.map(|rd| rd.filter_map(|e| Some(e.ok()?.path())).filter(|p| p.extension().is_some_and(|x| x == "json")).collect())
.unwrap_or_default();
files.sort();
if files.is_empty() {
let (mut inst, scratch) = candidate.start()?;
tokio::time::sleep(SMOKE_TIME).await;
if inst.is_finished() {
let exit = inst.wait().await;
bail!(
"candidate ended during the smoke test ({exit}). stderr:\n{}",
stderr_tail(&scratch.dir.join("stderr.log"))
);
}
inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |_| {}).await;
return Ok(json!({"smoke_test": "passed", "files": 0}));
}
let mut passed = Vec::new();
for file in &files {
let name = file.file_name().unwrap().to_string_lossy().to_string();
@@ -126,10 +97,11 @@ impl Candidate<'_> {
let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let dir = self.broker.world.scratch_dir().join(format!("{}-{}-{n}", self.id, std::process::id()));
let scratch = Scratch { dir: dir.clone() };
for (from, name) in [(self.data, "data"), (self.public, "public")] {
let world = &self.broker.world;
for (from, name) in [(world.data_dir(self.id), "data"), (world.public_dir(self.id), "public")] {
std::fs::create_dir_all(dir.join(name))?;
if from.is_dir() {
world::copy_tree(from, &dir.join(name))?;
world::copy_tree(&from, &dir.join(name))?;
}
}
let cfg = InstanceConfig {
cli/src/cargo_actor.rs
+11 -17
View File
@@ -8,8 +8,9 @@
//! `.wasm` file. Other actors can add other languages.
//!
//! Actors depend on the SDK with `sausage-actor = "0.1"`. The SDK is not
//! published: this actor copies the SDK of this host to `data/sdk/`, and
//! points the dependency there in the build copy of `Cargo.toml`.
//! published: this actor copies the SDK of this host to `public/sdk/` (so
//! that every actor can read it), and points the dependency there in the
//! build copy of `Cargo.toml`.
use std::path::{Path, PathBuf};
use std::sync::Arc;
@@ -44,9 +45,9 @@ pub struct Cargo {
}
impl Cargo {
/// Create the actor and install the SDK of this host into `data/sdk/`.
/// Create the actor and install the SDK of this host into `public/sdk/`.
pub fn new(world: World) -> Result<Self> {
write_dir(&SDK, &world.data_dir(ID).join("sdk"))?;
write_dir(&SDK, &world.public_dir(ID).join("sdk"))?;
Ok(Self { world })
}
@@ -58,11 +59,10 @@ impl Cargo {
}
let started = Instant::now();
let data = self.world.data_dir(ID);
let key = format!("{:016x}", fxhash(source.to_string_lossy().as_bytes()));
let build_dir = data.join("work").join(&key);
let build_dir = data.join("work");
world::replace_source(source, &build_dir)?;
let manifest = build_dir.join("Cargo.toml");
std::fs::write(&manifest, prepare_manifest(&std::fs::read_to_string(&manifest)?, Some(&data.join("sdk")))?)?;
std::fs::write(&manifest, prepare_manifest(&std::fs::read_to_string(&manifest)?, Some(&self.world.public_dir(ID).join("sdk")))?)?;
let mut cmd = Command::new("cargo");
cmd.arg("build")
@@ -99,8 +99,12 @@ impl Cargo {
}
/// The native `cargo` actor. Builds run one at a time, in order.
/// Replies (messages with `reply_to`) get no answer.
pub async fn run_actor(self, broker: Arc<Broker>, mut mailbox: mpsc::Receiver<Envelope>) {
while let Some(env) = mailbox.recv().await {
if env.reply_to.is_some() {
continue;
}
let body = self.handle(&broker, &env).await.unwrap_or_else(|e| {
json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}")})
});
@@ -124,7 +128,6 @@ impl Cargo {
"warnings": out.warnings,
}))
}
Some("shutdown") => Ok(json!({"type": "ignored"})),
other => bail!("unknown request {other:?}. Known: build"),
}
}
@@ -197,12 +200,3 @@ fn strip_progress(stderr: &str) -> String {
.collect::<Vec<_>>()
.join("\n")
}
fn fxhash(bytes: &[u8]) -> u64 {
let mut h: u64 = 0xcbf29ce484222325;
for b in bytes {
h ^= *b as u64;
h = h.wrapping_mul(0x100000001b3);
}
h
}
cli/src/console.rs
+38 -178
View File
@@ -3,205 +3,65 @@
//! - `@<actor> <json>` sends a JSON body. `@<actor> <text>` sends
//! `{"type": "text", "text": ...}`.
//! - A line without `@` goes to `agent` as text.
//! - `/deploy <actor> <folder>` builds a Rust folder with `cargo`, then spawns
//! or upgrades the actor.
//! - `/list`, `/sleep <seconds>`, `/help`, `/quit`.
//! - `/help`, `/quit`. EOF on stdin also quits.
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use std::sync::Arc;
use anyhow::{Result, anyhow, bail};
use serde_json::{Value, json};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::sync::{mpsc, oneshot};
use tokio::sync::mpsc;
use crate::broker::{Broker, Envelope};
pub const ID: &str = "console";
const HELP: &str = "\
@<actor> <json> send a JSON body to an actor
@<actor> <text> send {\"type\":\"text\",\"text\":...}
<text> send text to the agent
/deploy <actor> <dir> build a Rust folder with cargo, then spawn or upgrade
/list list actors
/sleep <seconds> wait (for scripts)
/await <actor> <secs> wait for the next message from an actor (for scripts)
/quit stop all actors and exit";
pub struct Console {
broker: Arc<Broker>,
waiters: Mutex<HashMap<String, oneshot::Sender<Envelope>>>,
seq: AtomicU64,
/// Number of messages received per sender, for `/await`.
seen: Mutex<HashMap<String, u64>>,
arrived: tokio::sync::Notify,
}
@<actor> <json> send a JSON body to an actor
@<actor> <text> send {\"type\":\"text\",\"text\":...}
<text> send text to the agent
/quit stop all actors and exit";
/// Print a message: text bodies as text, everything else as JSON.
fn show(env: &Envelope) {
let reply = if env.reply_to.is_some() { " ↩" } else { "" };
let b = &env.body;
if b["type"] == "text"
&& let Some(t) = b["text"].as_str()
{
println!("[{}{reply}] {t}", env.from);
return;
}
let mut compact = b.clone();
let mut extra = Vec::new();
if let Some(obj) = compact.as_object_mut() {
for key in ["errors", "warnings", "reason"] {
if let Some(Value::String(s)) = obj.get(key)
&& s.contains('\n')
{
extra.push((key, s.clone()));
obj.remove(key);
}
}
}
println!("[{}{reply}] {compact}", env.from);
for (k, s) in extra {
println!(" {k}:\n{}", s.lines().map(|l| format!(" {l}")).collect::<Vec<_>>().join("\n"));
match env.body["text"].as_str() {
Some(text) if env.body["type"] == "text" => println!("[{}{reply}] {text}", env.from),
_ => println!("[{}{reply}] {}", env.from, env.body),
}
}
impl Console {
pub fn new(broker: Arc<Broker>) -> Arc<Self> {
Arc::new(Self {
broker,
waiters: Mutex::new(HashMap::new()),
seq: AtomicU64::new(1),
seen: Mutex::new(HashMap::new()),
arrived: tokio::sync::Notify::new(),
})
/// Print every message that arrives for the console.
pub async fn run_mailbox(mut mailbox: mpsc::Receiver<Envelope>) {
while let Some(env) = mailbox.recv().await {
show(&env);
}
}
pub async fn run_mailbox(self: Arc<Self>, mut mailbox: mpsc::Receiver<Envelope>) {
while let Some(env) = mailbox.recv().await {
show(&env);
*self.seen.lock().unwrap().entry(env.from.clone()).or_default() += 1;
self.arrived.notify_waiters();
if let Some(r) = &env.reply_to
&& let Some(tx) = self.waiters.lock().unwrap().remove(r)
{
let _ = tx.send(env);
}
}
}
fn send(&self, to: &str, body: Value) -> String {
let own = format!("c{}", self.seq.fetch_add(1, Ordering::Relaxed));
self.broker.send(ID, Some(to), Some(&own), None, body)
}
async fn request(&self, to: &str, body: Value, timeout: Duration) -> Result<Envelope> {
let own = format!("c{}", self.seq.fetch_add(1, Ordering::Relaxed));
let (tx, rx) = oneshot::channel();
self.waiters.lock().unwrap().insert(format!("{ID}/{own}"), tx);
self.broker.send(ID, Some(to), Some(&own), None, body);
tokio::time::timeout(timeout, rx)
.await
.map_err(|_| anyhow!("no reply from {to} within {timeout:?}"))?
.map_err(|_| anyhow!("reply channel closed"))
}
async fn deploy(&self, actor: &str, dir: &str) -> Result<()> {
let path = std::path::absolute(dir)?;
let source = path.display().to_string();
println!("· building {source} …");
let built = self.request(crate::cargo_actor::ID, json!({"type": "build", "source": source}), Duration::from_secs(900)).await?;
if built.body["type"] != "built" {
bail!("build failed");
}
let exists = self.broker.meta(actor).is_some_and(|m| m.version.is_some());
let ty = if exists { "upgrade" } else { "spawn" };
println!("· {ty} {actor} …");
let body = json!({
"type": ty,
"actor": actor,
"wasm": built.body["wasm"],
"source": source,
"built_by": built.body["built_by"],
});
let r = self.request("broker", body, Duration::from_secs(300)).await?;
if r.body["type"] == "error" {
bail!("{ty} failed");
}
Ok(())
}
/// Read commands until EOF or `/quit`.
pub async fn run_input(self: Arc<Self>) {
println!("Sausage console. Type /help for commands.");
let mut lines = BufReader::new(tokio::io::stdin()).lines();
while let Ok(Some(line)) = lines.next_line().await {
let line = line.trim();
if line.is_empty() || line.starts_with('#') {
/// Read lines until EOF or `/quit`, and send them as messages.
pub async fn run_input(broker: Arc<Broker>) {
println!("Sausage console. Type /help for commands.");
let mut lines = BufReader::new(tokio::io::stdin()).lines();
while let Ok(Some(line)) = lines.next_line().await {
let line = line.trim();
let (to, body) = match line {
"" => continue,
"/quit" => break,
"/help" => {
println!("{HELP}");
continue;
}
if let Err(e) = self.command(line).await {
if e.to_string() == "quit" {
break;
}
println!("! {e:#}");
}
}
}
async fn command(&self, line: &str) -> Result<()> {
if let Some(rest) = line.strip_prefix('@') {
let (to, text) = rest.split_once(char::is_whitespace).unwrap_or((rest, ""));
let text = text.trim();
let body = match serde_json::from_str::<Value>(text) {
Ok(v @ Value::Object(_)) => v,
_ => json!({"type": "text", "text": text}),
};
self.send(to, body);
return Ok(());
}
if let Some(cmd) = line.strip_prefix('/') {
let mut parts = cmd.split_whitespace();
match parts.next() {
Some("help") => println!("{HELP}"),
Some("quit") => bail!("quit"),
Some("list") => {
self.send("broker", json!({"type": "list"}));
}
Some("sleep") => {
let secs: f64 = parts.next().and_then(|s| s.parse().ok()).unwrap_or(1.0);
tokio::time::sleep(Duration::from_secs_f64(secs)).await;
}
Some("await") => {
let from = parts.next().unwrap_or("agent").to_string();
let secs: f64 = parts.next().and_then(|s| s.parse().ok()).unwrap_or(60.0);
let count = |c: &Self| c.seen.lock().unwrap().get(&from).copied().unwrap_or(0);
let before = count(self);
let deadline = tokio::time::Instant::now() + Duration::from_secs_f64(secs);
loop {
let notified = self.arrived.notified();
tokio::pin!(notified);
notified.as_mut().enable();
if count(self) != before {
break;
}
if tokio::time::timeout_at(deadline, notified).await.is_err() {
bail!("no message from {from} within {secs} s");
}
}
}
Some("deploy") => {
let (Some(actor), Some(dir)) = (parts.next(), parts.next()) else {
bail!("usage: /deploy <actor> <dir>");
_ => match line.strip_prefix('@') {
Some(rest) => {
let (to, text) = rest.split_once(char::is_whitespace).unwrap_or((rest, ""));
let body = match serde_json::from_str::<Value>(text.trim()) {
Ok(v @ Value::Object(_)) => v,
_ => json!({"type": "text", "text": text.trim()}),
};
self.deploy(actor, dir).await?;
(to, body)
}
_ => bail!("unknown command. Type /help."),
}
return Ok(());
}
self.send("agent", json!({"type": "text", "text": line}));
Ok(())
None => ("agent", json!({"type": "text", "text": line})),
},
};
broker.send(ID, Some(to), None, None, body);
}
}
cli/src/fs.rs
+9 -47
View File
@@ -9,9 +9,7 @@
//!
//! An open of an existing file without truncate (append, in-place edit)
//! writes in place, so that appending to a log does not copy the log.
//!
//! A read-only mount of the world's `actor/` folder hides the private
//! `data/` folders of all actors (see `is_private`).
use std::collections::HashMap;
use std::path::{Component, Path, PathBuf};
@@ -42,17 +40,6 @@ struct DirInfo {
root: PathBuf,
/// Writable mount: new and truncated files commit on close.
commit: bool,
/// Read-only mount: private folders are hidden.
hide: bool,
}
/// `<id>/data` and `<id>/versions/<n>/snapshot/data`, relative to `actor/`.
fn is_private(rel: &Path) -> bool {
let c: Vec<_> = rel.components().map(|c| c.as_os_str()).collect();
c.get(1).is_some_and(|n| *n == "data")
|| (c.get(1).is_some_and(|n| *n == "versions")
&& c.get(3).is_some_and(|n| *n == "snapshot")
&& c.get(4).is_some_and(|n| *n == "data"))
}
struct Pending {
@@ -64,14 +51,12 @@ struct Pending {
pub struct FsExt {
mounts: Vec<Mount>,
dirs: HashMap<u32, DirInfo>,
/// Directory streams of hiding mounts, and the directory they list.
streams: HashMap<u32, DirInfo>,
pending: HashMap<u32, Pending>,
}
impl FsExt {
pub fn new(mounts: Vec<Mount>) -> Self {
Self { mounts, dirs: HashMap::new(), streams: HashMap::new(), pending: HashMap::new() }
Self { mounts, dirs: HashMap::new(), pending: HashMap::new() }
}
/// Commit every open write. Use this when the guest exits normally,
@@ -127,17 +112,6 @@ pub struct ActorFs<'a> {
pub ext: &'a mut FsExt,
}
impl ActorFs<'_> {
/// Whether `path` (relative to the directory `fd`) must look absent:
/// an uncommitted file, or a private folder in a hiding mount.
fn blocked(&self, fd: &Resource<types::Descriptor>, path: &str) -> bool {
is_tmp_name(path)
|| self.ext.dirs.get(&fd.rep()).is_some_and(|d| {
d.hide && resolve(d, path).is_some_and(|p| p.strip_prefix(&d.root).is_ok_and(is_private))
})
}
}
pub struct ActorFsData;
impl HasData for ActorFsData {
@@ -151,7 +125,7 @@ impl preopens::Host for ActorFs<'_> {
if let Some(m) = self.ext.mounts.iter().find(|m| &m.guest == name) {
self.ext.dirs.insert(
r.rep(),
DirInfo { host: m.host.clone(), root: m.host.clone(), commit: m.writable, hide: !m.writable },
DirInfo { host: m.host.clone(), root: m.host.clone(), commit: m.writable },
);
}
}
@@ -230,12 +204,7 @@ impl HostDescriptor for ActorFs<'_> {
&mut self,
fd: Resource<types::Descriptor>,
) -> FsResult<Resource<types::DirectoryEntryStream>> {
let dir = self.ext.dirs.get(&fd.rep()).filter(|d| d.hide).cloned();
let stream = HostDescriptor::read_directory(&mut self.inner, fd).await?;
if let Some(dir) = dir {
self.ext.streams.insert(stream.rep(), dir);
}
Ok(stream)
HostDescriptor::read_directory(&mut self.inner, fd).await
}
async fn sync(&mut self, fd: Resource<types::Descriptor>) -> FsResult<()> {
@@ -256,7 +225,7 @@ impl HostDescriptor for ActorFs<'_> {
path_flags: types::PathFlags,
path: String,
) -> FsResult<types::DescriptorStat> {
if self.blocked(&fd, &path) {
if is_tmp_name(&path) {
return Err(ErrorCode::NoEntry.into());
}
HostDescriptor::stat_at(&mut self.inner, fd, path_flags, path).await
@@ -292,7 +261,7 @@ impl HostDescriptor for ActorFs<'_> {
oflags: types::OpenFlags,
flags: types::DescriptorFlags,
) -> FsResult<Resource<types::Descriptor>> {
if self.blocked(&fd, &path) {
if is_tmp_name(&path) {
return Err(ErrorCode::NoEntry.into());
}
let parent = self.ext.dirs.get(&fd.rep()).cloned();
@@ -345,7 +314,7 @@ impl HostDescriptor for ActorFs<'_> {
}
async fn readlink_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<String> {
if self.blocked(&fd, &path) {
if is_tmp_name(&path) {
return Err(ErrorCode::NoEntry.into());
}
HostDescriptor::readlink_at(&mut self.inner, fd, path).await
@@ -441,7 +410,7 @@ impl HostDescriptor for ActorFs<'_> {
path_flags: types::PathFlags,
path: String,
) -> FsResult<types::MetadataHashValue> {
if self.blocked(&fd, &path) {
if is_tmp_name(&path) {
return Err(ErrorCode::NoEntry.into());
}
HostDescriptor::metadata_hash_at(&mut self.inner, fd, path_flags, path).await
@@ -460,21 +429,14 @@ impl HostDirectoryEntryStream for ActorFs<'_> {
Resource::new_borrow(rep),
)
.await?;
let hidden = |name: &str| {
name.starts_with(TMP_PREFIX)
|| self.ext.streams.get(&rep).is_some_and(|d| {
d.host.join(name).strip_prefix(&d.root).is_ok_and(is_private)
})
};
match entry {
Some(e) if hidden(&e.name) => continue,
Some(e) if e.name.starts_with(TMP_PREFIX) => continue,
other => return Ok(other),
}
}
}
fn drop(&mut self, stream: Resource<types::DirectoryEntryStream>) -> wasmtime::Result<()> {
self.ext.streams.remove(&stream.rep());
HostDirectoryEntryStream::drop(&mut self.inner, stream)
}
}
cli/src/lib.rs
+1
View File
@@ -9,6 +9,7 @@ pub mod cargo_actor;
pub mod console;
pub mod fs;
pub mod runtime;
pub mod supervisor;
pub mod world;
use std::path::Path;
cli/src/main.rs
+10 -28
View File
@@ -1,14 +1,13 @@
use std::io::Write;
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::{Context, Result, bail};
use clap::{Parser, Subcommand};
use include_dir::{Dir, include_dir};
use serde_json::json;
use sausage::world::{self, ActorMeta, VersionMeta, World};
use sausage::{Broker, Host, cargo_actor, console, runtime};
use sausage::world::{ActorMeta, World};
use sausage::{Host, cargo_actor, console};
/// The source of the agent actor. `init` builds it like any other actor.
static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
@@ -72,45 +71,28 @@ async fn init(dir: PathBuf) -> Result<()> {
let console = ActorMeta { subscriptions: ["agent".to_string()].into(), ..world.read_meta(console::ID) };
world.write_meta(console::ID, &console)?;
// The agent's private configuration. Only the agent can read /data.
// The agent's configuration, in its /data.
world.ensure_actor_dirs("agent")?;
let config = json!({"api": api, "base_url": base_url.trim_end_matches('/'), "model": model, "token": token});
let config_path = world.data_dir("agent").join("config.json");
std::fs::write(&config_path, serde_json::to_string_pretty(&config)?)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&config_path, std::fs::Permissions::from_mode(0o600))?;
}
let source = world.scratch_dir().join("seed").join("agent");
// The agent is an ordinary actor folder: `run` finds its `actor.wasm`,
// tests it, and installs it as version 1.
let source = world.source_dir("agent");
std::fs::create_dir_all(&source)?;
AGENT_SOURCE.extract(&source).context("extract agent source")?;
println!("Building the agent actor (the first build takes a few minutes) …");
let out = cargo_actor::Cargo::new(world.clone())?.build(&source).await.context("build agent")?;
println!("Built in {:.1} s.", out.seconds);
let broker = Broker::new(world.clone(), runtime::Runtime::new()?)?;
let meta = VersionMeta {
version: 1,
wasm: world.global_path(&out.wasm).unwrap_or_default(),
source: Some(source.display().to_string()),
built_by: Some(cargo_actor::ID.into()),
requested_by: "init".into(),
created_at: world::now(),
rejected: None,
};
broker.create_actor("agent", &out.wasm, Some(&source), meta, false).await?;
println!("World ready. Start it with: sausage run {}", dir.display());
std::fs::copy(&out.wasm, world.wasm_path("agent"))?;
println!("Built in {:.1} s. Start the world with: sausage run {}", out.seconds, dir.display());
Ok(())
}
async fn run(dir: PathBuf) -> Result<()> {
let host = Host::start(&dir).await?;
let rx = host.mailbox(console::ID);
let console = console::Console::new(host.broker.clone());
tokio::spawn(Arc::clone(&console).run_mailbox(rx));
console.run_input().await;
tokio::spawn(console::run_mailbox(host.mailbox(console::ID)));
console::run_input(host.broker.clone()).await;
println!("Stopping actors …");
host.stop().await;
Ok(())
cli/src/supervisor.rs
+296
View File
@@ -0,0 +1,296 @@
//! The supervisor of one WASM actor: it runs the instance, feeds it the
//! mailbox, routes its output, restarts it after a crash, and swaps versions.
//!
//! The broker talks to a supervisor only through `Control` requests.
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{Result, bail};
use serde_json::{Value, json};
use tokio::sync::{mpsc, oneshot};
use wasmtime::component::Component;
use crate::broker::{Broker, Envelope, SHUTDOWN_DEADLINE, shutdown_line};
use crate::runtime::{Exit, Instance, InstanceConfig};
/// A new version that crashes this often within `CRASH_WINDOW` after an
/// upgrade is rolled back automatically.
const CRASH_LIMIT: usize = 3;
const CRASH_WINDOW: Duration = Duration::from_secs(30);
pub type Reply = oneshot::Sender<Result<Value>>;
pub enum Control {
/// Swap to `version` (already in `versions/`, compiled as `component`).
Upgrade { version: u64, component: Component, requester: String, reply: Reply },
Rollback { version: u64, reply: Reply },
Stop { reply: Reply },
Start { reply: Reply },
Shutdown { reply: Reply },
}
/// The time after an upgrade in which crashes count toward a rollback.
struct Probation {
previous: u64,
since: Instant,
crashes: usize,
requester: String,
}
pub struct Supervisor {
broker: Arc<Broker>,
id: String,
version: u64,
running: bool,
mailbox: mpsc::Receiver<Envelope>,
control: mpsc::Receiver<Control>,
component: Option<Component>,
instance: Option<Instance>,
started_at: Instant,
restarts: u32,
restart_at: Option<Instant>,
probation: Option<Probation>,
/// A control request that arrived during a blocked delivery.
next_control: Option<Control>,
}
async fn next_line(instance: &mut Option<Instance>) -> Option<String> {
match instance {
Some(i) => i.stdout.recv().await,
None => std::future::pending().await,
}
}
async fn sleep_until(t: Option<Instant>) {
match t {
Some(t) => tokio::time::sleep_until(t.into()).await,
None => std::future::pending().await,
}
}
impl Supervisor {
pub fn new(
broker: Arc<Broker>,
id: String,
version: u64,
running: bool,
mailbox: mpsc::Receiver<Envelope>,
control: mpsc::Receiver<Control>,
) -> Self {
Self {
broker,
id,
version,
running,
mailbox,
control,
component: None,
instance: None,
started_at: Instant::now(),
restarts: 0,
restart_at: None,
probation: None,
next_control: None,
}
}
pub async fn run(mut self) {
loop {
if let Some(ctl) = self.next_control.take() {
if self.handle(ctl).await {
break;
}
continue;
}
if self.running && self.instance.is_none() && self.restart_at.is_none_or(|t| Instant::now() >= t) {
self.restart_at = None;
let _ = self.start().await;
}
let has_instance = self.instance.is_some();
tokio::select! {
ctl = self.control.recv() => {
let Some(ctl) = ctl else { break };
if self.handle(ctl).await {
break;
}
}
msg = self.mailbox.recv(), if has_instance => {
let Some(msg) = msg else { break };
self.deliver(msg).await;
}
line = next_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(self.restart_at), if !has_instance && self.running => {}
}
}
}
/// Write a message to the instance's stdin, and route its output
/// meanwhile. If a control request arrives before the actor reads the
/// message, the instance is killed (a partly written line would corrupt
/// the next message) and the request runs.
async fn deliver(&mut self, msg: Envelope) {
let Some(inst) = self.instance.as_mut() else { return };
let (broker, id, line) = (self.broker.clone(), self.id.clone(), msg.to_line());
let stuck = tokio::select! {
written = inst.send_line(&line, |l| broker.route_line(&id, &l)) => match written {
Ok(()) => return,
// The instance is ending. The main loop sees its closed stdout.
Err(_) => false,
},
Some(ctl) = self.control.recv() => {
self.next_control = Some(ctl);
true
}
};
if msg.from != "broker" {
let body = json!({"type": "undeliverable", "reason": "the actor stopped", "to": self.id});
self.broker.send("broker", Some(&msg.from), None, Some(&msg.id), body);
}
if stuck && let Some(inst) = &self.instance {
self.broker.log(&self.id, "killed: it did not read stdin when the broker needed it");
inst.kill();
}
}
/// Start an instance of the current version. If that fails, the actor
/// stays stopped until the next start, upgrade, or rollback.
async fn start(&mut self) -> Result<()> {
let started = self.try_start();
match &started {
Ok(()) => self.started_at = Instant::now(),
Err(e) => {
self.broker.log(&self.id, &format!("cannot start version {}: {e:#}", self.version));
self.running = false;
}
}
started
}
fn try_start(&mut self) -> Result<()> {
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()),
],
};
self.instance = Some(self.broker.runtime.start(self.component.as_ref().unwrap(), cfg)?);
Ok(())
}
/// Stop the instance gracefully. Its last messages still get routed.
async fn stop(&mut self) {
let Some(inst) = self.instance.take() else { return };
let (broker, id) = (self.broker.clone(), self.id.clone());
let exit = inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |line| broker.route_line(&id, &line)).await;
self.broker.log(&self.id, &format!("stopped: {exit}"));
}
async fn on_exit(&mut self, exit: Exit) {
self.broker.log(&self.id, &format!("instance ended unexpectedly: {exit}"));
if let Some(p) = &mut self.probation
&& p.since.elapsed() < CRASH_WINDOW
{
p.crashes += 1;
if p.crashes >= CRASH_LIMIT {
let (from, to, requester) = (self.version, p.previous, p.requester.clone());
self.broker.log(&self.id, &format!("version {from} crashed {CRASH_LIMIT} times after the upgrade; rolling back to {to}"));
let body = match self.rollback(to).await {
Ok(_) => json!({"type": "rolled_back", "actor": self.id, "from": from, "to": to,
"reason": "crash loop after upgrade", "last_exit": exit.to_string()}),
Err(e) => json!({"type": "error", "actor": self.id, "reason": format!("automatic rollback failed: {e:#}")}),
};
self.broker.send("broker", Some(&requester), None, None, body);
return;
}
}
// Restart after 0.5 s, doubling up to 30 s; a minute of uptime resets it.
if self.started_at.elapsed() > Duration::from_secs(60) {
self.restarts = 0;
}
let delay = Duration::from_millis(500 << self.restarts.min(6)).min(Duration::from_secs(30));
self.restarts += 1;
self.restart_at = Some(Instant::now() + delay);
}
/// Make `version` current. It is compiled first (unless `component` is
/// given): if that fails, nothing changes. `data/` and `public/` stay as
/// they are.
async fn install(&mut self, version: u64, component: Option<Component>) -> Result<()> {
let component = match component {
Some(c) => c,
None => self.broker.load_version(&self.id, version)?,
};
self.stop().await;
self.broker.install(&self.id, version)?;
self.version = version;
self.component = Some(component);
self.running = true;
self.restarts = 0;
self.restart_at = None;
Ok(())
}
async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result<Value> {
let previous = self.version;
self.install(version, Some(component)).await?;
self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester });
if let Err(e) = self.start().await {
self.rollback(previous).await?;
bail!("version {version} did not start ({e:#}); rolled back to {previous}");
}
Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous}))
}
async fn rollback(&mut self, version: u64) -> Result<Value> {
let from = self.version;
self.install(version, None).await?;
self.probation = None;
self.start().await?;
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version}))
}
/// Handle a control request. Returns `true` when the supervisor ends.
async fn handle(&mut self, ctl: Control) -> bool {
match ctl {
Control::Upgrade { version, component, requester, reply } => {
let _ = reply.send(self.upgrade(version, component, requester).await);
}
Control::Rollback { version, reply } => {
let _ = reply.send(self.rollback(version).await);
}
Control::Stop { reply } => {
self.running = false;
self.stop().await;
let r = self.broker.set_running(&self.id, false);
let _ = reply.send(r.map(|_| json!({"type": "stopped", "actor": self.id})));
}
Control::Start { reply } => {
self.running = true;
self.restart_at = None;
let r = self.broker.set_running(&self.id, true);
let _ = reply.send(r.map(|_| json!({"type": "started", "actor": self.id})));
}
Control::Shutdown { reply } => {
self.stop().await;
let _ = reply.send(Ok(json!({"type": "shut_down"})));
return true;
}
}
false
}
}
cli/src/world.rs
+15 -9
View File
@@ -8,11 +8,9 @@
//! actor.json version, running, subscriptions (broker)
//! source/ source of the current version
//! public/ written by the actor, read-only for others
//! data/ private state; hidden from other actors
//! status.json runtime state (broker)
//! data/ state, writable only by the actor
//! stderr.log stderr of the actor and notes of the broker
//! versions/<n>/ actor.wasm, source/, meta.json, snapshot/{data,public}
//! archive/ removed actors
//! versions/<n>/ every version that was installed: actor.wasm, source/, meta.json
//! ```
use std::collections::BTreeSet;
@@ -62,9 +60,6 @@ pub struct VersionMeta {
/// The actor that requested the version ("scan" for a file found on disk).
pub requested_by: String,
pub created_at: u64,
/// Why the broker did not install this version, if it did not.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub rejected: Option<String>,
}
pub fn valid_id(id: &str) -> bool {
@@ -126,8 +121,6 @@ impl World {
}
for id in NATIVE {
world.ensure_actor_dirs(id)?;
let status = serde_json::json!({"id": id, "kind": "native", "state": "built-in"});
write_atomic(&world.actor_dir(id).join("status.json"), serde_json::to_string_pretty(&status)?.as_bytes())?;
}
Ok(world)
}
@@ -161,6 +154,19 @@ impl World {
write_atomic(&self.actor_dir(id).join("actor.json"), serde_json::to_string_pretty(meta)?.as_bytes())
}
/// Make `actor.wasm` and `source/` a copy of `versions/<version>/`,
/// unless `actor.wasm` already is that version (it came from there).
pub fn install_files(&self, id: &str, version: u64) -> Result<()> {
let dir = self.version_dir(id, version);
self.ensure_actor_dirs(id)?;
let wasm = std::fs::read(dir.join("actor.wasm"))?;
if std::fs::read(self.wasm_path(id)).ok().as_ref() == Some(&wasm) {
return Ok(());
}
write_atomic(&self.wasm_path(id), &wasm)?;
replace_tree(&dir.join("source"), &self.source_dir(id))
}
/// Version numbers in `versions/`, ascending.
pub fn versions(&self, id: &str) -> Vec<u64> {
let mut v: Vec<u64> = std::fs::read_dir(self.versions_dir(id))
cli/tests/actors/toolbox/src/main.rs
-2
View File
@@ -146,8 +146,6 @@ fn main() {
}
Err(error) => io_error(error),
},
Some("crash") if version == "1" => panic!("asked to crash"),
Some("crash") => json!({"type": "replayed", "version": version}),
_ => json!({"type": "error", "error": "unknown message"}),
};
reply(&mut out, &message, response);
cli/tests/agent.rs
+23 -115
View File
@@ -9,7 +9,7 @@ use std::process::Command;
use std::sync::OnceLock;
use std::time::Duration;
use common::{create_world, receive_replies, receive_reply, request, source_with_test, start};
use common::{create_world, receive_message, receive_reply, request, source_with_test, start};
use sausage::{Envelope, Host, World};
use serde_json::{Value, json};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -319,6 +319,13 @@ async fn start_agent(
spawned.body["type"], "spawned",
"agent spawn failed: {spawned:?}"
);
let subscribed = request(
&host,
&mut mailbox,
json!({"type": "subscribe", "actor": "agent"}),
)
.await;
assert_eq!(subscribed.body["type"], "subscribed");
(world, host, mailbox)
}
@@ -332,11 +339,12 @@ fn send_text(host: &Host, text: &str) -> String {
)
}
async fn assert_answer(mailbox: &mut mpsc::Receiver<Envelope>, id: &str, text: &str) {
let reply = receive_reply(mailbox, id, Duration::from_secs(20)).await;
assert_eq!(reply.to, "console");
assert_eq!(reply.reply_to.as_deref(), Some(id));
assert_eq!(reply.body, json!({"type": "text", "text": text}));
async fn assert_published_text(mailbox: &mut mpsc::Receiver<Envelope>, text: &str) {
let message = receive_message(mailbox, Duration::from_secs(20)).await;
assert_eq!(message.from, "agent");
assert_eq!(message.to, "console");
assert_eq!(message.reply_to, None);
assert_eq!(message.body, json!({"type": "text", "text": text}));
}
fn history(world: &World) -> Vec<Value> {
@@ -439,9 +447,9 @@ async fn text_answers_use_both_inference_apis() {
.await;
let (_, host, mut mailbox) =
start_agent(api, &fake, &format!("agent-text-{}", api.name())).await;
let id = send_text(&host, "say hello");
send_text(&host, "say hello");
api.check_request(&fake.next_request().await);
assert_answer(&mut mailbox, &id, "hello from the fake model").await;
assert_published_text(&mut mailbox, "hello from the fake model").await;
host.stop().await;
}
}
@@ -455,7 +463,7 @@ async fn tool_calls_pair_with_results_in_both_inference_apis() {
.await;
let (world, host, mut mailbox) =
start_agent(api, &fake, &format!("agent-tools-{}", api.name())).await;
let id = send_text(&host, "list actors");
send_text(&host, "list actors");
api.check_request(&fake.next_request().await);
fake.answer(api.text_response("the actor list is ready"))
.await;
@@ -470,8 +478,9 @@ async fn tool_calls_pair_with_results_in_both_inference_apis() {
.iter()
.any(|entry| entry["name"] == "agent")
);
assert_answer(&mut mailbox, &id, "the actor list is ready").await;
assert_published_text(&mut mailbox, "the actor list is ready").await;
host.stop().await;
let history = history(&world);
assert_eq!(
history
@@ -482,7 +491,6 @@ async fn tool_calls_pair_with_results_in_both_inference_apis() {
);
assert_eq!(history[1]["id"], "list-1");
assert_eq!(history[2]["id"], "list-1");
host.stop().await;
}
}
@@ -513,8 +521,8 @@ async fn restart_resumes_a_turn_with_completed_tool_history() {
serde_json::from_slice::<Value>(
&std::fs::read(world.data_dir("agent").join("turn.json")).unwrap()
)
.unwrap()["active"]["id"],
id
.unwrap(),
json!(true)
);
let _ = release.send(());
fake.answer(api.text_response("resumed answer")).await;
@@ -528,22 +536,9 @@ async fn restart_resumes_a_turn_with_completed_tool_history() {
let resumed = fake.next_request().await;
api.check_request(&resumed);
assert_tool_pair(&resumed, api, "list-1", "list_dir", &args);
let replies = receive_replies(
&mut mailbox,
&[start_id.clone(), id.clone()],
Duration::from_secs(20),
)
.await;
let started = replies
.iter()
.find(|reply| reply.reply_to.as_deref() == Some(start_id.as_str()))
.unwrap();
let started = receive_reply(&mut mailbox, &start_id, Duration::from_secs(20)).await;
assert_eq!(started.body["type"], "started");
let answer = replies
.iter()
.find(|reply| reply.reply_to.as_deref() == Some(id.as_str()))
.unwrap();
assert_eq!(answer.body["text"], "resumed answer");
assert_published_text(&mut mailbox, "resumed answer").await;
let history = history(&world);
assert_history_tools_have_results(&history);
@@ -558,90 +553,3 @@ async fn restart_resumes_a_turn_with_completed_tool_history() {
host.stop().await;
}
}
#[tokio::test]
async fn queued_text_resumes_after_the_active_turn_across_restart() {
let wait_args = json!({"seconds": 10});
for api in [Api::Anthropic] {
let mut fake = FakeServer::start().await;
fake.answer(api.tool_response("wait-1", "wait_for_messages", &wait_args))
.await;
let (world, host, mut mailbox) =
start_agent(api, &fake, &format!("agent-queue-{}", api.name())).await;
let first_id = send_text(&host, "first turn");
api.check_request(&fake.next_request().await);
let release = fake.hold(api.text_response("discarded response")).await;
let second_id = send_text(&host, "second turn");
let second_request = fake.next_request().await;
api.check_request(&second_request);
let output = assert_tool_pair(
&second_request,
api,
"wait-1",
"wait_for_messages",
&wait_args,
);
let queued: Value = serde_json::from_str(&output).unwrap();
assert!(
queued["messages"]
.as_array()
.unwrap()
.iter()
.any(|message| message["message"]["id"] == second_id)
);
let turn: Value = serde_json::from_slice(
&std::fs::read(world.data_dir("agent").join("turn.json")).unwrap(),
)
.unwrap();
assert_eq!(turn["queue"][0]["id"], second_id);
let stopped = request(
&host,
&mut mailbox,
json!({"type": "stop", "actor": "agent"}),
)
.await;
assert_eq!(stopped.body["type"], "stopped");
let _ = release.send(());
fake.answer(api.text_response("first answer")).await;
fake.answer(api.text_response("second answer")).await;
let start_id = host.broker.send(
"console",
Some("broker"),
None,
None,
json!({"type": "start", "actor": "agent"}),
);
let resumed_first = fake.next_request().await;
api.check_request(&resumed_first);
assert!(resumed_first.body.to_string().contains("first turn"));
let next_turn = fake.next_request().await;
api.check_request(&next_turn);
assert!(next_turn.body.to_string().contains("second turn"));
let replies = receive_replies(
&mut mailbox,
&[start_id.clone(), first_id.clone(), second_id.clone()],
Duration::from_secs(20),
)
.await;
let started = replies
.iter()
.find(|reply| reply.reply_to.as_deref() == Some(start_id.as_str()))
.unwrap();
assert_eq!(started.body["type"], "started");
let first_position = replies
.iter()
.position(|reply| reply.reply_to.as_deref() == Some(first_id.as_str()))
.unwrap();
let second_position = replies
.iter()
.position(|reply| reply.reply_to.as_deref() == Some(second_id.as_str()))
.unwrap();
assert!(first_position < second_position);
assert_eq!(replies[first_position].body["text"], "first answer");
assert_eq!(replies[second_position].body["text"], "second answer");
assert_history_tools_have_results(&history(&world));
host.stop().await;
}
}
cli/tests/common/mod.rs
-23
View File
@@ -246,26 +246,3 @@ pub async fn upgrade_actor(
)
.await
}
pub async fn wait_for_parked(world: &World, actor: &str, message_id: &str) {
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let status_path = world.actor_dir(actor).join("status.json");
let parked_id = std::fs::read_to_string(status_path)
.ok()
.and_then(|text| serde_json::from_str::<Value>(&text).ok())
.and_then(|status| {
status
.pointer("/parked_message/id")
.and_then(Value::as_str)
.map(str::to_string)
});
if parked_id.as_deref() == Some(message_id) {
return;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("actor did not park the crashed message");
}
cli/tests/host.rs
+75 -210
View File
@@ -1,6 +1,6 @@
mod common;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use std::time::Duration;
use common::*;
use sausage::fs::TMP_PREFIX;
@@ -48,13 +48,6 @@ async fn receive_shutdown_messages(
.expect("shutdown output or lifecycle reply did not arrive");
}
fn epoch_second() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs()
}
#[tokio::test]
async fn routes_replies_publishes_and_reports_undeliverable() {
let world = create_world("routing");
@@ -112,6 +105,29 @@ async fn routes_replies_publishes_and_reports_undeliverable() {
host.stop().await;
}
#[tokio::test]
async fn broker_does_not_answer_messages_with_reply_to() {
let world = create_world("broker-reply-to");
let (host, mut mailbox) = start(&world).await;
let scanned = request(&host, &mut mailbox, json!({"type": "scan"})).await;
assert_eq!(scanned.body["type"], "scanned");
host.broker.send(
"console",
Some("broker"),
None,
Some("upstream-message"),
json!({"type": "scan"}),
);
assert!(
tokio::time::timeout(Duration::from_secs(1), mailbox.recv())
.await
.is_err(),
"broker replied to a message that already had reply_to"
);
host.stop().await;
}
#[tokio::test]
async fn filesystem_permissions_and_public_sharing() {
let (_, host, mut mailbox) = filesystem_world("filesystem-permissions").await;
@@ -148,6 +164,47 @@ async fn filesystem_permissions_and_public_sharing() {
host.stop().await;
}
#[tokio::test]
async fn actors_can_read_but_not_write_other_actor_data() {
let (_, host, mut mailbox) = filesystem_world("actor-data-sharing").await;
ask_type(
&host,
&mut mailbox,
"writer",
json!({"type": "write", "path": "/data/x", "text": "shared"}),
"ok",
)
.await;
let shared = ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "read", "path": "/actor/writer/data/x"}),
"content",
)
.await;
assert_eq!(shared.body["text"], "shared");
ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "write", "path": "/actor/writer/data/x", "text": "changed"}),
"error",
)
.await;
let unchanged = ask_type(
&host,
&mut mailbox,
"writer",
json!({"type": "read", "path": "/data/x"}),
"content",
)
.await;
assert_eq!(unchanged.body["text"], "shared");
host.stop().await;
}
#[tokio::test]
async fn filesystem_hides_new_and_truncated_writes_until_close() {
let (_, host, mut mailbox) = filesystem_world("filesystem-commit").await;
@@ -274,42 +331,6 @@ async fn filesystem_append_is_visible_before_close() {
host.stop().await;
}
#[tokio::test]
async fn filesystem_crash_discards_an_open_truncated_write() {
let (world, host, mut mailbox) = filesystem_world("filesystem-crash").await;
ask(
&host,
&mut mailbox,
"writer",
json!({"type": "write", "path": "/public/crash.txt", "text": "old"}),
)
.await;
ask(
&host,
&mut mailbox,
"writer",
json!({"type": "open", "path": "/public/crash.txt", "text": "uncommitted"}),
)
.await;
let crash_id = host.broker.send(
"console",
Some("writer"),
None,
None,
json!({"type": "crash"}),
);
wait_for_parked(&world, "writer", &crash_id).await;
let after_crash = ask(
&host,
&mut mailbox,
"reader",
json!({"type": "read", "path": "/actor/writer/public/crash.txt"}),
)
.await;
assert_eq!(after_crash.body["text"], "old");
host.stop().await;
}
#[tokio::test]
async fn filesystem_rejects_paths_that_escape_data() {
let (_, host, mut mailbox) = filesystem_world("filesystem-path-escape").await;
@@ -351,6 +372,8 @@ async fn upgrades_test_candidates_keep_old_versions_and_queue_messages_in_order(
assert_eq!(failed.body["type"], "error");
let old_version = ask(&host, &mut mailbox, "queue", json!({"type": "version"})).await;
assert_eq!(old_version.body["version"], "1");
// A version that fails its tests never enters versions/.
assert!(!world.actor_dir("queue").join("versions/2").exists());
let upgrade_id = host.broker.send(
"console",
@@ -408,72 +431,14 @@ async fn upgrades_test_candidates_keep_old_versions_and_queue_messages_in_order(
assert_eq!(envelope.from, "queue");
assert_eq!(envelope.body["type"], "ordered");
assert_eq!(envelope.body["n"], json!(index + 1));
assert_eq!(envelope.body["version"], "3");
assert_eq!(envelope.body["version"], "2");
}
host.stop().await;
}
#[tokio::test]
async fn redelivers_a_crashed_message_after_the_next_upgrade() {
let world = create_world("redelivery");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let source = passing_source(&world, "source");
assert_eq!(
spawn_actor(&host, &mut mailbox, "parked", &wasm, &source)
.await
.body["type"],
"spawned"
);
let crash_id = host.broker.send(
"console",
Some("parked"),
None,
None,
json!({"type": "crash"}),
);
wait_for_parked(&world, "parked", &crash_id).await;
assert!(matches!(
mailbox.try_recv(),
Err(tokio::sync::mpsc::error::TryRecvError::Empty)
));
let upgrade_id = host.broker.send(
"console",
Some("broker"),
None,
None,
json!({"type": "upgrade", "actor": "parked", "wasm": wasm.display().to_string(), "source": source.display().to_string()}),
);
let mut upgraded = None;
let mut replayed = None;
tokio::time::timeout(Duration::from_secs(15), async {
while upgraded.is_none() || replayed.is_none() {
let envelope = mailbox.recv().await.expect("console mailbox closed");
if envelope.reply_to.as_deref() == Some(upgrade_id.as_str()) {
upgraded = Some(envelope);
} else if envelope.reply_to.as_deref() == Some(crash_id.as_str()) {
replayed = Some(envelope);
} else {
panic!("unexpected message during parked delivery: {envelope:?}");
}
}
})
.await
.expect("upgrade did not redeliver the parked message");
assert_eq!(upgraded.unwrap().body["type"], "upgraded");
assert_eq!(
replayed.unwrap().body,
json!({"type": "replayed", "version": "2"})
);
host.stop().await;
}
#[tokio::test]
async fn rolls_back_a_crash_loop_and_restores_its_data_snapshot() {
async fn rolls_back_a_crash_loop_to_the_previous_version_without_restoring_data() {
let world = create_world("crash-loop");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
@@ -505,7 +470,7 @@ async fn rolls_back_a_crash_loop_and_restores_its_data_snapshot() {
);
let mut upgraded: Option<Value> = None;
let mut rolled_back: Option<Value> = None;
tokio::time::timeout(Duration::from_secs(20), async {
tokio::time::timeout(Duration::from_secs(35), async {
while upgraded.is_none() || rolled_back.is_none() {
let envelope = mailbox.recv().await.expect("console mailbox closed");
let kind = envelope.body["type"].as_str().map(str::to_string);
@@ -530,20 +495,20 @@ async fn rolls_back_a_crash_loop_and_restores_its_data_snapshot() {
let version = ask(&host, &mut mailbox, "crashloop", json!({"type": "version"})).await;
assert_eq!(version.body["version"], "1");
let restored = ask(
let current = ask(
&host,
&mut mailbox,
"crashloop",
json!({"type": "read", "path": "/data/state"}),
)
.await;
assert_eq!(restored.body["text"], "stable");
assert_eq!(current.body["text"], "bad-v2");
host.stop().await;
}
#[tokio::test]
async fn explicit_rollback_restores_the_previous_version_and_snapshot() {
async fn explicit_rollback_keeps_current_data() {
let world = create_world("explicit-rollback");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
@@ -581,17 +546,16 @@ async fn explicit_rollback_restores_the_previous_version_and_snapshot() {
assert_eq!(rolled_back.body["type"], "rolled_back");
assert_eq!(rolled_back.body["from"], 2);
assert_eq!(rolled_back.body["to"], 1);
assert_eq!(rolled_back.body["restored_state"], true);
let version = ask(&host, &mut mailbox, "recover", json!({"type": "version"})).await;
assert_eq!(version.body["version"], "1");
let restored = ask(
let current = ask(
&host,
&mut mailbox,
"recover",
json!({"type": "read", "path": "/data/state"}),
)
.await;
assert_eq!(restored.body["text"], "version-one");
assert_eq!(current.body["text"], "version-two");
host.stop().await;
}
@@ -721,37 +685,6 @@ async fn concurrent_spawns_of_one_id_leave_one_live_instance() {
host.stop().await;
}
#[tokio::test]
async fn rollback_preserves_target_and_git_files_in_data_and_public() {
let world = create_world("snapshot-special-directories");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let source = passing_source(&world, "source");
spawn_type(&host, &mut mailbox, "snapshot", &wasm, &source).await;
let files = [
("/data/target", "/data/target/x", "data-target"),
("/data/.git", "/data/.git/x", "data-git"),
("/public/target", "/public/target/x", "public-target"),
("/public/.git", "/public/.git/x", "public-git"),
];
for (dir, path, text) in files {
let mkdir = json!({"type": "mkdir", "path": dir});
ask_type(&host, &mut mailbox, "snapshot", mkdir, "ok").await;
let write = json!({"type": "write", "path": path, "text": text});
ask_type(&host, &mut mailbox, "snapshot", write, "ok").await;
}
let upgrade = json!({"type": "upgrade", "actor": "snapshot", "wasm": wasm.display().to_string(), "source": source.display().to_string()});
ask_type(&host, &mut mailbox, "broker", upgrade, "upgraded").await;
let rollback = json!({"type": "rollback", "actor": "snapshot", "version": 1});
ask_type(&host, &mut mailbox, "broker", rollback, "rolled_back").await;
for (_, path, text) in files {
let read = json!({"type": "read", "path": path});
let reply = ask_type(&host, &mut mailbox, "snapshot", read, "content").await;
assert_eq!(reply.body["text"], text);
}
host.stop().await;
}
#[tokio::test]
async fn stop_finishes_when_actor_stdin_is_full() {
let world = create_world("full-stdin-stop");
@@ -819,74 +752,6 @@ async fn shutdown_reply_and_publish_survive_stop_and_upgrade() {
host.stop().await;
}
#[tokio::test]
async fn rapid_remove_recreate_remove_keeps_actor_manageable() {
let world = create_world("rapid-remove");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let source = passing_source(&world, "source");
spawn_type(&host, &mut mailbox, "rapid", &wasm, &source).await;
let stop = json!({"type": "stop", "actor": "rapid"});
ask_type(&host, &mut mailbox, "broker", stop, "stopped").await;
while SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.subsec_millis()
>= 100
{
tokio::time::sleep(Duration::from_millis(10)).await;
}
let archive_second = epoch_second();
let remove = json!({"type": "remove", "actor": "rapid"});
let first = ask_type(&host, &mut mailbox, "broker", remove.clone(), "removed").await;
assert_eq!(
epoch_second(),
archive_second,
"first remove crossed an epoch-second boundary"
);
// Restore the stopped actor from its archive to avoid a candidate smoke test.
sausage::world::copy_tree(
first.body["archive"]
.as_str()
.map(std::path::Path::new)
.unwrap(),
&world.actor_dir("rapid"),
)
.unwrap();
let scan = json!({"type": "scan"});
ask_type(&host, &mut mailbox, "broker", scan.clone(), "scanned").await;
let second = ask_type(&host, &mut mailbox, "broker", remove.clone(), "removed").await;
assert_eq!(
epoch_second(),
archive_second,
"remove and recreate crossed an epoch-second boundary"
);
sausage::world::copy_tree(
second.body["archive"]
.as_str()
.map(std::path::Path::new)
.unwrap(),
&world.actor_dir("rapid"),
)
.unwrap();
ask_type(&host, &mut mailbox, "broker", scan, "scanned").await;
let start = json!({"type": "start", "actor": "rapid"});
ask_type(&host, &mut mailbox, "broker", start, "started").await;
let echo = ask_type(
&host,
&mut mailbox,
"rapid",
json!({"type": "echo", "value": "managed"}),
"echo",
)
.await;
assert_eq!(echo.body["value"], "managed");
host.stop().await;
}
#[tokio::test]
async fn close_commits_open_file_after_parent_directory_rename() {
let (_, host, mut mailbox) = filesystem_world("rename-open-file").await;