Compare commits

..
Author SHA1 Message Date
Sausage Agent f008d0cc60 nix: lock crane; don't store the test run's target folder
ci / check (push) Successful in 39s
2026-10-17 08:02:08 +00:00
Sausage Agent 81daddfd70 nix: build with crane, dependencies cached separately
ci / check (push) Successful in 10m50s
buildDepsOnly builds the dependencies of the host and the tests once
(keyed only on Cargo.toml/Cargo.lock); packages.default and checks.tests
reuse them, so a change to our crates no longer recompiles every crate.
checks.package builds the binary the host loop runs.
2026-10-17 07:49:54 +00:00
agent-pool-9 e8df22fe06 host tests: reuse the existing drain helper
ci / check (push) Successful in 8s
2026-10-17 07:23:30 +00:00
agent-pool-9 0995824c01 Watch the world folder: auto spawn/upgrade, broker watch/unwatch, retire scan
ci / check (push) Failing after 7m36s
The host watches the whole world folder recursively with the notify crate
(any writer counts, .sausage-tmp-* skipped, ~100 ms coalescing per path).
A new or changed actor/<id>/actor.wasm (or new folder actor/<id>) is
booted as at start once quiet: new = spawn (tested first), changed =
upgrade. The broker `scan` request is gone; the boot pass stays, plus a
full pass when the OS watcher overflows.

New broker requests watch/unwatch {path} (saved in actor.json `watches`)
deliver fs_changed {path, op: write|create|remove|rename, to?} from `fs`,
not routed, dropped when the mailbox is full; overflow sends every watcher
one fs_changed {path:"/actor", op:"rescan"}.

Card #91 (multi-host step 1).
2026-10-17 07:12:04 +00:00
agent-pool-8 6a136c1b17 Borrow host descriptors while tracking VFS paths
ci / check (push) Successful in 19s
2026-10-16 21:47:17 +00:00
agent-pool-8 3e90cd73d3 Keep discovery test file at served root 2026-10-16 21:47:17 +00:00
agent-pool-8 845c961f4f Expose advertised actor-served files in directory listings 2026-10-16 21:47:17 +00:00
Sausage agent d6d3c9a2f2 Bare sausage creates ./world if needed and runs it (default app)
ci / check (push) Successful in 5s
- cli: the subcommand is optional; without one, sausage runs ./world and
  runs init first when ./world is not a world yet (test: cli/tests/cli.rs).
  So `nix run git+https://gitea.surma.technology/surma/sausage` works.
