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

Merge the world layout (folder is the source of truth)

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