agent-pool-9
0995824c01
ci / check (push) Failing after 7m36s
The host watches the whole world folder recursively with the notify crate
(any writer counts, .sausage-tmp-* skipped, ~100 ms coalescing per path).
A new or changed actor/<id>/actor.wasm (or new folder actor/<id>) is
booted as at start once quiet: new = spawn (tested first), changed =
upgrade. The broker `scan` request is gone; the boot pass stays, plus a
full pass when the OS watcher overflows.
New broker requests watch/unwatch {path} (saved in actor.json `watches`)
deliver fs_changed {path, op: write|create|remove|rename, to?} from `fs`,
not routed, dropped when the mailbox is full; overflow sends every watcher
one fs_changed {path:"/actor", op:"rescan"}.
Card #91 (multi-host step 1).
87 lines
2.9 KiB
Rust
87 lines
2.9 KiB
Rust
//! Sausage: grow a system of WASI actors by talking to it.
|
|
//!
|
|
//! The host runs the broker, the native actors (`broker`, `cargo`, and the
|
|
//! console), one supervisor per WASM actor, and the file watcher of the
|
|
//! world folder (`watch.rs`). See `PROTOCOL.md`.
|
|
|
|
pub mod broker;
|
|
pub mod candidate;
|
|
pub mod cargo_actor;
|
|
pub mod console;
|
|
pub mod runtime;
|
|
pub mod supervisor;
|
|
pub mod vfs;
|
|
pub mod watch;
|
|
pub mod world;
|
|
|
|
use std::path::Path;
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::{Result, bail};
|
|
use tokio::sync::mpsc;
|
|
|
|
pub use broker::{Broker, Envelope};
|
|
pub use world::World;
|
|
|
|
/// The contract between the host and the actors. The host publishes it at
|
|
/// `/actor/broker/public/PROTOCOL.md`, so that every actor can read it.
|
|
const PROTOCOL: &str = include_str!("../../PROTOCOL.md");
|
|
|
|
/// A running world.
|
|
pub struct Host {
|
|
pub world: World,
|
|
pub broker: Arc<Broker>,
|
|
native: Vec<tokio::task::JoinHandle<()>>,
|
|
watcher: Option<watch::Watcher>,
|
|
_lock: std::fs::File,
|
|
}
|
|
|
|
impl Host {
|
|
/// Start the broker, `cargo`, the file watcher, and every registered
|
|
/// actor of the world at `root`. Only one host can run a world at a time.
|
|
pub async fn start(root: &Path) -> Result<Host> {
|
|
let world = World::new(std::path::absolute(root)?);
|
|
if !world.exists() {
|
|
bail!("{} is not a Sausage world. Create one with: sausage init", root.display());
|
|
}
|
|
let lock = std::fs::File::create(world.root.join(".lock"))?;
|
|
if lock.try_lock().is_err() {
|
|
bail!("another sausage process runs {}", root.display());
|
|
}
|
|
let broker = Broker::new(world.clone(), runtime::Runtime::new()?)?;
|
|
world::write_atomic(&world.public_dir("broker").join("PROTOCOL.md"), PROTOCOL.as_bytes())?;
|
|
|
|
let cargo = cargo_actor::Cargo::new(world.clone())?;
|
|
let native = vec![
|
|
tokio::spawn(broker.clone().run_actor(broker.open_mailbox("broker"))),
|
|
tokio::spawn(cargo.run_actor(broker.clone(), broker.open_mailbox(cargo_actor::ID))),
|
|
];
|
|
// Watch first, so that no change during the boot pass is lost.
|
|
let watcher = watch::start(broker.clone())
|
|
.inspect_err(|e| broker.host_error(&format!("no file watching: {e}")))
|
|
.ok();
|
|
for change in broker.scan().await {
|
|
eprintln!("sausage: {change}");
|
|
}
|
|
Ok(Host { world, broker, native, watcher, _lock: lock })
|
|
}
|
|
|
|
/// Register a native actor with the ID `id` and return its mailbox.
|
|
pub fn mailbox(&self, id: &str) -> mpsc::Receiver<Envelope> {
|
|
self.broker.open_mailbox(id)
|
|
}
|
|
|
|
/// Stop the file watcher, the broker and `cargo` (with the requests they
|
|
/// still run), then every actor, gracefully.
|
|
pub async fn stop(mut self) {
|
|
drop(self.watcher.take());
|
|
for task in &self.native {
|
|
task.abort();
|
|
}
|
|
for task in self.native {
|
|
let _ = task.await;
|
|
}
|
|
self.broker.shutdown_all().await;
|
|
}
|
|
}
|