- flake: .#loop without a folder runs bare sausage.
- README: mention the bare command and the bare loop.
2026-10-16 21:09:06 +00:00
Sausage Agent 0951ba18e1 README: run straight from the public URL (nix run git+https://...), clone optional
ci / check (push) Successful in 8s
2026-10-16 20:55:39 +00:00
Sausage Agent d0876626df Public repo: .#loop fetches over public HTTPS, README for outsiders
ci / check (push) Successful in 8s
- flake.nix: .#loop runs git+https://gitea.surma.technology/surma/sausage
  (SAUSAGE_FLAKE overrides it); drop loop-internal / --internal (LAN SSH).
- README: clone step, LLM token requirement, loop URL and SAUSAGE_FLAKE.
- ci.yml: comment without internal host name.
2026-10-16 20:40:26 +00:00
15 changed files with 980 additions and 94 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
# Runs on the Nix runner on nexus (label `nixos`, host executor). # Runs on a self-hosted Nix runner (label `nixos`, host executor).
name: ci name: ci
on: [push, pull_request] on: [push, pull_request]
Generated
+97
View File
@@ -678,6 +678,15 @@ dependencies = [
"windows-sys 0.59.0", "windows-sys 0.59.0",
] ]
[[package]]
name = "fsevent-sys"
version = "4.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2"
dependencies = [
"libc",
]
[[package]] [[package]]
name = "futures" name = "futures"
version = "0.3.34" version = "0.3.34"
@@ -1066,6 +1075,26 @@ dependencies = [
"serde_core", "serde_core",
] ]
[[package]]
name = "inotify"
version = "0.11.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4cc00ea907cab49550b7da656f80ebb97be1b997d931fbcd28d39734e17ce592"
dependencies = [
"bitflags",
"inotify-sys",
"libc",
]
[[package]]
name = "inotify-sys"
version = "0.1.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d"
dependencies = [
"libc",
]
[[package]] [[package]]
name = "instant" name = "instant"
version = "0.1.13" version = "0.1.13"
@@ -1165,6 +1194,26 @@ dependencies = [
"wasm-bindgen", "wasm-bindgen",
] ]
[[package]]
name = "kqueue"
version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8d763e5b24120b4ddf50de6c92308156765aabfbbccebf401da7cff2d70a41ea"
dependencies = [
"kqueue-sys",
"libc",
]
[[package]]
name = "kqueue-sys"
version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087"
dependencies = [
"bitflags",
"libc",
]
[[package]] [[package]]
name = "leb128" name = "leb128"
version = "0.2.7" version = "0.2.7"
@@ -1259,6 +1308,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee"
dependencies = [ dependencies = [
"libc", "libc",
"log",
"wasi", "wasi",
"windows-sys 0.61.2", "windows-sys 0.61.2",
] ]
@@ -1269,6 +1319,33 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d" checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d"
[[package]]
name = "notify"
version = "8.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3"
dependencies = [
"bitflags",
"fsevent-sys",
"inotify",
"kqueue",
"libc",
"log",
"mio",
"notify-types",
"walkdir",
"windows-sys 0.60.2",
]
[[package]]
name = "notify-types"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a"
dependencies = [
"bitflags",
]
[[package]] [[package]]
name = "object" name = "object"
version = "0.40.0" version = "0.40.0"
@@ -1580,6 +1657,15 @@ version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f"
[[package]]
name = "same-file"
version = "1.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502"
dependencies = [
"winapi-util",
]
[[package]] [[package]]
name = "sausage" name = "sausage"
version = "0.1.0" version = "0.1.0"
@@ -1590,6 +1676,7 @@ dependencies = [
"http", "http",
"http-body-util", "http-body-util",
"include_dir", "include_dir",
"notify",
"serde", "serde",
"serde_json", "serde_json",
"tokio", "tokio",
@@ -2075,6 +2162,16 @@ version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7" checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7"
[[package]]
name = "walkdir"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b"
dependencies = [
"same-file",
"winapi-util",
]
[[package]] [[package]]
name = "want" name = "want"
version = "0.3.2" version = "0.3.2"
+46 -9
View File
@@ -68,7 +68,7 @@ Each instance sees three folders:
| `/actor` | read-only | the `actor/` folder of the world | | `/actor` | read-only | the `actor/` folder of the world |
`/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json` `/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json`
(installed version, running, subscriptions, tapped), `source/`, `public/`, `data/`, (installed version, running, subscriptions, tapped, watches), `source/`, `public/`, `data/`,
`stderr.log`, and `versions/<n>/`. `ls /actor` lists all actors. `stderr.log`, and `versions/<n>/`. `ls /actor` lists all actors.
Files are plain POSIX files: a write is visible at once. To publish a file Files are plain POSIX files: a write is visible at once. To publish a file
@@ -85,6 +85,13 @@ Every actor can read and write them. The host turns each file operation into a
message from `fs` to `<id>` and waits up to 10 seconds for the reply. `path` message from `fs` to `<id>` and waits up to 10 seconds for the reply. `path`
is relative to `fs/` and starts with `/` (the folder itself is `/`). is relative to `fs/` and starts with `/` (the folder itself is `/`).
To make this virtual directory discoverable, an actor that serves it declares
`Actor-served files` and `fs_read` in its leading `//!` documentation in
`source/src/main.rs`. The host uses that declaration when listing
`/actor/<id>/`; it does not probe every actor (which could stall on actors that
do not handle filesystem messages). This is discovery metadata only: opening
and reading the path still requires a valid `fs_read` reply.
| Request | Reply | | Request | Reply |
| --- | --- | | --- | --- |
| `{"type": "fs_read", "path"}` | `{"type": "fs_file", "text"}` (or `"base64"` instead of `"text"`), or `{"type": "fs_dir", "entries": [{"name", "type": "file" \| "dir"}]}` | | `{"type": "fs_read", "path"}` | `{"type": "fs_file", "text"}` (or `"base64"` instead of `"text"`), or `{"type": "fs_dir", "entries": [{"name", "type": "file" \| "dir"}]}` |
@@ -103,14 +110,15 @@ Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`.
- No reply within 10 seconds, or an actor that does not exist, makes the - No reply within 10 seconds, or an actor that does not exist, makes the
operation fail (`io`, or `no-entry`). An actor that opens its own `fs/` operation fail (`io`, or `no-entry`). An actor that opens its own `fs/`
waits for itself and fails after the timeout. waits for itself and fails after the timeout.
- `fs` is reserved: no actor can have that ID. - `fs` is reserved: no actor can have that ID. It also sends `fs_changed`
to watchers (see Watches).
## The broker ## The broker
Requests to `broker`. The answer is a message whose `reply_to` is the ID of Requests to `broker`. The answer is a message whose `reply_to` is the ID of
the request: the request's `type` in past tense (`spawned`, `upgraded`, the request: the request's `type` in past tense (`spawned`, `upgraded`,
`rolled_back`, `stopped`, `started`, `removed`, `subscribed`, `unsubscribed`, `rolled_back`, `stopped`, `started`, `removed`, `subscribed`, `unsubscribed`,
`tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`. `tapping`, `untapped`, `watching`, `unwatched`, `exiting`), or `{"type": "error", "reason": "..."}`.
| `type` | Fields | Effect | | `type` | Fields | Effect |
| --- | --- | --- | | --- | --- | --- |
@@ -121,7 +129,7 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
| `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<id>/` (all versions, `data/`, `public/`, logs); answer `{"type": "removed", "actor"}`. Messages to it are then undeliverable, and the ID can be spawned again. An unknown actor or a native one (`broker`, `console`, `cargo`, `fs`) is an `error` | | `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<id>/` (all versions, `data/`, `public/`, logs); answer `{"type": "removed", "actor"}`. Messages to it are then undeliverable, and the ID can be spawned again. An unknown actor or a native one (`broker`, `console`, `cargo`, `fs`) is an `error` |
| `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes | | `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
| `tap`, `untap` | | receive (or not) a copy of every message the broker routes | | `tap`, `untap` | | receive (or not) a copy of every message the broker routes |
| `scan` | | boot new or changed `actor.wasm` files | | `watch`, `unwatch` | `path` | receive (or not) `fs_changed` for every change of `path` or anything under it; answer `{"type": "watching" \| "unwatched", "path"}` with the path as an `/actor/...` path |
| `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) | | `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) |
Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`. Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`.
@@ -144,6 +152,31 @@ including its own), for debugging, logging, or a live view of the world:
- Like subscriptions, a tap is saved in `actor.json` (`"tapped": true`) and - Like subscriptions, a tap is saved in `actor.json` (`"tapped": true`) and
survives restarts and upgrades. `untap` ends it. survives restarts and upgrades. `untap` ends it.
**Watches.** The host watches the whole world folder with the OS file
watcher: every change inside it counts, whoever makes it (actors, a 9P mount,
a shell or `git` on the host). Actor-served files (`/actor/<id>/fs/`) are not
on disk and do not count. After `watch {path}`, the sender gets for each
change of `path` or a path under it (`/actor/a` covers `/actor/a/x`, not
`/actor/ab`):
```json
{"type": "fs_changed", "path": "/actor/notes/public/list.md", "op": "write"}
```
- `op` is `write`, `create`, `remove`, or `rename` (then `to` is the new
path; a watcher of either path gets it).
- Changes of one path within about 100 ms become one event (a `write` after
a `create` stays a `create`).
- The host's temporary files (names starting with `.sausage-tmp-`) are
skipped; renaming one onto a file is a `write` of that file.
- Events come from `fs`. Like tap copies, they are not routed, a full
mailbox drops them, and they never block the writer. Never answer them.
- If the OS drops events, every watcher gets one
`{"type": "fs_changed", "path": "/actor", "op": "rescan"}`: anything may
have changed.
- A watch is saved in `actor.json` (`"watches": [...]`) and survives
restarts and upgrades. `unwatch` with the same path ends it.
**Versions.** A new version (from `spawn`, `upgrade`, or a changed **Versions.** A new version (from `spawn`, `upgrade`, or a changed
`actor.wasm` on disk) is tested first, while the old version keeps running. `actor.wasm` on disk) is tested first, while the old version keeps running.
Only when its tests pass does the broker copy it to `versions/<n>/` Only when its tests pass does the broker copy it to `versions/<n>/`
@@ -182,11 +215,15 @@ the delay up to 30 s. A new version that crashes 3 times within 30 seconds
after an upgrade goes back to the previous version. The actor that asked for after an upgrade goes back to the previous version. The actor that asked for
the upgrade gets `{"type": "rolled_back", "from", "to", "reason", "last_exit"}`. the upgrade gets `{"type": "rolled_back", "from", "to", "reason", "last_exit"}`.
**The folder is the truth.** At start, and on `scan`, the broker boots every **The folder is the truth.** At start, the broker boots every
`actor/<id>/actor.wasm`. If the file differs from the installed version, it `actor/<id>/actor.wasm`. Afterwards, the file watcher reports a new or
is a new version, tested with the tests in `actor/<id>/source/`. If its changed `actor/<id>/actor.wasm` (or a new folder `actor/<id>/`) once it is
tests fail, the file stays where it is, the reason goes to `stderr.log`, and quiet for about 100 ms: a new file becomes a new actor, a changed one an
the next scan tries again. upgrade, tested first with the tests in `actor/<id>/source/`. A file equal
to the installed version changes nothing. If its tests fail, the file stays
where it is, the reason goes to `stderr.log`, and the next change of the
file tries again. When the OS drops events, the broker checks every actor
folder again, as at start.
## The `cargo` actor ## The `cargo` actor
+25 -9
View File
@@ -10,28 +10,43 @@ change its own code.
## Start ## Start
You need [Nix](https://nixos.org/download) with flakes. You need [Nix](https://nixos.org/download) with flakes, and an API token for
an LLM (OpenAI Responses or Anthropic Messages API). No clone needed:
```sh ```sh
nix run -- init # creates ./world and builds the agent (a few minutes) nix run git+https://gitea.surma.technology/surma/sausage
nix run -- run # starts the world and a console ```
This creates a world in `./world` the first time (it builds the agent, which
takes a few minutes), then starts it with a console. Later runs start the
existing world. The two steps on their own:
```sh
nix run git+https://gitea.surma.technology/surma/sausage -- init # creates ./world and builds the agent
nix run git+https://gitea.surma.technology/surma/sausage -- run # starts the world and a console
``` ```
`init` asks for the API shape (`openai-responses` or `anthropic-messages`), `init` asks for the API shape (`openai-responses` or `anthropic-messages`),
the base URL, the model, and the token. It reads `SAUSAGE_API`, the base URL, the model, and the token. It reads `SAUSAGE_API`,
`SAUSAGE_BASE_URL`, `SAUSAGE_MODEL`, and `SAUSAGE_API_TOKEN` first. The `SAUSAGE_BASE_URL`, `SAUSAGE_MODEL`, and `SAUSAGE_API_TOKEN` first. The
values go to `world/actor/agent/data/config.json`. Both commands take another values go to `world/actor/agent/data/config.json`. Both commands take another
folder as an argument, for example `nix run -- run my-world`. folder as an argument, for example `... -- run my-world`.
To keep a world on the newest version, run it in a loop: To keep a world on the newest version, run it in a loop:
```sh ```sh
nix run .#loop -- my-world # or .#loop-internal (Gitea on the LAN) nix run git+https://gitea.surma.technology/surma/sausage#loop # ./world, created if needed
nix run git+https://gitea.surma.technology/surma/sausage#loop -- my-world # an existing world
``` ```
The loop runs `nix run --refresh git+ssh://…/surma/sausage -- run my-world`. The loop runs `nix run --refresh git+https://gitea.surma.technology/surma/sausage -- run my-world`.
When the host crashes or an actor sends `@broker {"type":"exit"}`, the loop When the host crashes or an actor sends `@broker {"type":"exit"}`, the loop
fetches the newest commit and starts it again. `/quit` ends the loop. fetches the newest commit and starts it again. `/quit` ends the loop. Set
`SAUSAGE_FLAKE` to run another flake URL instead, for example your fork.
From a clone (`git clone https://gitea.surma.technology/surma/sausage`), the
same commands are `nix run`, `nix run -- init`, `nix run -- run` and
`nix run .#loop`.
In the console: In the console:
@@ -57,8 +72,9 @@ versions/<n>/ every installed version: actor.wasm, source/, meta.json
``` ```
To add or replace an actor by hand, put an `actor.wasm` (and its `source/`, To add or replace an actor by hand, put an `actor.wasm` (and its `source/`,
with the tests in `source/tests/`) into its folder. Then restart the world or with the tests in `source/tests/`) into its folder; the host watches the
send `@broker {"type":"scan"}`. The broker tests the file before it uses it. world folder and picks the file up once it stops changing. The broker tests
the file before it uses it.
If the tests fail, the old version keeps running, and `stderr.log` says why. If the tests fail, the old version keeps running, and `stderr.log` says why.
## More ## More
+1
View File
@@ -15,6 +15,7 @@ clap = { version = "4", features = ["derive"] }
http = "1" http = "1"
http-body-util = "0.1" http-body-util = "0.1"
include_dir = "0.7" include_dir = "0.7"
notify = "8"
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" serde_json = "1"
tokio = { version = "1", features = ["full"] } tokio = { version = "1", features = ["full"] }
+111 -23
View File
@@ -1,11 +1,12 @@
//! The broker: message routing, mailboxes, subscriptions, taps, the version //! The broker: message routing, mailboxes, subscriptions, taps, watches, the
//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start, //! version store, and the requests to `broker` (spawn, upgrade, rollback,
//! remove, subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs` //! stop, start, remove, subscribe, unsubscribe, tap, untap, watch, unwatch,
//! runs each WASM actor. //! exit). `supervisor.rs` runs each WASM actor.
//! //!
//! The broker accepts only `.wasm` files. It does not know any source language. //! The broker accepts only `.wasm` files. It does not know any source language.
//! The world folder is the source of truth: `scan` boots every //! The world folder is the source of truth: at start, the broker boots every
//! `actor/<id>/actor.wasm` and treats a changed file as an upgrade. //! `actor/<id>/actor.wasm`; afterwards the file watcher (`watch.rs`) reports
//! changes: a new `actor.wasm` becomes an actor, a changed one an upgrade.
use std::collections::{BTreeMap, HashMap}; use std::collections::{BTreeMap, HashMap};
use std::path::Path; use std::path::Path;
@@ -167,14 +168,19 @@ impl Broker {
/// they are never tapped themselves; a full mailbox drops the copy. /// they are never tapped themselves; a full mailbox drops the copy.
fn tap(&self, tappers: &[String], message: Value) { fn tap(&self, tappers: &[String], message: Value) {
let ts = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_millis() as u64); let ts = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_millis() as u64);
let body = json!({"type": "tapped", "message": message, "ts": ts}); self.inject(tappers, "broker", json!({"type": "tapped", "message": message, "ts": ts}));
}
/// Put a message of the host straight into the mailboxes of `to`: not
/// routed (so not tapped), never blocking, dropped if a mailbox is full.
fn inject(&self, to: &[String], from: &str, body: Value) {
let mailboxes = self.mailboxes.lock().unwrap(); let mailboxes = self.mailboxes.lock().unwrap();
for tapper in tappers { for id in to {
if let Some(tx) = mailboxes.get(tapper) { if let Some(tx) = mailboxes.get(id) {
let _ = tx.try_send(Envelope { let _ = tx.try_send(Envelope {
id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)), id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
from: "broker".into(), from: from.into(),
to: tapper.clone(), to: id.clone(),
reply_to: None, reply_to: None,
body: body.clone(), body: body.clone(),
}); });
@@ -406,31 +412,90 @@ impl Broker {
Ok(json!({"type": "removed", "actor": id})) Ok(json!({"type": "removed", "actor": id}))
} }
/// Boot the actors in `actor/`. Called at start and on `scan`. /// Boot the actors in `actor/`: at start, and as a full pass when the
/// file watcher dropped events (`overflowed`).
/// ///
/// - An `actor.wasm` equal to the installed version starts as is. /// - An `actor.wasm` equal to the installed version starts as is.
/// - A new or changed `actor.wasm` is tested where it lies. A new actor /// - A new or changed `actor.wasm` is tested where it lies. A new actor
/// is installed, a changed one is an upgrade. If its tests fail, /// is installed, a changed one is an upgrade. If its tests fail,
/// nothing changes, and the file stays for the next scan. /// nothing changes, and the file stays until it changes again.
pub async fn scan(self: &Arc<Self>) -> Vec<Value> { pub async fn scan(self: &Arc<Self>) -> Vec<Value> {
let mut report = Vec::new(); let mut report = Vec::new();
for id in self.world.actor_ids().unwrap_or_default() { for id in self.world.actor_ids().unwrap_or_default() {
if NATIVE.contains(&id.as_str()) || !self.world.wasm_path(&id).is_file() { if let Some(r) = self.boot_actor(&id).await {
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.push(r);
} }
} }
report report
} }
/// The file watcher saw `actor/<id>/actor.wasm` (or the folder) appear or
/// change: start, install or upgrade `id` as at boot, and log the result.
pub async fn check_actor(self: Arc<Self>, id: String) {
if let Some(result) = self.boot_actor(&id).await {
self.log(&id, &format!("actor.wasm changed: {result}"));
eprintln!("sausage: {result}");
}
}
/// The OS file watcher dropped events: tell every watcher with one
/// `fs_changed {path: "/actor", op: "rescan"}`, then run a full pass.
pub async fn overflowed(self: Arc<Self>) -> Vec<Value> {
self.host_error("the file watcher dropped events; watchers get a rescan, actors a full pass");
let watchers: Vec<String> =
self.metas.lock().unwrap().iter().filter(|(_, m)| !m.watches.is_empty()).map(|(id, _)| id.clone()).collect();
self.inject(&watchers, vfs::ID, json!({"type": "fs_changed", "path": "/actor", "op": "rescan"}));
let report = self.scan().await;
for change in &report {
eprintln!("sausage: {change}");
}
report
}
/// Send `fs_changed` for one coalesced change (from `watch.rs`) to every
/// actor that watches `path` or `to`. Like taps: straight into the
/// mailbox, dropped if it is full.
pub fn notify_watchers(&self, path: &str, op: &str, to: Option<&str>) {
let watchers: Vec<String> = {
let metas = self.metas.lock().unwrap();
metas
.iter()
.filter(|(_, m)| m.watches.iter().any(|w| under(path, w) || to.is_some_and(|t| under(t, w))))
.map(|(id, _)| id.clone())
.collect()
};
if watchers.is_empty() {
return;
}
let mut body = json!({"type": "fs_changed", "path": path, "op": op});
if let Some(to) = to {
body["to"] = json!(to);
}
self.inject(&watchers, vfs::ID, body);
}
/// A problem of the host itself: the broker's `stderr.log` and the terminal.
pub fn host_error(&self, text: &str) {
self.log("broker", text);
eprintln!("sausage: {text}");
}
async fn boot_actor(self: &Arc<Self>, id: &str) -> Option<Value> {
if NATIVE.contains(&id) || !world::valid_id(id) {
return None;
}
let lock = self.op_lock(id);
let _guard = lock.lock().await;
if !self.world.wasm_path(id).is_file() {
return None;
}
self.scan_actor(id).await
}
async fn scan_actor(self: &Arc<Self>, id: &str) -> Option<Value> { async fn scan_actor(self: &Arc<Self>, id: &str) -> Option<Value> {
let meta = self.world.read_meta(id); // The broker's copy wins: actors change theirs (subscribe, ...) without the op lock.
let meta = self.metas.lock().unwrap().entry(id.to_string()).or_insert_with(|| self.world.read_meta(id)).clone();
let installed = meta.version.filter(|v| self.world.version_dir(id, *v).join("actor.wasm").is_file()); 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 if let Some(v) = installed
&& !self.is_supervised(id) && !self.is_supervised(id)
{ {
@@ -529,14 +594,32 @@ impl Broker {
let answer = if on { "tapping" } else { "untapped" }; let answer = if on { "tapping" } else { "untapped" };
Ok(json!({"type": answer})) Ok(json!({"type": answer}))
} }
"scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})), "watch" | "unwatch" => {
let on = ty == "watch";
let path_s = b["path"].as_str().ok_or_else(|| anyhow!("missing \"path\""))?;
let global = self.world.global_path(&self.world.resolve(&env.from, path_s)?);
let path = global.ok_or_else(|| anyhow!("{path_s} is not under /actor"))?;
let path = match path.trim_end_matches('/') {
"" => "/actor".to_string(),
p => p.to_string(),
};
self.update_meta(&env.from, |m| {
if on {
m.watches.insert(path.clone());
} else {
m.watches.remove(&path);
}
})?;
let answer = if on { "watching" } else { "unwatched" };
Ok(json!({"type": answer, "path": path}))
}
"exit" => { "exit" => {
self.log("broker", &format!("exit requested by {}", env.from)); self.log("broker", &format!("exit requested by {}", env.from));
// notify_one keeps the permit if nobody waits yet. // notify_one keeps the permit if nobody waits yet.
self.exit.notify_one(); self.exit.notify_one();
Ok(json!({"type": "exiting"})) Ok(json!({"type": "exiting"}))
} }
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, scan, exit"), other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, watch, unwatch, exit"),
} }
} }
@@ -563,6 +646,11 @@ fn same_file(a: &Path, b: &Path) -> bool {
matches!((std::fs::read(a), std::fs::read(b)), (Ok(x), Ok(y)) if x == y) matches!((std::fs::read(a), std::fs::read(b)), (Ok(x), Ok(y)) if x == y)
} }
/// Whether the namespace path `path` is `prefix` or inside it.
fn under(path: &str, prefix: &str) -> bool {
path.strip_prefix(prefix).is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
}
pub fn truncate(s: &str, max: usize) -> String { pub fn truncate(s: &str, max: usize) -> String {
if s.len() <= max { if s.len() <= max {
return s.to_string(); return s.to_string();
+15 -7
View File
@@ -1,7 +1,8 @@
//! Sausage: grow a system of WASI actors by talking to it. //! Sausage: grow a system of WASI actors by talking to it.
//! //!
//! The host runs the broker, the native actors (`broker`, `cargo`, and the //! The host runs the broker, the native actors (`broker`, `cargo`, and the
//! console), and one supervisor per WASM actor. See `PROTOCOL.md`. //! console), one supervisor per WASM actor, and the file watcher of the
//! world folder (`watch.rs`). See `PROTOCOL.md`.
pub mod broker; pub mod broker;
pub mod candidate; pub mod candidate;
@@ -10,6 +11,7 @@ pub mod console;
pub mod runtime; pub mod runtime;
pub mod supervisor; pub mod supervisor;
pub mod vfs; pub mod vfs;
pub mod watch;
pub mod world; pub mod world;
use std::path::Path; use std::path::Path;
@@ -30,12 +32,13 @@ pub struct Host {
pub world: World, pub world: World,
pub broker: Arc<Broker>, pub broker: Arc<Broker>,
native: Vec<tokio::task::JoinHandle<()>>, native: Vec<tokio::task::JoinHandle<()>>,
watcher: Option<watch::Watcher>,
_lock: std::fs::File, _lock: std::fs::File,
} }
impl Host { impl Host {
/// Start the broker, `cargo`, and every registered actor of the world at /// Start the broker, `cargo`, the file watcher, and every registered
/// `root`. Only one host can run a world at a time. /// actor of the world at `root`. Only one host can run a world at a time.
pub async fn start(root: &Path) -> Result<Host> { pub async fn start(root: &Path) -> Result<Host> {
let world = World::new(std::path::absolute(root)?); let world = World::new(std::path::absolute(root)?);
if !world.exists() { if !world.exists() {
@@ -53,10 +56,14 @@ impl Host {
tokio::spawn(broker.clone().run_actor(broker.open_mailbox("broker"))), tokio::spawn(broker.clone().run_actor(broker.open_mailbox("broker"))),
tokio::spawn(cargo.run_actor(broker.clone(), broker.open_mailbox(cargo_actor::ID))), tokio::spawn(cargo.run_actor(broker.clone(), broker.open_mailbox(cargo_actor::ID))),
]; ];
// Watch first, so that no change during the boot pass is lost.
let watcher = watch::start(broker.clone())
.inspect_err(|e| broker.host_error(&format!("no file watching: {e}")))
.ok();
for change in broker.scan().await { for change in broker.scan().await {
eprintln!("sausage: {change}"); eprintln!("sausage: {change}");
} }
Ok(Host { world, broker, native, _lock: lock }) Ok(Host { world, broker, native, watcher, _lock: lock })
} }
/// Register a native actor with the ID `id` and return its mailbox. /// Register a native actor with the ID `id` and return its mailbox.
@@ -64,9 +71,10 @@ impl Host {
self.broker.open_mailbox(id) self.broker.open_mailbox(id)
} }
/// Stop the broker and `cargo` (with the requests they still run), then /// Stop the file watcher, the broker and `cargo` (with the requests they
/// every actor, gracefully. /// still run), then every actor, gracefully.
pub async fn stop(self) { pub async fn stop(mut self) {
drop(self.watcher.take());
for task in &self.native { for task in &self.native {
task.abort(); task.abort();
} }
+13 -3
View File
@@ -12,11 +12,13 @@ use sausage::{Host, cargo_actor, console};
/// The source of the agent actor. `init` builds it like any other actor. /// The source of the agent actor. `init` builds it like any other actor.
static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent"); static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
// Without a command, `sausage` runs the world in `./world` and creates it
// first (like `init`) if it does not exist yet.
#[derive(Parser)] #[derive(Parser)]
#[command(version, about)] #[command(version, about)]
struct Cli { struct Cli {
#[command(subcommand)] #[command(subcommand)]
command: Cmd, command: Option<Cmd>,
} }
#[derive(Subcommand)] #[derive(Subcommand)]
@@ -112,7 +114,15 @@ async fn run(dir: PathBuf) -> Result<()> {
#[tokio::main] #[tokio::main]
async fn main() -> Result<()> { async fn main() -> Result<()> {
match Cli::parse().command { match Cli::parse().command {
Cmd::Init { dir } => init(dir).await, Some(Cmd::Init { dir }) => init(dir).await,
Cmd::Run { dir } => run(dir).await, Some(Cmd::Run { dir }) => run(dir).await,
None => {
let dir = PathBuf::from("world");
if !dir.join("actor").is_dir() {
println!("No world in ./world yet, creating it (same as `sausage init`).");
init(dir.clone()).await?;
}
run(dir).await
}
} }
} }
+95 -15
View File
@@ -4,7 +4,8 @@
//! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`. //! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`.
//! A path under `/actor/<id>/fs/` does not reach the disk: the host asks the //! A path under `/actor/<id>/fs/` does not reach the disk: the host asks the
//! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for //! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for
//! the reply. Every other path goes to `wasmtime-wasi` unchanged. //! the reply. Other paths go to `wasmtime-wasi` unchanged except that an
//! actor directory listing advertises `fs/` when its source docs declare it.
//! //!
//! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}` //! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}`
//! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`, //! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`,
@@ -14,7 +15,7 @@
//! A file is read once when it is opened. After a change, it is written back //! A file is read once when it is opened. After a change, it is written back
//! in one `fs_write` when it is synced or closed. //! in one `fs_write` when it is synced or closed.
use std::collections::HashSet; use std::collections::HashMap;
use std::hash::{Hash, Hasher}; use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, Weak}; use std::sync::{Arc, Mutex, Weak};
@@ -41,13 +42,14 @@ pub struct VfsState {
broker: Weak<Broker>, broker: Weak<Broker>,
/// The actor of this instance, for notes in its `stderr.log`. /// The actor of this instance, for notes in its `stderr.log`.
me: String, me: String,
/// Descriptors of the `/actor` preopen. /// Native directory descriptors and their absolute paths. Keeping every
roots: HashSet<u32>, /// `/actor` child (not only the preopen) lets `fs/` work after traversal.
dirs: HashMap<u32, String>,
} }
impl VfsState { impl VfsState {
pub fn new(broker: Weak<Broker>, me: String) -> Self { pub fn new(broker: Weak<Broker>, me: String) -> Self {
Self { broker, me, roots: HashSet::new() } Self { broker, me, dirs: HashMap::new() }
} }
} }
@@ -199,6 +201,33 @@ fn join(base: &str, rel: &str) -> String {
format!("/{}", parts.join("/")) format!("/{}", parts.join("/"))
} }
/// Split a global path under an actor's virtual `fs/` into its owner and
/// actor-relative path.
fn actor_fs_path(full: &str) -> Option<(String, String)> {
let rest = full.strip_prefix("/actor/")?;
let (actor, rest) = rest.split_once('/')?;
let (fs, path) = rest.split_once('/').unwrap_or((rest, ""));
(fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{path}")))
}
fn actor_root(path: &str) -> Option<&str> {
let actor = path.strip_prefix("/actor/")?;
(!actor.is_empty() && !actor.contains('/')).then_some(actor)
}
/// An actor advertises its virtual filesystem in its leading `//!` docs.
/// This is discovery metadata; actual reads still go through `fs_read`.
fn declares_actor_fs(source: &str) -> bool {
let docs = source
.lines()
.take_while(|line| line.trim().starts_with("//!"))
.map(|line| line.trim().trim_start_matches("//!").trim())
.collect::<Vec<_>>()
.join(" ")
.to_ascii_lowercase();
docs.contains("actor-served files") && docs.contains("fs_read")
}
fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat { fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat {
types::DescriptorStat { types::DescriptorStat {
type_, type_,
@@ -266,6 +295,14 @@ impl VfsView<'_> {
Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() } Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() }
} }
fn actor_serves_fs(&self, actor: &str) -> bool {
self.state
.broker
.upgrade()
.and_then(|broker| std::fs::read_to_string(broker.world.source_dir(actor).join("src/main.rs")).ok())
.is_some_and(|source| declares_actor_fs(&source))
}
fn virt(&mut self, fd: &Resource<types::Descriptor>) -> Option<Virt> { fn virt(&mut self, fd: &Resource<types::Descriptor>) -> Option<Virt> {
self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned() self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned()
} }
@@ -282,16 +319,27 @@ impl VfsView<'_> {
match self.virt(fd) { match self.virt(fd) {
Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))), Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))),
Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))), Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))),
None if self.state.roots.contains(&fd.rep()) => { None => {
let full = join("/", path); let base = self.state.dirs.get(&fd.rep())?;
let mut parts = full.splitn(4, '/').skip(1); actor_fs_path(&join(base, path))
let (actor, fs, rest) = (parts.next()?, parts.next()?, parts.next().unwrap_or(""));
(fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{rest}")))
} }
None => None,
} }
} }
async fn native_directory_entries(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Vec<DirectoryEntry>> {
let stream = HostDescriptor::read_directory(&mut self.inner, fd).await?;
let stream_rep = stream.rep();
let mut entries = Vec::new();
loop {
match HostDirectoryEntryStream::read_directory_entry(&mut self.inner, Resource::new_borrow(stream_rep)).await? {
Some(entry) => entries.push(entry),
None => break,
}
}
HostDirectoryEntryStream::drop(&mut self.inner, stream).map_err(|_| ErrorCode::Io)?;
Ok(entries)
}
fn push(&mut self, v: Virt) -> FsResult<Resource<types::Descriptor>> { fn push(&mut self, v: Virt) -> FsResult<Resource<types::Descriptor>> {
let r = self.inner.table.push(v)?; let r = self.inner.table.push(v)?;
Ok(Resource::new_own(r.rep())) Ok(Resource::new_own(r.rep()))
@@ -343,7 +391,7 @@ impl preopens::Host for VfsView<'_> {
let dirs = preopens::Host::get_directories(&mut self.inner)?; let dirs = preopens::Host::get_directories(&mut self.inner)?;
for (fd, name) in &dirs { for (fd, name) in &dirs {
if name == "/actor" { if name == "/actor" {
self.state.roots.insert(fd.rep()); self.state.dirs.insert(fd.rep(), "/actor".to_string());
} }
} }
Ok(dirs) Ok(dirs)
@@ -453,7 +501,19 @@ impl HostDescriptor for VfsView<'_> {
Node::File(_) => Err(ErrorCode::NotDirectory.into()), Node::File(_) => Err(ErrorCode::NotDirectory.into()),
}, },
Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()), Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()),
None => HostDescriptor::read_directory(&mut self.inner, fd).await, None => {
let actor = self.state.dirs.get(&fd.rep()).and_then(|path| actor_root(path)).map(str::to_string);
if actor.as_deref().is_some_and(|id| self.actor_serves_fs(id)) {
let mut entries = self.native_directory_entries(fd).await?;
if !entries.iter().any(|entry| entry.name == "fs") {
entries.push(DirectoryEntry { type_: DescriptorType::Directory, name: "fs".to_string() });
}
let r = self.inner.table.push(VDir(entries.into_iter()))?;
Ok(Resource::new_own(r.rep()))
} else {
HostDescriptor::read_directory(&mut self.inner, fd).await
}
}
} }
} }
@@ -525,7 +585,16 @@ impl HostDescriptor for VfsView<'_> {
) -> FsResult<Resource<types::Descriptor>> { ) -> FsResult<Resource<types::Descriptor>> {
match self.target(&fd, &path) { match self.target(&fd, &path) {
Some((actor, path)) => self.open(actor, path, oflags, flags).await, Some((actor, path)) => self.open(actor, path, oflags, flags).await,
None => HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await, None => {
let child_path = self.state.dirs.get(&fd.rep()).map(|base| join(base, &path));
let opened = HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await?;
if let Some(path) = child_path {
if HostDescriptor::get_type(&mut self.inner, Resource::new_borrow(opened.rep())).await? == DescriptorType::Directory {
self.state.dirs.insert(opened.rep(), path);
}
}
Ok(opened)
}
} }
} }
@@ -538,7 +607,7 @@ impl HostDescriptor for VfsView<'_> {
} }
return Ok(()); return Ok(());
} }
self.state.roots.remove(&fd.rep()); self.state.dirs.remove(&fd.rep());
HostDescriptor::drop(&mut self.inner, fd) HostDescriptor::drop(&mut self.inner, fd)
} }
@@ -698,6 +767,17 @@ fn base64_decode(text: &str) -> Option<Vec<u8>> {
mod tests { mod tests {
use super::*; use super::*;
#[test]
fn actor_fs_paths_and_discovery_docs() {
assert_eq!(actor_fs_path("/actor/brain/fs"), Some(("brain".to_string(), "/".to_string())));
assert_eq!(actor_fs_path("/actor/brain/fs/docs/readme.md"), Some(("brain".to_string(), "/docs/readme.md".to_string())));
assert_eq!(actor_fs_path("/actor/brain/public/fs/file"), None);
assert_eq!(actor_root("/actor/brain"), Some("brain"));
assert_eq!(actor_root("/actor/brain/fs"), None);
assert!(declares_actor_fs("//! brain\n//! Actor-served files include fs_read.\nfn main() {}"));
assert!(!declares_actor_fs("//! actor without fs support\nfn main() {}"));
}
#[test] #[test]
fn base64_round_trip() { fn base64_round_trip() {
for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] { for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] {
+256
View File
@@ -0,0 +1,256 @@
//! File watching: the host watches the whole world folder with the OS file
//! watcher (`notify`, inotify on Linux), whoever writes: actors, a 9P mount,
//! a shell or `git` on the host. Nothing outside the folder counts.
//!
//! Events are coalesced per path: once a path is quiet for `QUIET` (or has
//! kept changing for `MAX_DELAY`), it becomes one change
//! `{path, op: write | create | remove | rename, to?}` with a namespace path
//! (`/actor/...`). Then the broker
//!
//! - sends `fs_changed` to the actors that watch the path (`watch` request),
//! - starts or upgrades an actor whose `actor/<id>/actor.wasm` (or folder
//! `actor/<id>`) changed, tested first, as at boot (`Broker::check_actor`).
//!
//! The host's temporary files (`.sausage-tmp-*`) are skipped: a rename of one
//! onto a file is a `write` of that file. If the OS drops events, the broker
//! runs a full pass and tells every watcher (`Broker::overflowed`).
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use notify::event::{EventKind, ModifyKind, RenameMode};
use notify::{RecursiveMode, Watcher as _};
use tokio::sync::mpsc;
use tokio::time::Instant;
use crate::broker::Broker;
use crate::world::TMP_PREFIX;
/// A path that is quiet this long becomes one change.
pub const QUIET: Duration = Duration::from_millis(100);
/// A path that keeps changing becomes a change at least this often.
pub const MAX_DELAY: Duration = Duration::from_secs(1);
/// The running watcher. Dropping it stops watching.
pub struct Watcher {
_os: notify::RecommendedWatcher,
task: tokio::task::JoinHandle<()>,
}
impl Drop for Watcher {
fn drop(&mut self) {
self.task.abort();
}
}
/// Watch the world folder of `broker` recursively. A failure to watch part
/// of it (for example too few inotify watches) is logged, and the rest is
/// watched.
pub fn start(broker: Arc<Broker>) -> notify::Result<Watcher> {
let (tx, rx) = mpsc::unbounded_channel();
let mut os = notify::recommended_watcher(move |event| {
let _ = tx.send(event);
})?;
if let Err(e) = os.watch(&broker.world.root, RecursiveMode::Recursive) {
broker.host_error(&format!("cannot watch all of {}: {e}", broker.world.root.display()));
}
let task = tokio::spawn(run(broker, rx));
Ok(Watcher { _os: os, task })
}
/// A change waiting for its path to be quiet.
struct Pending {
op: &'static str,
to: Option<String>,
first: Instant,
last: Instant,
}
impl Pending {
fn due(&self) -> Instant {
(self.last + QUIET).min(self.first + MAX_DELAY)
}
}
async fn run(broker: Arc<Broker>, mut events: mpsc::UnboundedReceiver<notify::Result<notify::Event>>) {
let root = broker.world.actors_root();
let mut pending: HashMap<String, Pending> = HashMap::new();
// Actor checks and full passes. Dropping the set (when the host stops) aborts them.
let mut tasks = tokio::task::JoinSet::new();
loop {
let due = pending.values().map(Pending::due).min();
tokio::select! {
event = events.recv() => match event {
Some(Ok(event)) => {
if event.need_rescan() {
let broker = broker.clone();
tasks.spawn(async move {
broker.overflowed().await;
});
}
let now = Instant::now();
for (path, op, to) in changes(&root, &event) {
add(&mut pending, path, op, to, now);
}
}
Some(Err(e)) => broker.host_error(&format!("file watcher: {e}")),
None => break,
},
_ = sleep_until(due) => {
let now = Instant::now();
let ready: Vec<String> = pending.iter().filter(|(_, p)| p.due() <= now).map(|(k, _)| k.clone()).collect();
for path in ready {
let p = pending.remove(&path).unwrap();
let op = if p.op == "moved-in" { "create" } else { p.op };
broker.notify_watchers(&path, op, p.to.as_deref());
if let Some(id) = new_wasm(&path, op, p.to.as_deref()) {
tasks.spawn(broker.clone().check_actor(id));
}
}
}
Some(_) = tasks.join_next() => {}
}
}
}
async fn sleep_until(t: Option<Instant>) {
match t {
Some(t) => tokio::time::sleep_until(t).await,
None => std::future::pending().await,
}
}
/// The namespace path (`/actor/...`) of a host path, unless it is outside
/// `actor/` or a temporary file of the host.
fn namespace(root: &Path, path: &Path) -> Option<String> {
let rel = path.strip_prefix(root).ok()?;
let mut out = String::from("/actor");
for part in rel.components() {
let name = part.as_os_str().to_str()?;
if name.starts_with(TMP_PREFIX) {
return None;
}
out.push('/');
out.push_str(name);
}
Some(out)
}
/// One change: namespace path, op, and the new path of a rename.
type Change = (String, &'static str, Option<String>);
/// The changes of one OS event. inotify reports a rename three times:
/// `From` (the old name), `To` (the new name) and `Both`. `To` becomes the
/// internal op `moved-in`, which `Both` replaces.
fn changes(root: &Path, event: &notify::Event) -> Vec<Change> {
let path = |i: usize| event.paths.get(i).and_then(|p| namespace(root, p));
let one = |op: &'static str| -> Vec<Change> { path(0).map(|p| (p, op, None)).into_iter().collect() };
match event.kind {
EventKind::Create(_) => one("create"),
EventKind::Remove(_) => one("remove"),
EventKind::Modify(ModifyKind::Metadata(_)) => vec![],
EventKind::Modify(ModifyKind::Name(RenameMode::From)) => one("remove"),
EventKind::Modify(ModifyKind::Name(RenameMode::To)) => one("moved-in"),
EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => match (path(0), path(1)) {
(Some(from), Some(to)) => vec![(from, "rename", Some(to))],
// Renamed to a temporary name: gone.
(Some(from), None) => vec![(from, "remove", None)],
// A temporary file renamed onto `to`: an atomic write.
(None, Some(to)) => vec![(to, "write", None)],
(None, None) => vec![],
},
EventKind::Modify(_) => one("write"),
_ => vec![],
}
}
/// Coalesce a change into `pending`. The last op wins, except that a write
/// after a create stays a create. A rename replaces the `moved-in` of its
/// new name.
fn add(pending: &mut HashMap<String, Pending>, path: String, op: &'static str, to: Option<String>, now: Instant) {
if let Some(to) = &to
&& pending.get(to).is_some_and(|p| p.op == "moved-in")
{
pending.remove(to);
}
match pending.get_mut(&path) {
Some(p) => {
if !(p.op == "create" && op == "write") {
p.op = op;
p.to = to;
}
p.last = now;
}
None => {
pending.insert(path, Pending { op, to, first: now, last: now });
}
}
}
/// The actor to check when `actor/<id>/actor.wasm` or the folder `actor/<id>`
/// appeared or changed.
fn new_wasm(path: &str, op: &str, to: Option<&str>) -> Option<String> {
let target = match op {
"remove" => return None,
"rename" => to?,
_ => path,
};
let rest = target.strip_prefix("/actor/")?;
let id = rest.strip_suffix("/actor.wasm").unwrap_or(rest);
(!id.is_empty() && !id.contains('/')).then(|| id.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn wasm_and_folder_changes_name_the_actor() {
assert_eq!(new_wasm("/actor/a/actor.wasm", "write", None).as_deref(), Some("a"));
assert_eq!(new_wasm("/actor/a", "create", None).as_deref(), Some("a"));
assert_eq!(new_wasm("/actor/x/y", "rename", Some("/actor/b/actor.wasm")).as_deref(), Some("b"));
assert_eq!(new_wasm("/actor/a/actor.wasm", "remove", None), None);
assert_eq!(new_wasm("/actor/a/versions/1/actor.wasm", "write", None), None);
assert_eq!(new_wasm("/actor/a/public/x", "write", None), None);
assert_eq!(new_wasm("/actor", "write", None), None);
}
#[test]
fn a_rename_is_one_change_and_temp_files_are_skipped() {
let root = Path::new("/w/actor");
let now = Instant::now();
let mut pending = HashMap::new();
let ev = |kind, paths: &[&str]| {
let mut e = notify::Event::new(kind);
for p in paths {
e = e.add_path(std::path::PathBuf::from(*p));
}
e
};
let name = |mode| EventKind::Modify(ModifyKind::Name(mode));
for e in [
ev(name(RenameMode::From), &["/w/actor/a/public/x"]),
ev(name(RenameMode::To), &["/w/actor/a/public/y"]),
ev(name(RenameMode::Both), &["/w/actor/a/public/x", "/w/actor/a/public/y"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/actor/a/.sausage-tmp-host-1-actor.json"]),
ev(name(RenameMode::Both), &["/w/actor/a/.sausage-tmp-host-1-actor.json", "/w/actor/a/actor.json"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/.lock"]),
] {
for (path, op, to) in changes(root, &e) {
add(&mut pending, path, op, to, now);
}
}
let mut got: Vec<(String, &str, Option<String>)> =
pending.into_iter().map(|(k, p)| (k, p.op, p.to)).collect();
got.sort();
assert_eq!(
got,
vec![
("/actor/a/actor.json".to_string(), "write", None),
("/actor/a/public/x".to_string(), "rename", Some("/actor/a/public/y".to_string())),
]
);
}
}
+6 -2
View File
@@ -1,11 +1,12 @@
//! On-disk layout of a Sausage world. The folder is the source of truth: //! On-disk layout of a Sausage world. The folder is the source of truth:
//! `sausage run` boots every actor that it finds in `actor/`. //! `sausage run` boots every actor that it finds in `actor/`, and then
//! watches the folder (`watch.rs`).
//! //!
//! ```text //! ```text
//! <world>/ //! <world>/
//! actor/<id>/ //! actor/<id>/
//! actor.wasm the current version //! actor.wasm the current version
//! actor.json version, running, subscriptions, tapped (broker) //! actor.json version, running, subscriptions, tapped, watches (broker)
//! source/ source of the current version //! source/ source of the current version
//! public/ written by the actor, read-only for others //! public/ written by the actor, read-only for others
//! data/ state, writable only by the actor //! data/ state, writable only by the actor
@@ -46,6 +47,9 @@ pub struct ActorMeta {
/// Whether this actor gets a copy of every routed message (`tap`). /// Whether this actor gets a copy of every routed message (`tap`).
#[serde(default)] #[serde(default)]
pub tapped: bool, pub tapped: bool,
/// Namespace paths whose changes this actor gets as `fs_changed` (`watch`).
#[serde(default)]
pub watches: BTreeSet<String>,
} }
fn yes() -> bool { fn yes() -> bool {
+41
View File
@@ -0,0 +1,41 @@
//! Tests of the `sausage` command line.
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::time::{SystemTime, UNIX_EPOCH};
fn empty_dir(label: &str) -> PathBuf {
let nonce = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.parent()
.unwrap()
.join("target/test-cli")
.join(format!("{label}-{}-{nonce}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
dir
}
/// Bare `sausage` (as in `nix run <flake>`) without a world starts `init`
/// for ./world instead of printing a usage error.
#[test]
fn bare_command_without_a_world_starts_init() {
let dir = empty_dir("bare");
let output = Command::new(env!("CARGO_BIN_EXE_sausage"))
.current_dir(&dir)
.env_remove("SAUSAGE_API")
.env_remove("SAUSAGE_BASE_URL")
.env_remove("SAUSAGE_MODEL")
.env_remove("SAUSAGE_API_TOKEN")
.stdin(Stdio::null())
.output()
.unwrap();
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
// With no answers on stdin, init stops at the first required value.
assert!(!output.status.success(), "stdout:\n{stdout}\nstderr:\n{stderr}");
assert!(stdout.contains("No world in ./world yet"), "stdout:\n{stdout}\nstderr:\n{stderr}");
assert!(stderr.contains("SAUSAGE_BASE_URL"), "stdout:\n{stdout}\nstderr:\n{stderr}");
assert!(!stderr.contains("Usage:"), "stderr:\n{stderr}");
// It asks before it creates anything.
assert!(!dir.join("world").exists());
}
+216 -4
View File
@@ -108,15 +108,15 @@ async fn routes_replies_publishes_and_reports_undeliverable() {
async fn broker_does_not_answer_messages_with_reply_to() { async fn broker_does_not_answer_messages_with_reply_to() {
let world = create_world("broker-reply-to"); let world = create_world("broker-reply-to");
let (host, mut mailbox) = start(&world).await; let (host, mut mailbox) = start(&world).await;
let scanned = request(&host, &mut mailbox, json!({"type": "scan"})).await; let untapped = request(&host, &mut mailbox, json!({"type": "untap"})).await;
assert_eq!(scanned.body["type"], "scanned"); assert_eq!(untapped.body["type"], "untapped");
host.broker.send( host.broker.send(
"console", "console",
Some("broker"), Some("broker"),
None, None,
Some("upstream-message"), Some("upstream-message"),
json!({"type": "scan"}), json!({"type": "untap"}),
); );
assert!( assert!(
tokio::time::timeout(Duration::from_secs(1), mailbox.recv()) tokio::time::timeout(Duration::from_secs(1), mailbox.recv())
@@ -246,6 +246,71 @@ async fn actors_serve_files_under_fs() {
host.stop().await; host.stop().await;
} }
#[tokio::test]
async fn actor_fs_is_discoverable_only_when_its_source_docs_advertise_it() {
let world = create_world("actor-fs-discovery");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let advertised = passing_source(&world, "advertised-fs");
std::fs::create_dir_all(advertised.join("src")).unwrap();
std::fs::write(
advertised.join("src/main.rs"),
"//! test actor\n//! Actor-served files: implements fs_read and fs_dir.\n",
)
.unwrap();
spawn_type(&host, &mut mailbox, "served", &wasm, &advertised).await;
let undocumented = passing_source(&world, "undocumented-fs");
std::fs::create_dir_all(undocumented.join("src")).unwrap();
std::fs::write(undocumented.join("src/main.rs"), "//! test actor without a virtual filesystem\n").unwrap();
spawn_type(&host, &mut mailbox, "plain", &wasm, &undocumented).await;
let reader_source = passing_source(&world, "fs-reader");
spawn_type(&host, &mut mailbox, "reader", &wasm, &reader_source).await;
let served_entries = ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "list", "path": "/actor/served"}),
"entries",
)
.await;
assert!(served_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
let plain_entries = ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "list", "path": "/actor/plain"}),
"entries",
)
.await;
assert!(!plain_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
let path = "/actor/served/fs/readme.txt";
ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "write", "path": path, "text": "virtual document"}),
"ok",
)
.await;
let document = ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "read", "path": path}),
"content",
)
.await;
assert_eq!(document.body["text"], "virtual document");
host.stop().await;
}
#[tokio::test] #[tokio::test]
async fn files_are_plain_posix_files() { async fn files_are_plain_posix_files() {
let (_, host, mut mailbox) = filesystem_world("posix").await; let (_, host, mut mailbox) = filesystem_world("posix").await;
@@ -546,7 +611,10 @@ async fn unknown_broker_request_lists_the_known_ones() {
let unknown = request(&host, &mut mailbox, json!({"type": "frobnicate"})).await; let unknown = request(&host, &mut mailbox, json!({"type": "frobnicate"})).await;
assert_eq!(unknown.body["type"], "error"); assert_eq!(unknown.body["type"], "error");
let reason = unknown.body["reason"].as_str().unwrap(); let reason = unknown.body["reason"].as_str().unwrap();
assert!(reason.contains("remove") && reason.contains("tap"), "{reason}"); assert!(reason.contains("remove") && reason.contains("tap") && reason.contains("watch"), "{reason}");
// `scan` is retired: the host watches the world folder instead.
let scan = request(&host, &mut mailbox, json!({"type": "scan"})).await;
assert_eq!(scan.body["type"], "error");
host.stop().await; host.stop().await;
} }
@@ -909,3 +977,147 @@ async fn a_full_tapper_mailbox_drops_copies_without_blocking() {
assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped"); assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped");
host.stop().await; host.stop().await;
} }
/// Ask `actor` again and again (it may not exist yet) until `want` holds for
/// the answer.
async fn ask_until(
host: &Host,
mailbox: &mut mpsc::Receiver<Envelope>,
actor: &str,
body: Value,
want: impl Fn(&Value) -> bool,
) -> Envelope {
tokio::time::timeout(Duration::from_secs(30), async {
loop {
let reply = ask(host, mailbox, actor, body.clone()).await;
if want(&reply.body) {
return reply;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
})
.await
.unwrap_or_else(|_| panic!("{actor} never gave the wanted answer"))
}
/// The `fs_changed` bodies among `messages` (all from `fs`).
fn fs_events(messages: Vec<Envelope>) -> Vec<Value> {
messages
.into_iter()
.filter(|m| m.body["type"] == "fs_changed")
.map(|m| {
assert_eq!(m.from, "fs");
m.body
})
.collect()
}
#[tokio::test]
async fn a_new_actor_wasm_on_disk_becomes_an_actor() {
let world = create_world("watch-spawn");
let (host, mut mailbox) = start(&world).await;
// Any writer counts: here the test itself, like a shell on the host.
let tests = world.source_dir("dropped").join("tests");
std::fs::create_dir_all(&tests).unwrap();
let test = json!({"steps": [{"send": {"type": "echo", "value": "candidate"}, "expect": {"type": "echo", "value": "candidate"}}]});
std::fs::write(tests.join("candidate.json"), test.to_string()).unwrap();
std::fs::copy(toolbox_wasm(), world.wasm_path("dropped")).unwrap();
let echo = json!({"type": "echo", "value": "hi"});
let reply = ask_until(&host, &mut mailbox, "dropped", echo, |b| b["type"] == "echo").await;
assert_eq!(reply.from, "dropped");
assert_eq!(world.read_meta("dropped").version, Some(1));
let meta = std::fs::read_to_string(world.version_dir("dropped", 1).join("meta.json")).unwrap();
assert_eq!(serde_json::from_str::<Value>(&meta).unwrap()["requested_by"], "scan");
host.stop().await;
}
#[tokio::test]
async fn a_changed_actor_wasm_on_disk_is_an_upgrade() {
let world = create_world("watch-upgrade");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let source = passing_source(&world, "source");
spawn_type(&host, &mut mailbox, "changing", &wasm, &source).await;
// The same program plus a custom section: a different file.
let mut changed = std::fs::read(&wasm).unwrap();
changed.extend_from_slice(&[0, 4, 1, b'x', b'v', b'2']);
std::fs::write(world.wasm_path("changing"), changed).unwrap();
let version = json!({"type": "version"});
ask_until(&host, &mut mailbox, "changing", version, |b| b["version"] == "2").await;
assert_eq!(world.read_meta("changing").version, Some(2));
assert_eq!(world.versions("changing"), vec![1, 2]);
host.stop().await;
}
#[tokio::test]
async fn watchers_get_one_event_per_change_from_any_writer() {
let (world, host, mut mailbox) = filesystem_world("watch-events").await;
let watch = json!({"type": "watch", "path": "/actor/writer/public/"});
let watching = request(&host, &mut mailbox, watch).await;
assert_eq!(watching.body, json!({"type": "watching", "path": "/actor/writer/public"}));
assert!(world.read_meta("console").watches.contains("/actor/writer/public"));
drain(&mut mailbox, Duration::from_millis(300)).await;
let quiet = Duration::from_millis(600);
let event = |path: &str, op: &str| json!({"type": "fs_changed", "path": format!("/actor/writer/public/{path}"), "op": op});
// An actor creates a file: one create, not create + write.
let write = json!({"type": "write", "path": "/public/a.txt", "text": "a"});
ask_type(&host, &mut mailbox, "writer", write, "ok").await;
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("a.txt", "create")]);
// The host (a shell, git, ...) changes it.
std::fs::write(world.public_dir("writer").join("a.txt"), "b").unwrap();
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("a.txt", "write")]);
// A rename is one event with `to`.
let rename = json!({"type": "rename", "path": "/public/a.txt", "to": "/public/b.txt"});
ask_type(&host, &mut mailbox, "writer", rename, "ok").await;
let mut renamed = event("a.txt", "rename");
renamed["to"] = json!("/actor/writer/public/b.txt");
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![renamed]);
std::fs::remove_file(world.public_dir("writer").join("b.txt")).unwrap();
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("b.txt", "remove")]);
// Nothing outside the watched path.
let outside = json!({"type": "write", "path": "/data/x", "text": "x"});
ask_type(&host, &mut mailbox, "writer", outside, "ok").await;
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), Vec::<Value>::new());
let unwatch = json!({"type": "unwatch", "path": "/actor/writer/public"});
let unwatched = request(&host, &mut mailbox, unwatch).await;
assert_eq!(unwatched.body, json!({"type": "unwatched", "path": "/actor/writer/public"}));
assert!(world.read_meta("console").watches.is_empty());
let write = json!({"type": "write", "path": "/public/c.txt", "text": "c"});
ask_type(&host, &mut mailbox, "writer", write, "ok").await;
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), Vec::<Value>::new());
host.stop().await;
}
#[tokio::test]
async fn a_watcher_overflow_sends_one_rescan_and_runs_a_full_pass() {
let world = create_world("watch-overflow");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let mut bystander = host.mailbox("bystander");
let watch = json!({"type": "watch", "path": "/actor/elsewhere"});
assert_eq!(request(&host, &mut mailbox, watch).await.body["type"], "watching");
// A dropped actor that the full pass finds (before the watcher's quiet time).
let tests = world.source_dir("planted").join("tests");
std::fs::create_dir_all(&tests).unwrap();
let test = json!({"steps": [{"send": {"type": "echo", "value": "candidate"}, "expect": {"type": "echo", "value": "candidate"}}]});
std::fs::write(tests.join("candidate.json"), test.to_string()).unwrap();
std::fs::copy(&wasm, world.wasm_path("planted")).unwrap();
let report = host.broker.clone().overflowed().await;
assert!(report.iter().any(|r| r["type"] == "spawned" && r["actor"] == "planted"), "{report:?}");
let rescan = json!({"type": "fs_changed", "path": "/actor", "op": "rescan"});
assert_eq!(fs_events(drain(&mut mailbox, Duration::from_millis(600)).await), vec![rescan]);
assert!(bystander.try_recv().is_err(), "an actor without watches got the rescan");
host.stop().await;
}
Generated
+16
View File
@@ -1,5 +1,20 @@
{ {
"nodes": { "nodes": {
"crane": {
"locked": {
"lastModified": 1791754810,
"narHash": "sha256-cN0Mgk+sQtCJcGWs9H4M0CnLRAuC4I8aCwiBTcBJs6E=",
"owner": "ipetkov",
"repo": "crane",
"rev": "bd63bf1d0a0f6dcacd7549c714a671ee63b0db6a",
"type": "github"
},
"original": {
"owner": "ipetkov",
"repo": "crane",
"type": "github"
}
},
"flake-utils": { "flake-utils": {
"inputs": { "inputs": {
"systems": "systems" "systems": "systems"
@@ -36,6 +51,7 @@
}, },
"root": { "root": {
"inputs": { "inputs": {
"crane": "crane",
"flake-utils": "flake-utils", "flake-utils": "flake-utils",
"nixpkgs": "nixpkgs", "nixpkgs": "nixpkgs",
"rust-overlay": "rust-overlay" "rust-overlay": "rust-overlay"
+41 -21
View File
@@ -6,9 +6,10 @@
flake-utils.url = "github:numtide/flake-utils"; flake-utils.url = "github:numtide/flake-utils";
rust-overlay.url = "github:oxalica/rust-overlay"; rust-overlay.url = "github:oxalica/rust-overlay";
rust-overlay.inputs.nixpkgs.follows = "nixpkgs"; rust-overlay.inputs.nixpkgs.follows = "nixpkgs";
crane.url = "github:ipetkov/crane";
}; };
outputs = { nixpkgs, flake-utils, rust-overlay, ... }: outputs = { nixpkgs, flake-utils, rust-overlay, crane, ... }:
flake-utils.lib.eachSystem flake-utils.lib.allSystems (system: flake-utils.lib.eachSystem flake-utils.lib.allSystems (system:
let let
pkgs = import nixpkgs { pkgs = import nixpkgs {
@@ -20,13 +21,28 @@
targets = [ "wasm32-wasip2" ]; targets = [ "wasm32-wasip2" ];
extensions = [ "rust-src" ]; extensions = [ "rust-src" ];
}; };
rustPlatform = pkgs.makeRustPlatform { cargo = rust; rustc = rust; }; # Crane builds the dependencies once (`cargoArtifacts`, keyed only on
sausage = rustPlatform.buildRustPackage { # Cargo.toml/Cargo.lock), so a change to our own crates rebuilds
# just them.
craneLib = (crane.mkLib pkgs).overrideToolchain rust;
common = {
pname = "sausage"; pname = "sausage";
version = "0.1.0"; version = "0.1.0";
src = ./.; src = ./.;
cargoLock.lockFile = ./Cargo.lock; };
cargoBuildFlags = [ "-p" "sausage" ]; # The test packages for `checks.tests`.
testPackages = "-p sausage -p sausage-actor -p agent";
cargoArtifacts = craneLib.buildDepsOnly (common // {
doCheck = false;
# The dependencies of the host binary and of the tests.
buildPhaseCargoCommand = ''
cargoWithProfile build --locked -p sausage
cargoWithProfile test --locked --no-run ${testPackages}
'';
});
sausage = craneLib.buildPackage (common // {
inherit cargoArtifacts;
cargoExtraArgs = "--locked -p sausage";
doCheck = false; doCheck = false;
nativeBuildInputs = [ pkgs.makeWrapper ]; nativeBuildInputs = [ pkgs.makeWrapper ];
# The `cargo` actor runs cargo at runtime. Build scripts need a C linker. # The `cargo` actor runs cargo at runtime. Build scripts need a C linker.
@@ -34,20 +50,21 @@
wrapProgram $out/bin/sausage --prefix PATH : ${pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ]} wrapProgram $out/bin/sausage --prefix PATH : ${pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ]}
''; '';
meta.mainProgram = "sausage"; meta.mainProgram = "sausage";
}; });
# `nix run .#loop -- <world>` runs the newest `sausage run <world>` # `nix run .#loop -- <world>` runs the newest `sausage run <world>`
# from the Gitea repo, again and again: after a crash or a broker # (without a folder: bare `sausage`, i.e. ./world, created if needed)
# from the public repo, again and again: after a crash or a broker
# `exit` request it refreshes the flake and starts the host again, # `exit` request it refreshes the flake and starts the host again,
# so the world updates itself. `/quit` (exit code 0) ends the loop. # so the world updates itself. `/quit` (exit code 0) ends the loop.
# `--internal` (or `.#loop-internal`) uses Gitea's SSH port on the LAN. # SAUSAGE_FLAKE overrides where it fetches from (any flake URL,
publicUrl = "git+ssh://containeruser@gitea.surma.technology:2222/surma/sausage"; # e.g. a fork or a local checkout).
internalUrl = "git+ssh://containeruser@10.0.0.2:2222/surma/sausage"; repoUrl = "git+https://gitea.surma.technology/surma/sausage";
loop = name: url: pkgs.writeShellScriptBin name '' loop = pkgs.writeShellScriptBin "sausage-loop" ''
url="${url}" url="''${SAUSAGE_FLAKE:-${repoUrl}}"
if [ "''${1:-}" = "--internal" ]; then url="${internalUrl}"; shift; fi # No folder: bare `sausage` (creates ./world first if needed).
if [ $# -eq 0 ]; then set -- world; fi if [ $# -gt 0 ]; then set -- run "$@"; fi
while true; do while true; do
nix run --refresh "$url#default" -- run "$@" nix run --refresh "$url#default" -- "$@"
code=$? code=$?
if [ "$code" -eq 0 ]; then exit 0; fi if [ "$code" -eq 0 ]; then exit 0; fi
echo "sausage-loop: sausage exited with $code, starting the newest version in 2 s …" >&2 echo "sausage-loop: sausage exited with $code, starting the newest version in 2 s …" >&2
@@ -61,10 +78,14 @@
# `nix flake check` runs every test of the host, the SDK, and the # `nix flake check` runs every test of the host, the SDK, and the
# agent. The tests build WASM actors with cargo, offline, from the # agent. The tests build WASM actors with cargo, offline, from the
# vendored crates of Cargo.lock. # vendored crates of Cargo.lock.
checks.tests = sausage.overrideAttrs (old: { # The binary the host loop runs (`nix run`), built from the same
pname = "sausage-tests"; # cached dependencies as the tests.
doCheck = true; checks.package = sausage;
cargoTestFlags = [ "-p" "sausage" "-p" "sausage-actor" "-p" "agent" ]; checks.tests = craneLib.cargoTest (common // {
inherit cargoArtifacts;
cargoExtraArgs = "--locked ${testPackages}";
# Nothing builds on the test run's target folder: don't store it.
doInstallCargoArtifacts = false;
preCheck = '' preCheck = ''
# Its own lock file pins crates that the vendored set may lack. # Its own lock file pins crates that the vendored set may lack.
rm -f cli/tests/actors/toolbox/Cargo.lock rm -f cli/tests/actors/toolbox/Cargo.lock
@@ -72,8 +93,7 @@
}); });
apps.default = { type = "app"; program = "${sausage}/bin/sausage"; }; apps.default = { type = "app"; program = "${sausage}/bin/sausage"; };
apps.loop = { type = "app"; program = "${loop "sausage-loop" publicUrl}/bin/sausage-loop"; }; apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };
apps.loop-internal = { type = "app"; program = "${loop "sausage-loop-internal" internalUrl}/bin/sausage-loop-internal"; };
devShells.default = pkgs.mkShell { devShells.default = pkgs.mkShell {
packages = [ rust pkgs.wasmtime pkgs.wasm-tools ]; packages = [ rust pkgs.wasmtime pkgs.wasm-tools ];