diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..759f3c5 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,39 @@ +# Working on Sausage + +Sausage is meant to be read and changed by agents, including its own agent. +Keep it that way. + +## Priorities + +1. **Correct.** Every behavior in `PROTOCOL.md` has a test in `cli/tests/`. +2. **Simple and short.** Prefer the obvious code. No abstraction for one + caller. No configuration that nobody asked for. +3. **Readable.** A new reader must understand a file without the others. + Each module starts with a doc comment that says what it owns. +4. **Efficient** on the paths that run per message. +5. **Raw.** Sausage has no permissions, sandboxing beyond WASI, limits, or + authentication. Do not add them to the core. People who want them can + build them as actors or as a fork. + +## Layout + +- `cli/`: the host (`sausage` binary and library). `broker.rs` routes and + supervises, `fs.rs` is the filesystem layer, `runtime.rs` runs Wasmtime, + `candidate.rs` tests new versions, `cargo_actor.rs` builds Rust actors. +- `sdk/`: `sausage-actor`, the SDK for Rust actors. Never published: the + `cargo` actor points `sausage-actor = "0.1"` at its local copy. +- `agent/`: the agent actor. `sausage init` embeds and builds it. +- `examples/`: small actors. They are not workspace members. +- `PROTOCOL.md`: the contract between the host and the actors. Change it in + the same commit as the behavior. + +## Checks + +```sh +nix develop -c cargo test -p sausage -p sausage-actor +nix develop -c cargo build --release -p agent --target wasm32-wasip2 +nix develop -c cargo clippy --workspace +``` + +`nix run -- init` and `nix run -- run` must keep working without other +setup. diff --git a/Cargo.lock b/Cargo.lock index 8fced98..58e4cee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1593,6 +1593,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "toml_edit", "wasmtime", "wasmtime-wasi", "wasmtime-wasi-http", @@ -1939,6 +1940,19 @@ dependencies = [ "serde_core", ] +[[package]] +name = "toml_edit" +version = "0.25.17+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3641d5bbb5349a79e1020a242d251efbc546ad8048d133958323ce9c40a9c9c" +dependencies = [ + "indexmap", + "toml_datetime", + "toml_parser", + "toml_writer", + "winnow", +] + [[package]] name = "toml_parser" version = "1.1.5+spec-1.1.0" @@ -2801,6 +2815,9 @@ name = "winnow" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" +dependencies = [ + "memchr", +] [[package]] name = "winx" diff --git a/PROTOCOL.md b/PROTOCOL.md new file mode 100644 index 0000000..312c692 --- /dev/null +++ b/PROTOCOL.md @@ -0,0 +1,138 @@ +# The Sausage protocol + +This document is the contract between the host and the actors. An actor in +any language that follows it works in Sausage. + +## Actors + +An actor is a WASI 0.2 command (`wasm32-wasip2`): a program with `main`. The +host runs one instance per actor. An instance can run forever. + +The instance gets: + +- **stdin:** one message per line, from the broker +- **stdout:** one message per line, to the broker +- **stderr:** free text, appended to `/actor//stderr.log` +- **environment:** `SAUSAGE_ACTOR` (its ID), `SAUSAGE_VERSION` (its version), + and `SAUSAGE_CANDIDATE=1` during a test run +- **outgoing HTTP and HTTPS** through `wasi:http` (no raw sockets) +- **the filesystem** described below + +## Messages + +A line on **stdin**: + +```json +{"id": "m-17", "from": "console", "reply_to": "agent/a5", "body": {"type": "text", "text": "hi"}} +``` + +A line on **stdout**: + +```json +{"to": "console", "id": "a6", "reply_to": "m-17", "body": {"type": "text", "text": "hello"}} +``` + +- `body` is required. By convention it is an object with a `"type"`. +- `to` is the recipient. Without `to`, the broker publishes the message to + every subscriber of the sender. +- `id` is optional. If the actor sets it, the message ID is `"/"`. + Otherwise, the broker assigns `m-`. The broker always sets `from`. +- `reply_to` marks a reply. There is no request/response: a reply is a new + message whose `reply_to` is the ID of the message it answers. +- Delivery is at most once. Each actor has a mailbox of 1,000 messages. The + mailbox keeps messages while the actor restarts or is upgraded. +- If the broker cannot deliver a message, the sender gets + `{"type": "undeliverable", "reason": "...", "to": "..."}` from `broker`. + +Before the broker stops an instance, it sends +`{"id": "broker/shutdown", "from": "broker", "body": {"type": "shutdown"}}` +and then closes stdin. After 5 seconds, it kills the instance. + +## Filesystem + +Each instance sees three folders: + +| Path | Access | Contents | +| --- | --- | --- | +| `/data` | read-write | private state of the actor | +| `/public` | read-write | files that the actor publishes | +| `/actor` | read-only | the `actor/` folder of the world | + +`/actor//` contains `actor.wasm`, `actor.json`, `source/`, `public/`, +`status.json`, `stderr.log`, and `versions//` of every actor. The `data/` +folders (and the data snapshots in `versions/`) are hidden. + +A new or truncated file becomes visible when the writer closes it. Before +that, readers see the old file or no file. If the instance crashes, the write +is lost. A write to an existing file without truncate (append, edit) goes in +place. + +Send large or binary data as a path under `/actor//public/`, not inside a +message. + +## The broker + +Requests to `broker` (the reply has the same `reply_to`): + +| `type` | Fields | Effect | +| --- | --- | --- | +| `spawn` | `actor`, `wasm`, `source`?, `built_by`? | test, then start a new actor | +| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version | +| `rollback` | `actor`, `version`? | go back to an older version and its data | +| `stop`, `start` | `actor` | stop or start an actor | +| `remove` | `actor` | move the actor's folder to `archive/` | +| `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes | +| `scan` | | boot new or changed `actor.wasm` files | +| `list` | | all actors and their state | + +Paths: `/actor/...`, `/public/...` (the sender's), or `/data/...` (the +sender's). + +**Versions.** `versions//` keeps every version that the broker saw: +`actor.wasm`, `source/`, and `meta.json`. A version that the broker refused +has `"rejected"` in its `meta.json`. + +**Tests.** Before a spawn or an upgrade, the broker runs the candidate on a +copy of the actor's `/data` and `/public`, without HTTP. The tests are JSON +files in `source/tests/*.json`: + +```json +{ + "description": "add, then list", + "steps": [ + {"send": {"type": "add", "text": "milk"}, "expect": {"type": "added"}}, + {"send": {"type": "list"}, "expect": [{"type": "notes"}]} + ] +} +``` + +An `expect` entry matches when it is a subset of the reply body. A step can +also have `from` (the sender, default `test`) and `to` (the expected +recipient). Each file runs in a fresh instance. Without test files, the +candidate must stay up for 2 seconds. + +**Swap.** The broker holds the mailbox, stops the old instance, saves a +snapshot of `/data` and `/public` in the old version's folder, installs the new +files, and starts the new instance. + +**Crashes.** The broker restarts a crashed actor (0.5 s, then doubling, at +most 30 s). It keeps the message that the actor got just before the crash, +and delivers it again after the next upgrade. A new version that crashes 3 +times within 30 seconds after an upgrade goes back to the previous version, +with its data. The actor that asked for the upgrade gets `rolled_back`. + +**The folder is the truth.** At start, and on `scan`, the broker boots every +`actor//actor.wasm`. If the file differs from the installed version, the +broker copies it to `versions//` and upgrades to it. If the upgrade fails, +the installed files come back. + +## The `cargo` actor + +```json +{"type": "build", "source": "/actor/agent/public/drafts/notes"} +``` + +The reply is `{"type": "built", "wasm": "/actor/cargo/public/builds//.wasm", ...}` +or `{"type": "build_failed", "errors": "..."}`. A Rust actor depends on the SDK +with `sausage-actor = "0.1"`. `cargo` points that dependency at the SDK of the +host. diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 6b24ed7..29943cf 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -18,6 +18,7 @@ include_dir = "0.7" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["full"] } +toml_edit = "0.25" wasmtime = "49" wasmtime-wasi = "49" wasmtime-wasi-http = "49" diff --git a/cli/src/broker.rs b/cli/src/broker.rs index f185e68..be929d1 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -1,10 +1,12 @@ //! The broker: message routing, mailboxes, subscriptions, and the lifecycle of -//! WASM actors (spawn, upgrade, rollback, stop, start, remove). +//! WASM actors (spawn, upgrade, rollback, stop, start, remove, scan). //! //! The broker accepts only `.wasm` files. It does not know any source language. +//! The world folder is the source of truth: `scan` boots every +//! `actor//actor.wasm` and treats a changed file as an upgrade. -use std::collections::{BTreeSet, HashMap}; -use std::path::{Path, PathBuf}; +use std::collections::{BTreeMap, HashMap}; +use std::path::Path; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -17,7 +19,7 @@ use wasmtime::component::Component; use crate::candidate; use crate::fs::Mount; use crate::runtime::{Exit, Instance, InstanceConfig, Runtime}; -use crate::world::{self, ActorRecord, NATIVE, Registry, VersionMeta, World}; +use crate::world::{self, ActorMeta, NATIVE, VersionMeta, World}; pub const MAILBOX_SIZE: usize = 1000; pub const SHUTDOWN_DEADLINE: Duration = Duration::from_secs(5); @@ -51,13 +53,15 @@ pub fn shutdown_line() -> String { json!({"id": "broker/shutdown", "from": "broker", "body": {"type": "shutdown"}}).to_string() } +type Reply = oneshot::Sender>; + enum Control { - Upgrade { version: u64, requester: String, reply: oneshot::Sender> }, - Rollback { target: Option, reply: oneshot::Sender> }, - Stop { reply: oneshot::Sender> }, - Start { reply: oneshot::Sender> }, - Remove { reply: oneshot::Sender> }, - Shutdown { reply: oneshot::Sender> }, + Upgrade { version: u64, requester: String, reply: Reply }, + Rollback { target: Option, reply: Reply }, + Stop { reply: Reply }, + Start { reply: Reply }, + Remove { reply: Reply }, + Shutdown { reply: Reply }, } pub struct Broker { @@ -65,33 +69,36 @@ pub struct Broker { pub runtime: Arc, mailboxes: Mutex>>, controls: Mutex>>, - registry: Mutex, + /// `actor.json` of every actor, native and WASM. + metas: Mutex>, op_locks: Mutex>>>, seq: AtomicU64, } impl Broker { pub fn new(world: World, runtime: Arc) -> Result> { - let registry = world.load_registry()?; + let metas = world.actor_ids()?.into_iter().map(|id| (id.clone(), world.read_meta(&id))).collect(); Ok(Arc::new(Self { world, runtime, mailboxes: Mutex::new(HashMap::new()), controls: Mutex::new(HashMap::new()), - registry: Mutex::new(registry), + metas: Mutex::new(metas), op_locks: Mutex::new(HashMap::new()), seq: AtomicU64::new(1), })) } - pub fn registry(&self) -> Registry { - self.registry.lock().unwrap().clone() + pub fn meta(&self, id: &str) -> Option { + self.metas.lock().unwrap().get(id).cloned() } - fn update_registry(&self, f: impl FnOnce(&mut Registry)) -> Result<()> { - let mut reg = self.registry.lock().unwrap(); - f(&mut reg); - self.world.save_registry(®) + /// Change the `actor.json` of `id` (creating it if needed) and save it. + fn update_meta(&self, id: &str, f: impl FnOnce(&mut ActorMeta)) -> Result<()> { + let mut metas = self.metas.lock().unwrap(); + let meta = metas.entry(id.to_string()).or_insert_with(|| ActorMeta { running: true, ..Default::default() }); + f(meta); + self.world.write_meta(id, meta) } /// Create the mailbox of an actor. Messages wait here while the actor is @@ -102,27 +109,23 @@ impl Broker { rx } - fn next_id(&self) -> String { - format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)) - } - - /// Send a message from a native actor. `to: None` publishes to the - /// subscribers of `from`. Returns the message ID. - pub fn send(&self, from: &str, to: Option<&str>, id: Option<&str>, reply_to: Option<&str>, body: Value) -> String { - let id = match id { + /// Send a message. `to: None` publishes to the subscribers of `from`. + /// With `own_id`, the message ID is `"/"`. Returns the ID. + pub fn send(&self, from: &str, to: Option<&str>, own_id: Option<&str>, reply_to: Option<&str>, body: Value) -> String { + let id = match own_id { Some(own) => format!("{from}/{own}"), - None => self.next_id(), + None => format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)), }; let recipients: Vec = match to { Some(to) => vec![to.to_string()], None => self - .registry + .metas .lock() .unwrap() - .subscriptions - .get(from) - .map(|s| s.iter().cloned().collect()) - .unwrap_or_default(), + .iter() + .filter(|(_, m)| m.subscriptions.contains(from)) + .map(|(id, _)| id.clone()) + .collect(), }; for to in recipients { self.deliver(Envelope { @@ -142,8 +145,7 @@ impl Broker { if line.is_empty() { return; } - let parsed: Option = serde_json::from_str(line).ok(); - let Some(Value::Object(obj)) = parsed else { + let Ok(Value::Object(obj)) = serde_json::from_str::(line) else { self.log(from, &format!("stdout line is not a JSON object, ignored: {}", truncate(line, 500))); return; }; @@ -151,12 +153,12 @@ impl Broker { self.log(from, &format!("message without \"body\", ignored: {}", truncate(line, 500))); return; }; - let to = obj.get("to").and_then(Value::as_str); - let own_id = obj.get("id").and_then(|v| match v { - Value::String(s) => Some(s.clone()), - Value::Number(n) => Some(n.to_string()), + let own_id = match obj.get("id") { + Some(Value::String(s)) => Some(s.clone()), + Some(Value::Number(n)) => Some(n.to_string()), _ => None, - }); + }; + let to = obj.get("to").and_then(Value::as_str); let reply_to = obj.get("reply_to").and_then(Value::as_str); self.send(from, to, own_id.as_deref(), reply_to, body); } @@ -164,24 +166,16 @@ impl Broker { fn deliver(&self, env: Envelope) { let tx = self.mailboxes.lock().unwrap().get(&env.to).cloned(); let error = match tx { - None => Some(format!("no actor with ID {}", env.to)), + None => format!("no actor with ID {}", env.to), Some(tx) => match tx.try_send(env.clone()) { - Ok(()) => None, - Err(mpsc::error::TrySendError::Full(_)) => Some(format!("mailbox of {} is full", env.to)), - Err(mpsc::error::TrySendError::Closed(_)) => Some(format!("actor {} is gone", env.to)), + Ok(()) => return, + Err(mpsc::error::TrySendError::Full(_)) => format!("mailbox of {} is full", env.to), + Err(mpsc::error::TrySendError::Closed(_)) => format!("actor {} is gone", env.to), }, }; - if let Some(reason) = error - && env.from != "broker" - && env.from != "test" - { - self.send( - "broker", - Some(&env.from), - None, - Some(&env.id), - json!({"type": "undeliverable", "reason": reason, "to": env.to}), - ); + if env.from != "broker" { + let body = json!({"type": "undeliverable", "reason": error, "to": env.to}); + self.send("broker", Some(&env.from), None, Some(&env.id), body); } } @@ -198,7 +192,7 @@ impl Broker { self.op_locks.lock().unwrap().entry(id.to_string()).or_default().clone() } - /// Mounts of a running actor. + /// Mounts of a running actor. `/actor` hides every `data/` folder. pub fn mounts(&self, id: &str) -> Vec { vec![ Mount { guest: "/actor".into(), host: self.world.actors_root(), writable: false }, @@ -211,185 +205,280 @@ impl Broker { // Lifecycle // --------------------------------------------------------------------- - /// Start the supervisors of all registered WASM actors. - pub fn start_all(self: &Arc) { - let reg = self.registry(); - for (id, rec) in reg.actors { - self.start_supervisor(&id, rec.version, rec.running); - } - } - fn start_supervisor(self: &Arc, id: &str, version: u64, running: bool) { let mailbox = self.open_mailbox(id); let (ctl_tx, ctl_rx) = mpsc::channel(16); self.controls.lock().unwrap().insert(id.to_string(), ctl_tx); - let sv = Supervisor::new(self.clone(), id.to_string(), version, running, mailbox, ctl_rx); - tokio::spawn(sv.run()); + tokio::spawn(Supervisor::new(self.clone(), id.to_string(), version, running, mailbox, ctl_rx).run()); } - async fn control(&self, id: &str, make: impl FnOnce(oneshot::Sender>) -> Control) -> Result { + fn is_supervised(&self, id: &str) -> bool { + self.controls.lock().unwrap().contains_key(id) + } + + async fn control(&self, id: &str, make: impl FnOnce(Reply) -> Control) -> Result { let tx = self.controls.lock().unwrap().get(id).cloned().ok_or_else(|| anyhow!("no WASM actor with ID {id}"))?; let (reply, rx) = oneshot::channel(); tx.send(make(reply)).await.map_err(|_| anyhow!("supervisor of {id} is gone"))?; rx.await.map_err(|_| anyhow!("supervisor of {id} did not answer"))? } - /// Stop every actor gracefully (host shutdown). The registry keeps their - /// `running` flags, so the next `sausage run` starts them again. - pub async fn shutdown_all(&self) { + /// Stop every actor gracefully, in parallel (host shutdown). The + /// `running` flags stay, so the next `sausage run` starts them again. + pub async fn shutdown_all(self: &Arc) { let ids: Vec = self.controls.lock().unwrap().keys().cloned().collect(); - let mut waits = Vec::new(); + let mut tasks = tokio::task::JoinSet::new(); for id in ids { - waits.push(async move { - let _ = self.control(&id, |reply| Control::Shutdown { reply }).await; - }); + let this = self.clone(); + tasks.spawn(async move { this.control(&id, |reply| Control::Shutdown { reply }).await }); } - futures_join_all(waits).await; + tasks.join_all().await; } - /// Store a new version: copy the `.wasm` file and the source folder into - /// the version store. Returns the version directory and the component. - fn store_version(&self, id: &str, version: u64, wasm: &Path, source: Option<&Path>, meta: &VersionMeta) -> Result<(PathBuf, Component)> { - let dir = self.world.version_dir(id, version); - let _ = std::fs::remove_dir_all(&dir); + /// Copy a `.wasm` file and its source folder into `versions//`. + /// 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 { + let dir = self.world.version_dir(id, meta.version); std::fs::create_dir_all(&dir)?; std::fs::copy(wasm, dir.join("actor.wasm")).with_context(|| format!("copy {}", wasm.display()))?; - match source { - Some(src) => world::copy_tree(src, &dir.join("source"))?, - None => std::fs::create_dir_all(dir.join("source"))?, - } + world::replace_tree(source.unwrap_or(&dir.join("no-source")), &dir.join("source"))?; world::write_atomic(&dir.join("meta.json"), serde_json::to_string_pretty(meta)?.as_bytes())?; - let component = self.runtime.load(&dir.join("actor.wasm"))?; - Ok((dir, component)) + self.runtime.load(&dir.join("actor.wasm")) + } + + /// Record in `versions//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::(&s).ok()) + .is_some_and(|m| m.rejected.is_some()) + } + + /// Install `version` as the current files of a new actor. + fn install_files(&self, id: &str, version: u64) -> Result<()> { + let dir = self.world.version_dir(id, version); + self.world.ensure_actor_dirs(id)?; + world::write_atomic(&self.world.wasm_path(id), &std::fs::read(dir.join("actor.wasm"))?)?; + world::replace_tree(&dir.join("source"), &self.world.source_dir(id))?; + self.update_meta(id, |m| { + m.version = Some(version); + m.running = true; + }) } /// Create a new actor from a `.wasm` file. With `start`, also start its - /// supervisor (in a running host). Without `start` (during `init`), only - /// register it. - pub async fn create_actor( - self: &Arc, - id: &str, - wasm: &Path, - source: Option<&Path>, - meta: VersionMeta, - start: bool, - ) -> Result { + /// supervisor (in a running host). Without `start` (`init`), only install it. + pub async fn create_actor(self: &Arc, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta, start: bool) -> Result { if !world::valid_id(id) { bail!("invalid actor ID {id:?}: use a-z, 0-9, '-' and '_'"); } - if NATIVE.contains(&id) || self.registry().actors.contains_key(id) { + if NATIVE.contains(&id) || self.is_supervised(id) || self.world.wasm_path(id).exists() { bail!("an actor with ID {id} already exists"); } let lock = self.op_lock(id); let _guard = lock.lock().await; - let (dir, component) = self.store_version(id, 1, wasm, source, &meta)?; - let report = candidate::test(self, id, &component, &dir.join("source"), None, None).await; - let report = match report { + let version = self.world.next_version(id); + let meta = VersionMeta { version, ..meta }; + self.create_version(id, version, &meta, wasm, source, start).await + } + + async fn create_version(self: &Arc, id: &str, version: u64, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, start: bool) -> Result { + let tested = async { + let component = self.store_version(id, meta, wasm, source)?; + candidate::test(self, id, &component, &self.world.version_dir(id, version).join("source"), None, None).await + } + .await; + let report = match tested { Ok(r) => r, Err(e) => { - let _ = std::fs::remove_dir_all(self.world.versions_dir(id)); + self.reject(id, version, &e); return Err(e); } }; - self.world.ensure_actor_dirs(id)?; - world::replace_tree(&dir.join("source"), &self.world.source_dir(id))?; - self.update_registry(|r| { - r.actors.insert(id.to_string(), ActorRecord { version: 1, next_version: 2, running: true }); - })?; - write_status(&self.world, id, &json!({"id": id, "version": 1, "state": "created"})); + self.install_files(id, version)?; + write_status(&self.world, id, &json!({"id": id, "version": version, "state": "created"})); if start { - self.start_supervisor(id, 1, true); + self.start_supervisor(id, version, true); } - Ok(json!({"type": "spawned", "actor": id, "version": 1, "tests": report})) + Ok(json!({"type": "spawned", "actor": id, "version": version, "tests": report})) } - pub async fn upgrade_actor(self: &Arc, id: &str, wasm: &Path, source: Option<&Path>, mut meta: VersionMeta) -> Result { + pub async fn upgrade_actor(self: &Arc, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> Result { let lock = self.op_lock(id); let _guard = lock.lock().await; - let rec = self.registry().actors.get(id).cloned().ok_or_else(|| anyhow!("no WASM actor with ID {id}"))?; - let version = rec.next_version; - self.update_registry(|r| { - if let Some(a) = r.actors.get_mut(id) { - a.next_version = version + 1; - } - })?; - meta.version = version; - let (dir, component) = self.store_version(id, version, wasm, source, &meta)?; - let report = candidate::test( - self, - id, - &component, - &dir.join("source"), - Some(&self.world.data_dir(id)), - Some(&self.world.public_dir(id)), - ) - .await; - if let Err(e) = report { - let _ = std::fs::remove_dir_all(&dir); - return Err(e); + if !self.is_supervised(id) { + bail!("no WASM actor with ID {id}"); } - let mut result = self - .control(id, |reply| Control::Upgrade { version, requester: meta.requested_by.clone(), reply }) - .await?; - result["tests"] = report.unwrap(); + let version = self.world.next_version(id); + self.upgrade_version(id, &VersionMeta { version, ..meta }, wasm, source).await + } + + async fn upgrade_version(self: &Arc, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result { + let version = meta.version; + let tested = async { + let component = self.store_version(id, meta, wasm, source)?; + let source = self.world.version_dir(id, version).join("source"); + candidate::test(self, id, &component, &source, Some(&self.world.data_dir(id)), Some(&self.world.public_dir(id))).await + } + .await; + let report = match tested { + Ok(r) => r, + Err(e) => { + self.reject(id, version, &e); + return Err(e); + } + }; + let requester = meta.requested_by.clone(); + let mut result = self.control(id, |reply| Control::Upgrade { version, requester, reply }).await?; + result["tests"] = report; Ok(result) } + /// Boot the actors in `actor/`. Called at start and on `scan`. + /// + /// - A known `actor.wasm` (equal to the installed version) starts as is. + /// - A new or changed `actor.wasm` is first copied to `versions//`. + /// A new actor is tested and installed. A changed one is an upgrade: + /// if it fails, the installed files come back. + /// - A file that equals an already rejected version is not tried again. + pub async fn scan(self: &Arc) -> Vec { + let mut report = Vec::new(); + for id in self.world.actor_ids().unwrap_or_default() { + if NATIVE.contains(&id.as_str()) || !self.world.wasm_path(&id).is_file() { + continue; + } + let lock = self.op_lock(&id); + let _guard = lock.lock().await; + if let Some(r) = self.scan_actor(&id).await { + report.push(r); + } + } + report + } + + async fn scan_actor(self: &Arc, id: &str) -> Option { + let meta = self.world.read_meta(id); + let wasm = self.world.wasm_path(id); + let same = |v: u64| same_file(&wasm, &self.world.version_dir(id, v).join("actor.wasm")); + let installed = meta.version.filter(|v| self.world.version_dir(id, *v).join("actor.wasm").is_file()); + self.metas.lock().unwrap().insert(id.to_string(), meta.clone()); + if let Some(v) = installed + && !self.is_supervised(id) + { + self.start_supervisor(id, v, meta.running); + } + if installed.is_some_and(same) { + return None; + } + let latest = self.world.versions(id).last().copied(); + if let Some(v) = latest + && same(v) + && self.version_rejected(id, v) + { + if let Some(current) = installed { + let _ = self.restore_files(id, current); + } + self.log(id, &format!("actor.wasm equals the rejected version {v}; not installed")); + return Some(json!({"actor": id, "skipped": "rejected version", "version": v})); + } + let version = self.world.next_version(id); + let source = self.world.source_dir(id); + let vmeta = VersionMeta { + version, + wasm: format!("/actor/{id}/actor.wasm"), + source: Some(format!("/actor/{id}/source")), + built_by: None, + requested_by: "scan".into(), + created_at: world::now(), + rejected: None, + }; + let source = source.is_dir().then_some(source.as_path()); + let result = match installed { + None => self.create_version(id, version, &vmeta, &wasm, source, true).await, + Some(current) => { + let r = self.upgrade_version(id, &vmeta, &wasm, source).await; + if r.is_err() { + // Put the installed version back in place. + let _ = self.restore_files(id, current); + } + r + } + }; + Some(match result { + Ok(v) => v, + Err(e) => { + self.log(id, &format!("version {version} from disk rejected: {e:#}")); + json!({"type": "error", "actor": id, "version": version, "reason": format!("{e:#}")}) + } + }) + } + + fn restore_files(&self, id: &str, version: u64) -> Result<()> { + let dir = self.world.version_dir(id, version); + world::write_atomic(&self.world.wasm_path(id), &std::fs::read(dir.join("actor.wasm"))?)?; + world::replace_tree(&dir.join("source"), &self.world.source_dir(id)) + } + /// Handle one message to `broker`. async fn handle(self: Arc, env: Envelope) { - let result = self.clone().command(&env).await; - let body = match result { - Ok(v) => v, - Err(e) => json!({"type": "error", "request": env.body.get("type"), "reason": format!("{e:#}")}), - }; + let body = self.clone().command(&env).await.unwrap_or_else(|e| { + json!({"type": "error", "request": env.body.get("type"), "reason": format!("{e:#}")}) + }); self.send("broker", Some(&env.from), None, Some(&env.id), body); } async fn command(self: Arc, env: &Envelope) -> Result { let b = &env.body; - let ty = b.get("type").and_then(Value::as_str).unwrap_or(""); - let actor = || -> Result { - 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, VersionMeta)> { - let wasm_s = b.get("wasm").and_then(Value::as_str).ok_or_else(|| anyhow!("missing \"wasm\""))?; + let ty = b["type"].as_str().unwrap_or(""); + let actor = || b["actor"].as_str().map(str::to_string).ok_or_else(|| anyhow!("missing \"actor\"")); + let version_meta = || -> Result<(std::path::PathBuf, Option, VersionMeta)> { + let wasm_s = b["wasm"].as_str().ok_or_else(|| anyhow!("missing \"wasm\""))?; let wasm = self.world.resolve(&env.from, wasm_s)?; if !wasm.is_file() { bail!("{wasm_s} is not a file"); } - let source_s = b.get("source").and_then(Value::as_str); + let source_s = b["source"].as_str(); let source = source_s.map(|s| self.world.resolve(&env.from, s)).transpose()?; - if let Some(s) = &source - && !s.is_dir() + if let (Some(s), Some(p)) = (source_s, &source) + && !p.is_dir() { - bail!("{} is not a folder", source_s.unwrap()); + bail!("{s} is not a folder"); } - let m = VersionMeta { - version, + let meta = VersionMeta { + version: 0, wasm: wasm_s.to_string(), source: source_s.map(str::to_string), - built_by: b.get("built_by").and_then(Value::as_str).map(str::to_string), + built_by: b["built_by"].as_str().map(str::to_string), requested_by: env.from.clone(), created_at: world::now(), + rejected: None, }; - Ok((wasm, source, m)) + Ok((wasm, source, meta)) }; match ty { "spawn" => { - let id = actor()?; - let (wasm, source, m) = meta(1)?; - self.create_actor(&id, &wasm, source.as_deref(), m, true).await + let (wasm, source, meta) = version_meta()?; + self.create_actor(&actor()?, &wasm, source.as_deref(), meta, true).await } "upgrade" => { - let id = actor()?; - let (wasm, source, m) = meta(0)?; - self.upgrade_actor(&id, &wasm, source.as_deref(), m).await + let (wasm, source, meta) = version_meta()?; + self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta).await } "rollback" => { let id = actor()?; - let target = b.get("version").and_then(Value::as_u64); + let target = b["version"].as_u64(); let lock = self.op_lock(&id); - let _g = lock.lock().await; + let _guard = lock.lock().await; self.control(&id, |reply| Control::Rollback { target, reply }).await } "stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await, @@ -397,49 +486,44 @@ impl Broker { "remove" => { let id = actor()?; let lock = self.op_lock(&id); - let _g = lock.lock().await; + let _guard = lock.lock().await; let r = self.control(&id, |reply| Control::Remove { reply }).await?; self.mailboxes.lock().unwrap().remove(&id); self.controls.lock().unwrap().remove(&id); - self.update_registry(|reg| { - reg.actors.remove(&id); - reg.subscriptions.remove(&id); - for subs in reg.subscriptions.values_mut() { - subs.remove(&id); - } - })?; + self.metas.lock().unwrap().remove(&id); Ok(r) } "subscribe" | "unsubscribe" => { let publisher = actor()?; - let sub = env.from.clone(); let on = ty == "subscribe"; - self.update_registry(|reg| { - let set = reg.subscriptions.entry(publisher.clone()).or_insert_with(BTreeSet::new); + self.update_meta(&env.from, |m| { if on { - set.insert(sub.clone()); + m.subscriptions.insert(publisher.clone()); } else { - set.remove(&sub); + m.subscriptions.remove(&publisher); } })?; Ok(json!({"type": format!("{ty}d"), "actor": publisher})) } + "scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})), "list" => { - let reg = self.registry(); - let mut actors: Vec = 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::(&s).ok()); - actors.push(json!({ - "id": id, "kind": "wasm", "version": rec.version, "running": rec.running, - "state": status.as_ref().and_then(|s| s.get("state")).cloned(), - })); - } - Ok(json!({"type": "actors", "actors": actors, "subscriptions": reg.subscriptions})) + let metas = self.metas.lock().unwrap().clone(); + let actors: Vec = 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::(&s).ok()) + .unwrap_or_default(); + json!({"id": id, "kind": if NATIVE.contains(&id.as_str()) { "native" } else { "wasm" }, + "version": m.version, "running": m.running, "state": status["state"], + "subscriptions": m.subscriptions}) + }) + .collect(); + Ok(json!({"type": "actors", "actors": actors})) } "shutdown" => Ok(json!({"type": "ignored"})), - other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, list"), + other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, scan, list"), } } @@ -451,14 +535,8 @@ impl Broker { } } -async fn futures_join_all>(futs: Vec) { - let mut set = Vec::new(); - for f in futs { - set.push(Box::pin(f)); - } - for f in set { - f.await; - } +fn same_file(a: &Path, b: &Path) -> bool { + matches!((std::fs::read(a), std::fs::read(b)), (Ok(x), Ok(y)) if x == y) } pub fn truncate(s: &str, max: usize) -> String { @@ -723,20 +801,23 @@ impl Supervisor { Ok(()) } + /// Make `version` the current files: `actor.wasm`, `source/`, `actor.json`. fn install(&mut self, version: u64) -> Result<()> { - let src = self.world().version_dir(&self.id, version).join("source"); - world::replace_tree(&src, &self.world().source_dir(&self.id))?; + let dir = self.world().version_dir(&self.id, version); + world::write_atomic(&self.world().wasm_path(&self.id), &std::fs::read(dir.join("actor.wasm"))?)?; + world::replace_tree(&dir.join("source"), &self.world().source_dir(&self.id))?; self.version = version; self.component = None; - let id = self.id.clone(); - self.broker.update_registry(|r| { - if let Some(a) = r.actors.get_mut(&id) { - a.version = version; - a.running = true; - } + self.broker.update_meta(&self.id, |m| { + m.version = Some(version); + m.running = true; }) } + fn set_running(&self, running: bool) -> Result<()> { + self.broker.update_meta(&self.id, |m| m.running = running) + } + async fn upgrade(&mut self, version: u64, requester: String) -> Result { self.stop_instance().await; self.snapshot_current()?; @@ -759,7 +840,7 @@ impl Supervisor { Some(t) => t, None => (1..self.version) .rev() - .find(|v| self.world().version_dir(&self.id, *v).join("actor.wasm").is_file()) + .find(|v| self.world().version_dir(&self.id, *v).join("actor.wasm").is_file() && !self.broker.version_rejected(&self.id, *v)) .ok_or_else(|| anyhow!("{} has no version before {}", self.id, self.version))?, }; let dir = self.world().version_dir(&self.id, target); @@ -800,23 +881,13 @@ impl Supervisor { self.restart_at = None; self.stop_instance().await; self.state = "stopped"; - let id = self.id.clone(); - let r = self.broker.update_registry(|r| { - if let Some(a) = r.actors.get_mut(&id) { - a.running = false; - } - }); + let r = self.set_running(false); let _ = reply.send(r.map(|_| json!({"type": "stopped", "actor": self.id}))); } Control::Start { reply } => { self.desired_running = true; self.restart_at = None; - let id = self.id.clone(); - let r = self.broker.update_registry(|r| { - if let Some(a) = r.actors.get_mut(&id) { - a.running = true; - } - }); + let r = self.set_running(true); let _ = reply.send(r.map(|_| json!({"type": "started", "actor": self.id}))); } Control::Remove { reply } => { @@ -838,19 +909,12 @@ impl Supervisor { false } - /// Move every folder of the actor into `archive/`. Nothing is deleted. + /// Move the folder of the actor into `archive/`. Nothing is deleted. fn archive(&self) -> Result { - let dest = self.world().root.join("archive").join(format!("{}-{}", self.id, world::now())); - std::fs::create_dir_all(&dest)?; - for (from, name) in [ - (self.world().actor_dir(&self.id), "actor"), - (self.world().root.join("private").join(&self.id), "private"), - (self.world().versions_dir(&self.id), "versions"), - ] { - if from.exists() { - std::fs::rename(&from, dest.join(name))?; - } - } + let archive = self.world().root.join("archive"); + std::fs::create_dir_all(&archive)?; + let dest = archive.join(format!("{}-{}", self.id, world::now())); + std::fs::rename(self.world().actor_dir(&self.id), &dest)?; Ok(json!({"type": "removed", "actor": self.id, "archive": dest.display().to_string()})) } } diff --git a/cli/src/candidate.rs b/cli/src/candidate.rs index 00a2e79..e6a4eda 100644 --- a/cli/src/candidate.rs +++ b/cli/src/candidate.rs @@ -110,7 +110,9 @@ pub async fn test( } fn start(broker: &Broker, id: &str, component: &Component, data: Option<&Path>, public: Option<&Path>) -> Result<(Instance, Scratch)> { - let dir = broker.world.scratch_dir().join(format!("{id}-{}-{}", world::now(), rand_suffix())); + static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let dir = broker.world.scratch_dir().join(format!("{id}-{}-{n}", std::process::id())); let scratch = Scratch { dir: dir.clone() }; std::fs::create_dir_all(dir.join("data"))?; std::fs::create_dir_all(dir.join("public"))?; @@ -133,13 +135,6 @@ fn start(broker: &Broker, id: &str, component: &Component, data: Option<&Path>, Ok((broker.runtime.start(component, cfg)?, scratch)) } -fn rand_suffix() -> u32 { - use std::hash::{BuildHasher, Hasher}; - let mut h = std::collections::hash_map::RandomState::new().build_hasher(); - h.write_u64(std::process::id() as u64); - h.finish() as u32 -} - async fn run_file( broker: &Broker, id: &str, diff --git a/cli/src/cargo_actor.rs b/cli/src/cargo_actor.rs index c05075f..1e423c5 100644 --- a/cli/src/cargo_actor.rs +++ b/cli/src/cargo_actor.rs @@ -6,22 +6,30 @@ //! //! The broker knows nothing about Rust. This actor is one way to get a //! `.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`. use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Instant; use anyhow::{Context, Result, anyhow, bail}; use serde_json::{Value, json}; use tokio::process::Command; use tokio::sync::{Mutex, mpsc}; -use crate::broker::{Broker, Envelope, truncate}; +use include_dir::{Dir, include_dir}; + +use crate::broker::{Broker, Envelope}; use crate::world::{self, World}; pub const ID: &str = "cargo"; pub const TARGET: &str = "wasm32-wasip2"; -const BUILD_TIMEOUT: Duration = Duration::from_secs(900); + +/// The source of the `sausage-actor` SDK. +static SDK: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk"); pub struct BuildOutput { /// Host path of the published `.wasm` file. @@ -37,8 +45,10 @@ pub struct Cargo { } impl Cargo { - pub fn new(world: World) -> Self { - Self { world, lock: Mutex::new(()) } + /// Create the actor and install the SDK of this host into `data/sdk/`. + pub fn new(world: World) -> Result { + write_dir(&SDK, &world.data_dir(ID).join("sdk"))?; + Ok(Self { world, lock: Mutex::new(()) }) } /// Build the crate in `source` (a host path) and publish the `.wasm` file @@ -53,12 +63,8 @@ impl Cargo { let key = format!("{:016x}", fxhash(source.to_string_lossy().as_bytes())); let build_dir = data.join("work").join(&key); world::replace_tree(source, &build_dir)?; - // The crate must not join a parent workspace by accident. let manifest = build_dir.join("Cargo.toml"); - let text = std::fs::read_to_string(&manifest)?; - if !text.contains("[workspace]") { - std::fs::write(&manifest, format!("{text}\n[workspace]\n"))?; - } + std::fs::write(&manifest, prepare_manifest(&std::fs::read_to_string(&manifest)?, Some(&data.join("sdk")))?)?; let mut cmd = Command::new("cargo"); cmd.arg("build") @@ -70,13 +76,10 @@ impl Cargo { .arg(data.join("target")) .current_dir(&build_dir) .kill_on_drop(true); - let out = tokio::time::timeout(BUILD_TIMEOUT, cmd.output()) - .await - .map_err(|_| anyhow!("cargo build took longer than {BUILD_TIMEOUT:?}"))? - .context("cannot run cargo (is it on PATH, with the wasm32-wasip2 target?)")?; + let out = cmd.output().await.context("cannot run cargo (is it on PATH, with the wasm32-wasip2 target?)")?; let stderr = String::from_utf8_lossy(&out.stderr).to_string(); if !out.status.success() { - bail!("{}", truncate(&strip_progress(&stderr), 20_000)); + bail!("{}", strip_progress(&stderr)); } let mut artifact = None; for line in String::from_utf8_lossy(&out.stdout).lines() { @@ -123,7 +126,7 @@ impl Cargo { "source": source_s, "built_by": ID, "seconds": (out.seconds * 10.0).round() / 10.0, - "warnings": truncate(&out.warnings, 4000), + "warnings": out.warnings, })) } Some("shutdown") => Ok(json!({"type": "ignored"})), @@ -132,6 +135,47 @@ impl Cargo { } } +/// Point a `sausage-actor` dependency at `sdk` (keeping its features), and +/// add an empty `[workspace]`, so the crate never joins a parent workspace. +fn prepare_manifest(text: &str, sdk: Option<&Path>) -> Result { + let mut doc: toml_edit::DocumentMut = text.parse()?; + if let Some(sdk) = sdk { + for section in ["dependencies", "dev-dependencies", "build-dependencies"] { + let Some(dep) = doc.get_mut(section).and_then(|t| t.get_mut("sausage-actor")) else { continue }; + let mut table = dep.as_inline_table().cloned().unwrap_or_default(); + for key in ["version", "path", "git", "branch", "tag", "rev", "registry"] { + table.remove(key); + } + table.insert("path", sdk.display().to_string().into()); + *dep = toml_edit::value(table); + } + } + if doc.get("workspace").is_none() { + doc["workspace"] = toml_edit::table(); + } + Ok(doc.to_string()) +} + +/// Write the files of `dir` below `to`. Skip files that did not change, so +/// that cargo does not rebuild the SDK at every start. +fn write_dir(dir: &Dir<'_>, to: &Path) -> Result<()> { + for file in dir.files() { + let dest = to.join(file.path()); + let contents = match file.path().file_name() { + Some(n) if n == "Cargo.toml" => prepare_manifest(file.contents_utf8().context("Cargo.toml")?, None)?.into_bytes(), + _ => file.contents().to_vec(), + }; + if std::fs::read(&dest).ok() != Some(contents.clone()) { + std::fs::create_dir_all(dest.parent().context("no parent")?)?; + std::fs::write(&dest, contents)?; + } + } + for sub in dir.dirs() { + write_dir(sub, to)?; + } + Ok(()) +} + fn next_build_number(dir: &Path) -> Result { std::fs::create_dir_all(dir)?; let max = std::fs::read_dir(dir)? diff --git a/cli/src/console.rs b/cli/src/console.rs index c739c56..6a015ec 100644 --- a/cli/src/console.rs +++ b/cli/src/console.rs @@ -115,7 +115,7 @@ impl Console { if built.body["type"] != "built" { bail!("build failed"); } - let exists = self.broker.registry().actors.contains_key(actor); + 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!({ diff --git a/cli/src/fs.rs b/cli/src/fs.rs index 2afe1de..2f0d4bf 100644 --- a/cli/src/fs.rs +++ b/cli/src/fs.rs @@ -9,6 +9,9 @@ //! //! 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}; @@ -37,7 +40,19 @@ pub struct Mount { struct DirInfo { host: PathBuf, root: PathBuf, + /// Writable mount: new and truncated files commit on close. commit: bool, + /// Read-only mount: private folders are hidden. + hide: bool, +} + +/// `/data` and `/versions//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 { @@ -49,12 +64,14 @@ struct Pending { pub struct FsExt { mounts: Vec, dirs: HashMap, + /// Directory streams of hiding mounts, and the directory they list. + streams: HashMap, pending: HashMap, } impl FsExt { pub fn new(mounts: Vec) -> Self { - Self { mounts, dirs: HashMap::new(), pending: HashMap::new() } + Self { mounts, dirs: HashMap::new(), streams: HashMap::new(), pending: HashMap::new() } } /// Commit every open write. Use this when the guest exits normally, @@ -110,6 +127,17 @@ 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, 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 { @@ -123,7 +151,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 }, + DirInfo { host: m.host.clone(), root: m.host.clone(), commit: m.writable, hide: !m.writable }, ); } } @@ -202,7 +230,12 @@ impl HostDescriptor for ActorFs<'_> { &mut self, fd: Resource, ) -> FsResult> { - HostDescriptor::read_directory(&mut self.inner, fd).await + 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) } async fn sync(&mut self, fd: Resource) -> FsResult<()> { @@ -223,7 +256,7 @@ impl HostDescriptor for ActorFs<'_> { path_flags: types::PathFlags, path: String, ) -> FsResult { - if is_tmp_name(&path) { + if self.blocked(&fd, &path) { return Err(ErrorCode::NoEntry.into()); } HostDescriptor::stat_at(&mut self.inner, fd, path_flags, path).await @@ -259,7 +292,7 @@ impl HostDescriptor for ActorFs<'_> { oflags: types::OpenFlags, flags: types::DescriptorFlags, ) -> FsResult> { - if is_tmp_name(&path) { + if self.blocked(&fd, &path) { return Err(ErrorCode::NoEntry.into()); } let parent = self.ext.dirs.get(&fd.rep()).cloned(); @@ -294,7 +327,7 @@ impl HostDescriptor for ActorFs<'_> { && matches!(self.inner.table.get(&r)?, Descriptor::Dir(_)) && let Some(host) = resolve(&dir, &path) { - self.ext.dirs.insert(r.rep(), DirInfo { host, root: dir.root, commit: dir.commit }); + self.ext.dirs.insert(r.rep(), DirInfo { host, ..dir }); } Ok(r) } @@ -310,6 +343,9 @@ impl HostDescriptor for ActorFs<'_> { } async fn readlink_at(&mut self, fd: Resource, path: String) -> FsResult { + if self.blocked(&fd, &path) { + return Err(ErrorCode::NoEntry.into()); + } HostDescriptor::readlink_at(&mut self.inner, fd, path).await } @@ -385,6 +421,9 @@ impl HostDescriptor for ActorFs<'_> { path_flags: types::PathFlags, path: String, ) -> FsResult { + if self.blocked(&fd, &path) { + return Err(ErrorCode::NoEntry.into()); + } HostDescriptor::metadata_hash_at(&mut self.inner, fd, path_flags, path).await } } @@ -401,14 +440,21 @@ 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 e.name.starts_with(TMP_PREFIX) => continue, + Some(e) if hidden(&e.name) => continue, other => return Ok(other), } } } fn drop(&mut self, stream: Resource) -> wasmtime::Result<()> { + self.ext.streams.remove(&stream.rep()); HostDirectoryEntryStream::drop(&mut self.inner, stream) } } diff --git a/cli/src/lib.rs b/cli/src/lib.rs index d176c1a..2cfa971 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -44,9 +44,11 @@ impl Host { let rx = broker.open_mailbox("broker"); tokio::spawn(broker.clone().run_actor(rx)); let rx = broker.open_mailbox(cargo_actor::ID); - tokio::spawn(Arc::new(cargo_actor::Cargo::new(world.clone())).run_actor(broker.clone(), rx)); + tokio::spawn(Arc::new(cargo_actor::Cargo::new(world.clone())?).run_actor(broker.clone(), rx)); - broker.start_all(); + for change in broker.scan().await { + eprintln!("sausage: {change}"); + } Ok(Host { world, broker, _lock: lock }) } diff --git a/cli/src/main.rs b/cli/src/main.rs index 8a84d7b..a7994b5 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -7,7 +7,7 @@ use clap::{Parser, Subcommand}; use include_dir::{Dir, include_dir}; use serde_json::json; -use sausage::world::{self, VersionMeta, World}; +use sausage::world::{self, ActorMeta, VersionMeta, World}; use sausage::{Broker, Host, cargo_actor, console, runtime}; /// The source of the agent actor. `init` builds it like any other actor. @@ -69,9 +69,8 @@ async fn init(dir: PathBuf) -> Result<()> { let token = setting("SAUSAGE_API_TOKEN", "API token", None)?; let world = World::create(&dir)?; - let mut reg = world.load_registry()?; - reg.subscriptions.entry("agent".into()).or_default().insert("console".into()); - world.save_registry(®)?; + 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. world.ensure_actor_dirs("agent")?; @@ -84,11 +83,11 @@ async fn init(dir: PathBuf) -> Result<()> { std::fs::set_permissions(&config_path, std::fs::Permissions::from_mode(0o600))?; } - let source = world.root.join("broker").join("seed").join("agent"); + let source = world.scratch_dir().join("seed").join("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")?; + 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()?)?; @@ -99,6 +98,7 @@ async fn init(dir: PathBuf) -> Result<()> { 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()); diff --git a/cli/src/world.rs b/cli/src/world.rs index 13e9deb..1c78d1a 100644 --- a/cli/src/world.rs +++ b/cli/src/world.rs @@ -1,19 +1,22 @@ -//! On-disk layout of a Sausage workspace ("world"). +//! On-disk layout of a Sausage world. The folder is the source of truth: +//! `sausage run` boots every actor that it finds in `actor/`. //! //! ```text //! / -//! sausage.json registry: actors, versions, subscriptions -//! actor//public/ written by the actor, read-only for others -//! actor//source/ source of the running version (broker) -//! actor//status.json state of the actor (broker) -//! actor//stderr.log stderr of the actor (broker) -//! private//data/ private state of the actor -//! broker/versions/// actor.wasm, source/, meta.json, snapshot/ -//! archive/ removed actors +//! actor// +//! actor.wasm the current version +//! 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) +//! stderr.log stderr of the actor and notes of the broker +//! versions// actor.wasm, source/, meta.json, snapshot/{data,public} +//! archive/ removed actors //! ``` -use std::collections::{BTreeMap, BTreeSet}; -use std::path::{Path, PathBuf}; +use std::collections::BTreeSet; +use std::path::{Component, Path, PathBuf}; use anyhow::{Context, Result, bail}; use serde::{Deserialize, Serialize}; @@ -28,20 +31,25 @@ pub struct World { pub root: PathBuf, } -#[derive(Default, Serialize, Deserialize, Clone)] -pub struct Registry { - pub actors: BTreeMap, - /// Publisher ID → subscriber IDs. - pub subscriptions: BTreeMap>, -} - -#[derive(Serialize, Deserialize, Clone, Debug)] -pub struct ActorRecord { - pub version: u64, - pub next_version: u64, +/// `actor//actor.json`. +#[derive(Serialize, Deserialize, Clone, Debug, Default)] +pub struct ActorMeta { + /// The installed version. `None` for native actors and for a new + /// `actor.wasm` that the broker has not installed yet. + #[serde(default)] + pub version: Option, + #[serde(default = "yes")] pub running: bool, + /// The actors whose published messages this actor receives. + #[serde(default)] + pub subscriptions: BTreeSet, } +fn yes() -> bool { + true +} + +/// `versions//meta.json`. #[derive(Serialize, Deserialize, Clone, Debug, Default)] pub struct VersionMeta { pub version: u64, @@ -51,9 +59,12 @@ pub struct VersionMeta { pub source: Option, /// The actor that built the `.wasm` file. pub built_by: Option, - /// The actor that requested the spawn or upgrade. + /// 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, } pub fn valid_id(id: &str) -> bool { @@ -74,15 +85,15 @@ impl World { Self { root: root.into() } } - pub fn registry_path(&self) -> PathBuf { - self.root.join("sausage.json") - } - pub fn actor_dir(&self, id: &str) -> PathBuf { - self.root.join("actor").join(id) - } pub fn actors_root(&self) -> PathBuf { self.root.join("actor") } + pub fn actor_dir(&self, id: &str) -> PathBuf { + self.actors_root().join(id) + } + pub fn wasm_path(&self, id: &str) -> PathBuf { + self.actor_dir(id).join("actor.wasm") + } pub fn public_dir(&self, id: &str) -> PathBuf { self.actor_dir(id).join("public") } @@ -90,20 +101,21 @@ impl World { self.actor_dir(id).join("source") } pub fn data_dir(&self, id: &str) -> PathBuf { - self.root.join("private").join(id).join("data") + self.actor_dir(id).join("data") } pub fn versions_dir(&self, id: &str) -> PathBuf { - self.root.join("broker").join("versions").join(id) + self.actor_dir(id).join("versions") } pub fn version_dir(&self, id: &str, v: u64) -> PathBuf { self.versions_dir(id).join(v.to_string()) } + /// Private scratch space of the broker (candidate copies, seeds). pub fn scratch_dir(&self) -> PathBuf { - self.root.join("broker").join("scratch") + self.data_dir("broker").join("scratch") } pub fn exists(&self) -> bool { - self.registry_path().is_file() + self.actors_root().is_dir() } /// Create an empty world with the folders of the native actors. @@ -112,8 +124,6 @@ impl World { if world.exists() { bail!("{} is already a Sausage world", root.display()); } - std::fs::create_dir_all(&world.root)?; - world.save_registry(&Registry::default())?; for id in NATIVE { world.ensure_actor_dirs(id)?; let status = serde_json::json!({"id": id, "kind": "native", "state": "built-in"}); @@ -122,21 +132,46 @@ impl World { Ok(world) } - /// Create the directories of one actor. pub fn ensure_actor_dirs(&self, id: &str) -> Result<()> { std::fs::create_dir_all(self.public_dir(id))?; std::fs::create_dir_all(self.data_dir(id))?; Ok(()) } - pub fn load_registry(&self) -> Result { - let text = std::fs::read_to_string(self.registry_path()) - .with_context(|| format!("no Sausage world at {}", self.root.display()))?; - Ok(serde_json::from_str(&text)?) + /// IDs of every folder in `actor/`, sorted. + pub fn actor_ids(&self) -> Result> { + let mut ids: Vec = std::fs::read_dir(self.actors_root())? + .filter_map(|e| e.ok()) + .filter(|e| e.path().is_dir()) + .filter_map(|e| e.file_name().into_string().ok()) + .filter(|id| valid_id(id)) + .collect(); + ids.sort(); + Ok(ids) } - pub fn save_registry(&self, reg: &Registry) -> Result<()> { - write_atomic(&self.registry_path(), serde_json::to_string_pretty(reg)?.as_bytes()) + pub fn read_meta(&self, id: &str) -> ActorMeta { + std::fs::read_to_string(self.actor_dir(id).join("actor.json")) + .ok() + .and_then(|s| serde_json::from_str(&s).ok()) + .unwrap_or(ActorMeta { running: true, ..Default::default() }) + } + + pub fn write_meta(&self, id: &str, meta: &ActorMeta) -> Result<()> { + write_atomic(&self.actor_dir(id).join("actor.json"), serde_json::to_string_pretty(meta)?.as_bytes()) + } + + /// Version numbers in `versions/`, ascending. + pub fn versions(&self, id: &str) -> Vec { + let mut v: Vec = std::fs::read_dir(self.versions_dir(id)) + .map(|rd| rd.filter_map(|e| e.ok()?.file_name().to_str()?.parse().ok()).collect()) + .unwrap_or_default(); + v.sort(); + v + } + + pub fn next_version(&self, id: &str) -> u64 { + self.versions(id).last().map_or(1, |v| v + 1) } /// Map a path in an actor's namespace to a host path. @@ -145,25 +180,28 @@ impl World { /// - `/public/...` and `/data/...` belong to `sender`. /// - Native senders (the console) can also give absolute host paths. pub fn resolve(&self, sender: &str, path: &str) -> Result { - let clean = |base: PathBuf, rest: &str| -> Result { - let mut out = base.clone(); + let under = |base: PathBuf, rest: &str| -> Result { + let mut out = base; for c in Path::new(rest).components() { match c { - std::path::Component::Normal(n) => out.push(n), - std::path::Component::CurDir | std::path::Component::RootDir => {} + Component::Normal(n) => out.push(n), + Component::CurDir | Component::RootDir => {} _ => bail!("path must not contain '..': {path}"), } } Ok(out) }; - if let Some(rest) = path.strip_prefix("/actor/") { - return clean(self.actors_root(), rest); + let prefixed = |prefix: &str| { + path.strip_prefix(prefix).filter(|rest| rest.is_empty() || rest.starts_with('/')) + }; + if let Some(rest) = prefixed("/actor") { + return under(self.actors_root(), rest); } - if let Some(rest) = path.strip_prefix("/public/").or(path.strip_prefix("/public").filter(|r| r.is_empty())) { - return clean(self.public_dir(sender), rest); + if let Some(rest) = prefixed("/public") { + return under(self.public_dir(sender), rest); } - if let Some(rest) = path.strip_prefix("/data/").or(path.strip_prefix("/data").filter(|r| r.is_empty())) { - return clean(self.data_dir(sender), rest); + if let Some(rest) = prefixed("/data") { + return under(self.data_dir(sender), rest); } if NATIVE.contains(&sender) && Path::new(path).is_absolute() { return Ok(PathBuf::from(path)); @@ -182,11 +220,8 @@ impl World { pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> { let dir = path.parent().context("path has no parent")?; std::fs::create_dir_all(dir)?; - let tmp = dir.join(format!( - "{TMP_PREFIX}host-{}-{}", - std::process::id(), - path.file_name().and_then(|n| n.to_str()).unwrap_or("file") - )); + let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("file"); + let tmp = dir.join(format!("{TMP_PREFIX}host-{}-{name}", std::process::id())); std::fs::write(&tmp, bytes)?; std::fs::rename(&tmp, path)?; Ok(()) @@ -198,16 +233,13 @@ pub fn copy_tree(from: &Path, to: &Path) -> Result<()> { for entry in std::fs::read_dir(from).with_context(|| format!("read {}", from.display()))? { let entry = entry?; let name = entry.file_name(); - let name_str = name.to_string_lossy(); - if name_str.starts_with(TMP_PREFIX) { + let name = name.to_string_lossy(); + if name.starts_with(TMP_PREFIX) { continue; } + let dst = to.join(&*name); let ty = entry.file_type()?; - let dst = to.join(&name); - if ty.is_dir() { - if name_str == "target" || name_str == ".git" { - continue; - } + if ty.is_dir() && name != "target" && name != ".git" { copy_tree(&entry.path(), &dst)?; } else if ty.is_file() { std::fs::copy(entry.path(), &dst)?; @@ -216,8 +248,8 @@ pub fn copy_tree(from: &Path, to: &Path) -> Result<()> { Ok(()) } -/// Replace the directory `dst` with a copy of `src`, as close to atomic as a -/// directory can be: copy to a sibling, then swap with two renames. +/// Replace the directory `dst` with a copy of `src` (an empty folder if `src` +/// does not exist): copy to a sibling, then swap with two renames. pub fn replace_tree(src: &Path, dst: &Path) -> Result<()> { let parent = dst.parent().context("path has no parent")?; let name = dst.file_name().and_then(|n| n.to_str()).unwrap_or("dir"); diff --git a/examples/crasher/Cargo.lock b/examples/crasher/Cargo.lock new file mode 100644 index 0000000..9ab03ef --- /dev/null +++ b/examples/crasher/Cargo.lock @@ -0,0 +1,7 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "crasher" +version = "0.1.0" diff --git a/examples/notes-v2/Cargo.lock b/examples/notes-v2/Cargo.lock new file mode 100644 index 0000000..7c1167a --- /dev/null +++ b/examples/notes-v2/Cargo.lock @@ -0,0 +1,105 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "notes" +version = "0.1.0" +dependencies = [ + "serde_json", +] + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.154" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "syn" +version = "3.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "unicode-ident" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"