Compare commits

..
Author SHA1 Message Date
Sausage Agent 0bd046ed9f broker: add stopped-preserving upgrade without actor execution
ci / check (push) Failing after 1m47s
2026-10-22 06:49:58 +00:00
Sausage Agent a72e42a8c9 headless: register signal handlers synchronously, split tests, drop deployment examples and wording
ci / check (pull_request) Successful in 1m36s
ci / check (push) Successful in 11s
2026-10-18 15:31:41 +00:00
Sausage Agent 01a821ab72 Add --headless mode (no stdin read, runs until SIGTERM/exit) + systemd/NixOS examples
ci / check (push) Successful in 2m44s
2026-10-18 15:24:20 +00:00
Sausage Agent c204e1ef0c docs: clarify actor data is world-readable
ci / check (pull_request) Successful in 12s
ci / check (push) Successful in 6s
2026-10-17 17:22:50 +00:00
Sausage Agent 6923ca3e60 preserve the packaged runtime Rust toolchain
ci / check (push) Successful in 6s
2026-10-17 16:36:08 +00:00
Sausage Agent c504532513 drive packaged host runtime check over pipes
ci / check (push) Failing after 1m16s
2026-10-17 16:33:31 +00:00
Sausage Agent eb35a79698 trace the packaged host runtime check
ci / check (push) Failing after 1m47s
2026-10-17 16:29:02 +00:00
Sausage Agent bfcd71e2f4 test packaged host cargo runtime with an empty PATH
ci / check (push) Failing after 1m33s
2026-10-17 16:25:14 +00:00
agent-pool-12 172a776eec cargo: annotate test summary count parsing
ci / check (push) Successful in 5s
2026-10-17 16:00:54 +00:00
agent-pool-12 6a9d35ceb8 cargo: preserve build failures in result wrapper
ci / check (push) Failing after 6m15s
2026-10-17 15:52:37 +00:00
agent-pool-12 22590c608d cargo: run host unit tests before builds
ci / check (push) Failing after 8m51s
2026-10-17 15:50:09 +00:00
Sausage Agent 3bc462ffd2 agent seed: thinking levels low|medium|high|xhigh|max|off on both APIs (card #128)
ci / check (push) Successful in 5s
New config field `thinking` (OpenAI: reasoning_effort still wins): Anthropic adaptive thinking +
output_config.effort (thinking blocks kept and sent back), OpenAI reasoning.effort. A 400 that rejects
the effort is retried with the nearest accepted level (from the error's list, lower on a tie), logged,
never dropped to off.
2026-10-17 15:28:44 +00:00
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
24 changed files with 1507 additions and 197 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
on: [push, pull_request]
Generated
+97
View File
@@ -678,6 +678,15 @@ dependencies = [
"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]]
name = "futures"
version = "0.3.34"
@@ -1066,6 +1075,26 @@ dependencies = [
"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]]
name = "instant"
version = "0.1.13"
@@ -1165,6 +1194,26 @@ dependencies = [
"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]]
name = "leb128"
version = "0.2.7"
@@ -1259,6 +1308,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee"
dependencies = [
"libc",
"log",
"wasi",
"windows-sys 0.61.2",
]
@@ -1269,6 +1319,33 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
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]]
name = "object"
version = "0.40.0"
@@ -1580,6 +1657,15 @@ version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
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]]
name = "sausage"
version = "0.1.0"
@@ -1590,6 +1676,7 @@ dependencies = [
"http",
"http-body-util",
"include_dir",
"notify",
"serde",
"serde_json",
"tokio",
@@ -2075,6 +2162,16 @@ version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
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]]
name = "want"
version = "0.3.2"
+77 -17
View File
@@ -63,12 +63,14 @@ Each instance sees three folders:
| Path | Access | Contents |
| --- | --- | --- |
| `/data` | read-write | the actor's state |
| `/data` | read-write by its actor; readable by every actor | the actor's state |
| `/public` | read-write | the files that the actor publishes |
| `/actor` | read-only | the `actor/` folder of the world |
An actor can write only its own `/data`, but can read every actor's `/data` through `/actor/<id>/data/`. `/data` is not private: any unprotected secret file there is readable by every actor.
`/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.
Files are plain POSIX files: a write is visible at once. To publish a file
@@ -110,26 +112,29 @@ 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
operation fail (`io`, or `no-entry`). An actor that opens its own `fs/`
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
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`,
`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 |
| --- | --- | --- |
| `spawn` | `actor`, `wasm`, `source`?, `built_by`? | test, then start a new actor |
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version |
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`?, `preserve_stopped`? (boolean, default `false`) | test, then swap the version; with `preserve_stopped: true`, a stopped actor is compiled and installed without execution and stays stopped |
| `rollback` | `actor`, `version` | go back to an older version |
| `stop`, `start` | `actor` | stop or start an actor |
| `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 |
| `tap`, `untap` | | receive (or not) a copy of every message the broker routes |
| `scan` | | boot new or changed `actor.wasm` files |
| `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) |
| `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 with exit code 75 (a wrapper like `nix run .#loop` starts it again, updated) |
**Headless.** `sausage run --headless [dir]` (or `sausage --headless`, or `SAUSAGE_HEADLESS=1`) turns the console input off: the host does not read stdin, so end of input (`/dev/null`) does not stop the world. `nix run .#loop -- --headless <world>` passes it through the loop. It runs until SIGTERM or SIGINT (stops every actor, exit code 0) or a broker `exit` request (exit code 75, as above). Without the flag, EOF on stdin or `/quit` ends the host.
Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`.
@@ -151,11 +156,37 @@ 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
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
`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
(except the stopped-preserving upgrade described below).
Only when its tests pass does the broker copy it to `versions/<n>/`
(`actor.wasm`, `source/`, `meta.json`) and switch to it. So `versions/` holds
exactly the versions that were installed, and instances always run from
exactly the versions that were installed (including stopped installations), and instances always run from
there. If the tests fail, nothing changes, and the sender gets an `error`
with the reason. All versions share the actor's `data/` and `public/`: a
new version must read the data of the old one.
@@ -184,16 +215,35 @@ instance. A version without test files passes.
top-level `actor.wasm` and `source/` a copy of the new version, and starts
it. A rollback does the same with the older version.
**Stopped-preserving upgrade.** Send `upgrade` with `"preserve_stopped": true`
to install a stopped actor's new program without starting it, running `main`
(even as a candidate), or reading or sending any of its mail. The host compiles
the WASM and copies the new version, source and metadata as usual, increments
the version, and keeps `actor.json`'s `running: false`. Its mailbox, `data/`
and `public/` stay untouched. Because candidate tests execute `main`, they are
not run in this case: the `upgraded` reply contains
`"tests": {"skipped": "preserve_stopped"}`. Test the build separately before
installing it this way. An explicit `start` later runs the installed version
and delivers the queued mail. No startup/crash probation runs for this stopped
installation. Running actors still use the normal tested swap, even with this
option. Omitting the option (or `false`) retains the existing behavior,
including starting a stopped actor. Explicit stop/start requests serialize
with upgrades of the same actor.
**Crashes.** The broker restarts a crashed actor after 0.5 s, then doubles
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
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
`actor/<id>/actor.wasm`. If the file differs from the installed version, it
is a new version, tested with the tests in `actor/<id>/source/`. If its
tests fail, the file stays where it is, the reason goes to `stderr.log`, and
the next scan tries again.
**The folder is the truth.** At start, the broker boots every
`actor/<id>/actor.wasm`. Afterwards, the file watcher reports a new or
changed `actor/<id>/actor.wasm` (or a new folder `actor/<id>/`) once it is
quiet for about 100 ms: a new file becomes a new actor, a changed one an
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
@@ -201,9 +251,19 @@ the next scan tries again.
{"type": "build", "source": "/actor/agent/public/drafts/notes"}
```
The reply is `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", "built_by": "cargo", ...}`
or `{"type": "build_failed", "errors": "..."}`. Pass `wasm`, `source`, and
`built_by` on to `spawn` or `upgrade`.
The build request may include `"skip_tests": true` to skip Rust unit tests.
By default, `cargo` first runs `cargo test` for the host target (not WASI),
so pure-logic unit tests in a `wasm32-wasip2` actor crate can run without a
WASI test runner. If tests fail or cannot compile, the build fails before the
WASI build. The reply carries `tests: {"passed": <count>, "failed": <count>,
"output": "<cargo test stdout and stderr>"}`; counts sum Cargo's test-result
summaries, and a test command/compilation failure without a summary counts as
one failure. With `skip_tests: true`, counts are zero and output says tests
were skipped.
The reply is `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", "built_by": "cargo", "tests": {...}, ...}`
or `{"type": "build_failed", "errors": "...", "tests": {...}}`. Pass
`wasm`, `source`, and `built_by` on to `spawn` or `upgrade`.
A Rust actor depends on the SDK with `sausage-actor = "0.1"`. `cargo` points
that dependency at the SDK of the host, which it publishes at
+38 -10
View File
@@ -10,28 +10,55 @@ change its own code.
## 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
nix run -- init # creates ./world and builds the agent (a few minutes)
nix run -- run # starts the world and a console
nix run git+https://gitea.surma.technology/surma/sausage
```
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`),
the base URL, the model, and the token. It reads `SAUSAGE_API`,
`SAUSAGE_BASE_URL`, `SAUSAGE_MODEL`, and `SAUSAGE_API_TOKEN` first. The
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:
```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
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`.
### Headless
`--headless` (or `SAUSAGE_HEADLESS=1`) turns the console input off: the host
never reads stdin, so end of input does not stop the world. It runs until
SIGTERM/SIGINT (clean stop, exit code 0) or a broker `exit` request (exit
code 75, so the loop starts the newest version). Create the world first
(`init`), then pass the flag through the loop:
```sh
nix run git+https://gitea.surma.technology/surma/sausage#loop -- --headless my-world
```
In the console:
@@ -51,14 +78,15 @@ actor.wasm the current version
actor.json installed version, running, subscriptions
source/ the source of the current version
public/ files that the actor publishes
data/ the actor's state
data/ actor state; owner-writable, world-readable via /actor/<id>/data/ (not private)
stderr.log what the actor and the broker write about it
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/`,
with the tests in `source/tests/`) into its folder. Then restart the world or
send `@broker {"type":"scan"}`. The broker tests the file before it uses it.
with the tests in `source/tests/`) into its folder; the host watches the
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.
## More
+84 -10
View File
@@ -14,11 +14,49 @@ pub struct Config {
pub base_url: String,
pub model: String,
pub token: String,
/// Optional reasoning effort for OpenAI models ("low", "medium", ...).
/// Optional reasoning effort for OpenAI models only (wins over `thinking` there).
#[serde(default)]
pub reasoning_effort: Option<String>,
#[serde(default)]
pub max_tokens: Option<u32>,
/// Thinking effort for both APIs: "low", "medium", "high", "xhigh", "max" (`LEVELS`), or "off"/unset for none.
/// Anthropic: adaptive thinking + output_config.effort; OpenAI: reasoning.effort. A level the model rejects is
/// replaced by the nearest accepted one (logged), never by "off".
#[serde(default)]
pub thinking: Option<String>,
}
/// Thinking levels, lowest first.
pub const LEVELS: [&str; 5] = ["low", "medium", "high", "xhigh", "max"];
/// The level nearest to `want` (lower on a tie) among those a 400 error lists in quotes (else all levels), not
/// yet `tried`. None: nothing left, so the request goes out without an effort field (model default).
fn fallback_effort(want: &str, error: &str, tried: &[String]) -> Option<String> {
let listed: Vec<&str> = error.split('\'').skip(1).step_by(2).filter(|w| LEVELS.contains(w)).collect();
let pool = if listed.iter().any(|l| *l != want) { listed } else { LEVELS.to_vec() };
let at = LEVELS.iter().position(|l| *l == want).unwrap_or(1) as i32;
pool.into_iter()
.filter(|l| *l != want && !tried.iter().any(|t| t == l))
.min_by_key(|l| {
let i = LEVELS.iter().position(|x| x == l).unwrap_or(1) as i32;
((i - at).abs(), i > at)
})
.map(str::to_string)
}
#[cfg(test)]
mod tests {
use super::fallback_effort;
#[test]
fn nearest_accepted_level() {
let anthropic = "HTTP 400: output_config.effort: Input should be 'low', 'medium', 'high' or 'max'";
assert_eq!(fallback_effort("xhigh", anthropic, &["xhigh".into()]).as_deref(), Some("high"));
let openai = "HTTP 400: Invalid value: 'ultra'. Supported values are: 'none', 'minimal', 'low', 'medium', 'high', 'xhigh', and 'max'.";
assert_eq!(fallback_effort("ultra", openai, &["ultra".into()]).as_deref(), Some("medium"));
// No list in the error: step to the neighbour, never to off.
assert_eq!(fallback_effort("max", "HTTP 400: effort max not supported", &["max".into()]).as_deref(), Some("xhigh"));
}
}
#[derive(Serialize, Deserialize, Clone, Debug)]
@@ -32,7 +70,7 @@ pub enum Entry {
},
ToolCall { id: String, name: String, arguments: String },
ToolResult { id: String, output: String },
/// A provider item that only this API shape understands (OpenAI reasoning).
/// A provider item that only this API shape understands (OpenAI reasoning, Anthropic thinking).
Opaque { api: String, item: Value },
}
@@ -43,10 +81,28 @@ pub struct Tool {
}
pub async fn complete(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result<Vec<Entry>, String> {
match cfg.api.as_str() {
"openai-responses" => openai(cfg, system, tools, history).await,
"anthropic-messages" => anthropic(cfg, system, tools, history).await,
other => Err(format!("unknown API shape {other:?}")),
let want = match cfg.api.as_str() {
"openai-responses" => cfg.reasoning_effort.clone().or_else(|| cfg.thinking.clone()),
_ => cfg.thinking.clone(),
}
.filter(|t| t != "off");
let mut effort = want.clone();
let mut tried = Vec::new();
loop {
let result = match cfg.api.as_str() {
"openai-responses" => openai(cfg, system, tools, history, effort.as_deref()).await,
"anthropic-messages" => anthropic(cfg, system, tools, history, want.is_some(), effort.as_deref()).await,
other => return Err(format!("unknown API shape {other:?}")),
};
match (result, effort.clone()) {
// The model rejected the effort level: retry with the nearest accepted one (thinking stays on).
(Err(e), Some(current)) if e.starts_with("HTTP 400") && e.contains("effort") => {
tried.push(current.clone());
effort = fallback_effort(want.as_deref().unwrap_or(""), &e, &tried);
eprintln!("agent: {} rejected thinking effort {current:?}; retrying with {effort:?}: {e}", cfg.model);
}
(result, _) => return result,
}
}
}
@@ -67,7 +123,7 @@ async fn post(url: &str, headers: &[(&str, String)], body: &Value) -> Result<Val
serde_json::from_str(&text).map_err(|e| format!("invalid JSON in response: {e}"))
}
async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result<Vec<Entry>, String> {
async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry], effort: Option<&str>) -> Result<Vec<Entry>, String> {
let mut input = Vec::new();
for e in history {
match e {
@@ -104,7 +160,7 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -
if let Some(max_tokens) = cfg.max_tokens {
body["max_output_tokens"] = json!(max_tokens);
}
if let Some(effort) = &cfg.reasoning_effort {
if let Some(effort) = effort {
body["reasoning"] = json!({"effort": effort});
}
let url = format!("{}/v1/responses", cfg.base_url);
@@ -143,7 +199,14 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -
Ok(out)
}
async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result<Vec<Entry>, String> {
async fn anthropic(
cfg: &Config,
system: &str,
tools: &[Tool],
history: &[Entry],
thinking: bool,
effort: Option<&str>,
) -> Result<Vec<Entry>, String> {
// Merge consecutive blocks of the same role into one message.
let mut messages: Vec<(String, Vec<Value>)> = Vec::new();
let mut push = |role: &str, block: Value| match messages.last_mut() {
@@ -163,6 +226,8 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]
Entry::ToolResult { id, output } => {
push("user", json!({"type": "tool_result", "tool_use_id": id, "content": output}))
}
// Thinking blocks go back unchanged (with their signature): required next to tool use.
Entry::Opaque { api, item } if api == "anthropic-messages" && thinking => push("assistant", item.clone()),
Entry::Opaque { .. } => {}
}
}
@@ -171,13 +236,19 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]
.iter()
.map(|t| json!({"name": t.name, "description": t.description, "input_schema": t.parameters}))
.collect();
let body = json!({
let mut body = json!({
"model": cfg.model,
"system": system,
"messages": messages,
"tools": tools,
"max_tokens": cfg.max_tokens.unwrap_or(32000),
});
if thinking {
body["thinking"] = json!({"type": "adaptive"});
if let Some(effort) = effort {
body["output_config"] = json!({"effort": effort});
}
}
let url = format!("{}/v1/messages", cfg.base_url);
let resp = post(
&url,
@@ -199,6 +270,9 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]
name: block["name"].as_str().unwrap_or_default().to_string(),
arguments: block["input"].to_string(),
}),
Some("thinking") | Some("redacted_thinking") => {
out.push(Entry::Opaque { api: "anthropic-messages".into(), item: block })
}
_ => {}
}
}
+2
View File
@@ -3,3 +3,5 @@ People talk to you through other actors. Your text goes to your subscribers.
Before writing an actor, read `/actor/broker/public/PROTOCOL.md` for messages, the broker, `cargo`, and tests.
Read the SDK at `/actor/cargo/public/sdk/` before you write an actor.
You can change your own source at `/actor/agent/source`.
The global filesystem is readable by every actor, including every actor's `/data` at `/actor/<id>/data/`. An actor can write only its own `/data` and `/public`. `/data` is not private: any unprotected secret file there is readable by every actor.
+1
View File
@@ -15,6 +15,7 @@ clap = { version = "4", features = ["derive"] }
http = "1"
http-body-util = "0.1"
include_dir = "0.7"
notify = "8"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1", features = ["full"] }
+147 -41
View File
@@ -1,11 +1,12 @@
//! The broker: message routing, mailboxes, subscriptions, taps, the version
//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
//! remove, subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs`
//! runs each WASM actor.
//! The broker: message routing, mailboxes, subscriptions, taps, watches, the
//! version store, and the requests to `broker` (spawn, upgrade, rollback,
//! stop, start, remove, subscribe, unsubscribe, tap, untap, watch, unwatch,
//! exit). `supervisor.rs` runs each WASM actor.
//!
//! The broker accepts only `.wasm` files. It does not know any source language.
//! The world folder is the source of truth: `scan` boots every
//! `actor/<id>/actor.wasm` and treats a changed file as an upgrade.
//! The world folder is the source of truth: at start, the broker boots every
//! `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::path::Path;
@@ -167,14 +168,19 @@ impl Broker {
/// they are never tapped themselves; a full mailbox drops the copy.
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 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();
for tapper in tappers {
if let Some(tx) = mailboxes.get(tapper) {
for id in to {
if let Some(tx) = mailboxes.get(id) {
let _ = tx.try_send(Envelope {
id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
from: "broker".into(),
to: tapper.clone(),
from: from.into(),
to: id.clone(),
reply_to: None,
body: body.clone(),
});
@@ -307,12 +313,16 @@ impl Broker {
Ok(component)
}
/// Test a new version where it lies. Only when the tests pass, copy it
/// to `versions/<n>/`: that folder holds only versions that run(ran).
/// Returns the component and the test report.
async fn try_version(&self, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<(Component, Value)> {
/// Compile a new version where it lies and, normally, test it. Copy it
/// to `versions/<n>/` only on success. A stopped-preserving install skips
/// execution, including candidate tests. Return the component and report.
async fn try_version(&self, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, run_tests: bool) -> Result<(Component, Value)> {
let component = self.load(id, wasm)?;
let tests = candidate::test(self, id, meta.version, &component, source).await;
let tests = if run_tests {
candidate::test(self, id, meta.version, &component, source).await
} else {
Ok(json!({"skipped": "preserve_stopped"}))
};
if let Err(e) = &tests {
self.log(id, &format!("version {} from {} failed its tests: {e:#}", meta.version, meta.wasm));
}
@@ -326,11 +336,11 @@ impl Broker {
}
/// Make `version` the current `actor.wasm`, `source/`, and `actor.json`.
pub fn install(&self, id: &str, version: u64) -> Result<()> {
pub fn install(&self, id: &str, version: u64, running: bool) -> Result<()> {
self.world.install_files(id, version)?;
self.update_meta(id, |m| {
m.version = Some(version);
m.running = true;
m.running = running;
})
}
@@ -353,26 +363,29 @@ impl Broker {
}
async fn create_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
let (_, tests) = self.try_version(id, meta, wasm, source).await?;
self.install(id, meta.version)?;
let (_, tests) = self.try_version(id, meta, wasm, source, true).await?;
self.install(id, meta.version, true)?;
self.start_supervisor(id, meta.version, true);
Ok(json!({"type": "spawned", "actor": id, "version": meta.version, "tests": tests}))
}
pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> Result<Value> {
pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta, preserve_stopped: bool) -> Result<Value> {
let lock = self.op_lock(id);
let _guard = lock.lock().await;
if !self.is_supervised(id) {
bail!("no WASM actor with ID {id}");
}
let version = self.world.next_version(id);
self.upgrade_version(id, &VersionMeta { version, ..meta }, wasm, source).await
self.upgrade_version(id, &VersionMeta { version, ..meta }, wasm, source, preserve_stopped).await
}
async fn upgrade_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
let (component, tests) = self.try_version(id, meta, wasm, source).await?;
async fn upgrade_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, preserve_stopped: bool) -> Result<Value> {
// Lifecycle requests hold the same op lock, so a stopped actor cannot
// start between this decision and the supervisor's installation.
let run_tests = !preserve_stopped || self.meta(id).is_some_and(|m| m.running);
let (component, tests) = self.try_version(id, meta, wasm, source, run_tests).await?;
let (version, requester) = (meta.version, meta.requested_by.clone());
let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, reply }).await?;
let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, preserve_stopped, reply }).await?;
result["tests"] = tests;
Ok(result)
}
@@ -406,31 +419,90 @@ impl Broker {
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.
/// - A new or changed `actor.wasm` is tested where it lies. A new actor
/// is installed, a changed one is an upgrade. If its tests fail,
/// nothing changes, and the file stays for the next scan.
/// nothing changes, and the file stays until it changes 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 {
if let Some(r) = self.boot_actor(&id).await {
report.push(r);
}
}
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> {
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());
self.metas.lock().unwrap().insert(id.to_string(), meta.clone());
if let Some(v) = installed
&& !self.is_supervised(id)
{
@@ -452,7 +524,7 @@ impl Broker {
let source = source.is_dir().then_some(source.as_path());
let result = match installed {
None => self.create_version(id, &vmeta, &wasm, source).await,
Some(_) => self.upgrade_version(id, &vmeta, &wasm, source).await,
Some(_) => self.upgrade_version(id, &vmeta, &wasm, source, false).await,
};
Some(result.unwrap_or_else(|e| json!({"type": "error", "actor": id, "reason": format!("{e:#}")})))
}
@@ -499,7 +571,11 @@ impl Broker {
}
"upgrade" => {
let (wasm, source, meta) = version_meta()?;
self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta).await
let preserve_stopped = match b.get("preserve_stopped") {
None => false,
Some(value) => value.as_bool().ok_or_else(|| anyhow!("preserve_stopped must be a boolean"))?,
};
self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta, preserve_stopped).await
}
"rollback" => {
let id = actor()?;
@@ -508,8 +584,15 @@ impl Broker {
let _guard = lock.lock().await;
self.control(&id, |reply| Control::Rollback { version, reply }).await
}
"stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await,
"start" => self.control(&actor()?, |reply| Control::Start { reply }).await,
"stop" | "start" => {
let id = actor()?;
let lock = self.op_lock(&id);
let _guard = lock.lock().await;
self.control(&id, |reply| match ty {
"stop" => Control::Stop { reply },
_ => Control::Start { reply },
}).await
},
"remove" => self.remove_actor(&actor()?).await,
"subscribe" | "unsubscribe" => {
let publisher = actor()?;
@@ -529,14 +612,32 @@ impl Broker {
let answer = if on { "tapping" } else { "untapped" };
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" => {
self.log("broker", &format!("exit requested by {}", env.from));
// notify_one keeps the permit if nobody waits yet.
self.exit.notify_one();
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 +664,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)
}
/// 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 {
if s.len() <= max {
return s.to_string();
+173 -47
View File
@@ -1,8 +1,9 @@
//! The native `cargo` actor: builds Rust source folders into `.wasm` files.
//!
//! Request: `{"type": "build", "source": "/actor/<id>/public/<folder>"}`
//! Replies: `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", ...}`
//! `{"type": "build_failed", "errors": "..."}`
//! optional `"skip_tests": true` skips native `cargo test` (tests run by default).
//! Replies: `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", "tests": {"passed": 1, "failed": 0, "output": "..."}, ...}`
//! `{"type": "build_failed", "errors": "...", "tests": {"passed": 0, "failed": 1, "output": "..."}}`
//!
//! The broker knows nothing about Rust. This actor is one way to get a
//! `.wasm` file. Other actors can add other languages.
@@ -32,11 +33,49 @@ pub const TARGET: &str = "wasm32-wasip2";
/// The source of the `sausage-actor` SDK.
static SDK: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk");
#[derive(Clone, Default, Debug)]
pub struct TestResults {
pub passed: u64,
pub failed: u64,
pub output: String,
}
impl TestResults {
fn value(&self) -> Value {
json!({"passed": self.passed, "failed": self.failed, "output": self.output})
}
}
pub struct BuildOutput {
/// Host path of the published `.wasm` file.
pub wasm: PathBuf,
pub seconds: f64,
pub warnings: String,
pub tests: TestResults,
}
#[derive(Debug)]
pub struct BuildFailure {
pub reason: String,
pub tests: TestResults,
}
impl std::fmt::Display for BuildFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if self.tests.failed > 0 && !self.tests.output.is_empty() {
write!(f, "{}:\n{}", self.reason, self.tests.output)
} else {
f.write_str(&self.reason)
}
}
}
impl std::error::Error for BuildFailure {}
impl From<anyhow::Error> for BuildFailure {
fn from(error: anyhow::Error) -> Self {
Self { reason: format!("{error:#}"), tests: TestResults::default() }
}
}
/// All builds share one target folder, so they run one at a time.
@@ -51,51 +90,84 @@ impl Cargo {
Ok(Self { world })
}
/// Build the crate in `source` (a host path) and publish the `.wasm` file
/// under `/actor/cargo/public/builds/<n>/`.
pub async fn build(&self, source: &Path) -> Result<BuildOutput> {
/// Run native `cargo test` unless opted out, then build and publish the
/// crate's `wasm32-wasip2` binary under `/actor/cargo/public/builds/<n>/`.
pub async fn build(&self, source: &Path, skip_tests: bool) -> std::result::Result<BuildOutput, BuildFailure> {
if !source.join("Cargo.toml").is_file() {
bail!("{} has no Cargo.toml", source.display());
return Err(BuildFailure {
reason: format!("{} has no Cargo.toml", source.display()),
tests: TestResults::default(),
});
}
let started = Instant::now();
let data = self.world.data_dir(ID);
let build_dir = data.join("work");
world::replace_source(source, &build_dir)?;
let manifest = build_dir.join("Cargo.toml");
std::fs::write(&manifest, prepare_manifest(&std::fs::read_to_string(&manifest)?, Some(&self.world.public_dir(ID).join("sdk")))?)?;
let original = std::fs::read_to_string(&manifest).context("read build manifest")?;
let prepared = prepare_manifest(&original, Some(&self.world.public_dir(ID).join("sdk")))?;
std::fs::write(&manifest, prepared).context("write prepared manifest")?;
let mut cmd = Command::new("cargo");
cmd.arg("build")
.arg("--release")
.arg("--target")
.arg(TARGET)
.arg("--message-format=json-render-diagnostics")
.arg("--target-dir")
.arg(data.join("target"))
.current_dir(&build_dir)
.kill_on_drop(true);
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!("{}", strip_progress(&stderr));
}
let mut artifact = None;
for line in String::from_utf8_lossy(&out.stdout).lines() {
if let Ok(v) = serde_json::from_str::<Value>(line)
&& v["reason"] == "compiler-artifact"
&& let Some(exe) = v["executable"].as_str()
{
artifact = Some(PathBuf::from(exe));
let target_dir = data.join("target");
let tests = if skip_tests {
TestResults { output: "skipped (skip_tests=true)".into(), ..TestResults::default() }
} else {
match run_tests(&build_dir, &target_dir).await {
Ok(tests) => tests,
Err(error) => {
let output = format!("cannot run cargo tests: {error:#}");
return Err(BuildFailure {
reason: output.clone(),
tests: TestResults { failed: 1, output, ..TestResults::default() },
});
}
}
};
if tests.failed > 0 {
return Err(BuildFailure { reason: "unit tests failed".into(), tests });
}
let artifact = artifact.ok_or_else(|| anyhow!("cargo built no binary. Does the crate have a src/main.rs?"))?;
let n = next_build_number(&self.world.public_dir(ID).join("builds"))?;
let dest_dir = self.world.public_dir(ID).join("builds").join(n.to_string());
let name = artifact.file_name().context("artifact has no file name")?;
let dest = dest_dir.join(name);
world::write_atomic(&dest, &std::fs::read(&artifact)?)?;
Ok(BuildOutput { wasm: dest, seconds: started.elapsed().as_secs_f64(), warnings: strip_progress(&stderr) })
let build_result: Result<BuildOutput> = async {
let mut cmd = Command::new("cargo");
cmd.arg("build")
.arg("--release")
.arg("--target")
.arg(TARGET)
.arg("--message-format=json-render-diagnostics")
.arg("--target-dir")
.arg(&target_dir)
.current_dir(&build_dir)
.kill_on_drop(true);
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!("{}", strip_progress(&stderr));
}
let mut artifact = None;
for line in String::from_utf8_lossy(&out.stdout).lines() {
if let Ok(v) = serde_json::from_str::<Value>(line)
&& v["reason"] == "compiler-artifact"
&& let Some(exe) = v["executable"].as_str()
{
artifact = Some(PathBuf::from(exe));
}
}
let artifact = artifact.ok_or_else(|| anyhow!("cargo built no binary. Does the crate have a src/main.rs?"))?;
let n = next_build_number(&self.world.public_dir(ID).join("builds"))?;
let dest_dir = self.world.public_dir(ID).join("builds").join(n.to_string());
let name = artifact.file_name().context("artifact has no file name")?;
let dest = dest_dir.join(name);
world::write_atomic(&dest, &std::fs::read(&artifact)?)?;
Ok(BuildOutput {
wasm: dest,
seconds: started.elapsed().as_secs_f64(),
warnings: strip_progress(&stderr),
tests: tests.clone(),
})
}
.await;
build_result.map_err(|error| BuildFailure { reason: format!("{error:#}"), tests })
}
/// The native `cargo` actor. Builds run one at a time, in order.
@@ -106,7 +178,7 @@ impl Cargo {
continue;
}
let body = self.handle(&broker, &env).await.unwrap_or_else(|e| {
json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}")})
json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}"), "tests": TestResults::default().value()})
});
broker.send(ID, Some(&env.from), None, Some(&env.id), body);
}
@@ -117,22 +189,76 @@ impl Cargo {
Some("build") => {
let source_s = env.body.get("source").and_then(Value::as_str).ok_or_else(|| anyhow!("missing \"source\""))?;
let source = broker.world.resolve(&env.from, source_s)?;
let out = self.build(&source).await?;
let wasm = self.world.global_path(&out.wasm).unwrap_or_else(|| out.wasm.display().to_string());
Ok(json!({
"type": "built",
"wasm": wasm,
"source": source_s,
"built_by": ID,
"seconds": (out.seconds * 10.0).round() / 10.0,
"warnings": out.warnings,
}))
let skip_tests = env.body.get("skip_tests").and_then(Value::as_bool).unwrap_or(false);
match self.build(&source, skip_tests).await {
Ok(out) => {
let wasm = self.world.global_path(&out.wasm).unwrap_or_else(|| out.wasm.display().to_string());
Ok(json!({
"type": "built",
"wasm": wasm,
"source": source_s,
"built_by": ID,
"seconds": (out.seconds * 10.0).round() / 10.0,
"warnings": out.warnings,
"tests": out.tests.value(),
}))
}
Err(failure) => Ok(json!({
"type": "build_failed",
"source": source_s,
"errors": failure.reason,
"tests": failure.tests.value(),
})),
}
}
other => bail!("unknown request {other:?}. Known: build"),
}
}
}
/// Run tests for the host triple: WASI crates' pure-logic tests can use the
/// normal Rust test harness without needing a WASI test runner.
async fn run_tests(build_dir: &Path, target_dir: &Path) -> Result<TestResults> {
let mut cmd = Command::new("cargo");
cmd.arg("test")
.arg("--target-dir")
.arg(target_dir)
.current_dir(build_dir)
.kill_on_drop(true);
let out = cmd.output().await.context("cannot run cargo test")?;
let stdout = String::from_utf8_lossy(&out.stdout).to_string();
let stderr = String::from_utf8_lossy(&out.stderr).to_string();
let combined = match (stdout.is_empty(), stderr.is_empty()) {
(true, true) => String::new(),
(false, true) => stdout,
(true, false) => stderr,
(false, false) => format!("{stdout}\n{stderr}"),
};
let output = strip_progress(&combined);
let passed = test_count(&output, "passed");
let mut failed = test_count(&output, "failed");
// Compilation failures and runner errors have no `test result` summary,
// but still make the default test phase fail.
if !out.status.success() && failed == 0 {
failed = 1;
}
Ok(TestResults { passed, failed, output })
}
/// Count the `passed` or `failed` totals in cargo's per-target test summaries.
fn test_count(output: &str, field: &str) -> u64 {
output
.lines()
.filter_map(|line| line.split_once("test result:").map(|(_, rest)| rest))
.flat_map(|summary| summary.split(';'))
.filter_map(|part| {
let words: Vec<_> = part.split_whitespace().collect();
let index = words.iter().position(|word| *word == field)?;
words.get(index.checked_sub(1)?)?.parse::<u64>().ok()
})
.sum()
}
/// 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> {
+2 -1
View File
@@ -3,7 +3,8 @@
//! - `@<actor> <json>` sends a JSON body. `@<actor> <text>` sends
//! `{"type": "text", "text": ...}`.
//! - A line without `@` goes to `agent` as text.
//! - `/help`, `/quit`. EOF on stdin also quits.
//! - `/help`, `/quit`. EOF on stdin also quits (not with `--headless`, which
//! never reads stdin).
use std::sync::Arc;
+15 -7
View File
@@ -1,7 +1,8 @@
//! Sausage: grow a system of WASI actors by talking to it.
//!
//! 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 candidate;
@@ -10,6 +11,7 @@ pub mod console;
pub mod runtime;
pub mod supervisor;
pub mod vfs;
pub mod watch;
pub mod world;
use std::path::Path;
@@ -30,12 +32,13 @@ pub struct Host {
pub world: World,
pub broker: Arc<Broker>,
native: Vec<tokio::task::JoinHandle<()>>,
watcher: Option<watch::Watcher>,
_lock: std::fs::File,
}
impl Host {
/// Start the broker, `cargo`, and every registered actor of the world at
/// `root`. Only one host can run a world at a time.
/// Start the broker, `cargo`, the file watcher, and every registered
/// actor of the world at `root`. Only one host can run a world at a time.
pub async fn start(root: &Path) -> Result<Host> {
let world = World::new(std::path::absolute(root)?);
if !world.exists() {
@@ -53,10 +56,14 @@ impl Host {
tokio::spawn(broker.clone().run_actor(broker.open_mailbox("broker"))),
tokio::spawn(cargo.run_actor(broker.clone(), broker.open_mailbox(cargo_actor::ID))),
];
// Watch first, so that no change during the boot pass is lost.
let watcher = watch::start(broker.clone())
.inspect_err(|e| broker.host_error(&format!("no file watching: {e}")))
.ok();
for change in broker.scan().await {
eprintln!("sausage: {change}");
}
Ok(Host { world, broker, native, _lock: lock })
Ok(Host { world, broker, native, watcher, _lock: lock })
}
/// Register a native actor with the ID `id` and return its mailbox.
@@ -64,9 +71,10 @@ impl Host {
self.broker.open_mailbox(id)
}
/// Stop the broker and `cargo` (with the requests they still run), then
/// every actor, gracefully.
pub async fn stop(self) {
/// Stop the file watcher, the broker and `cargo` (with the requests they
/// still run), then every actor, gracefully.
pub async fn stop(mut self) {
drop(self.watcher.take());
for task in &self.native {
task.abort();
}
+55 -9
View File
@@ -5,6 +5,7 @@ use anyhow::{Context, Result, bail};
use clap::{Parser, Subcommand};
use include_dir::{Dir, include_dir};
use serde_json::json;
use tokio::signal::unix::Signal;
use sausage::world::{ActorMeta, World};
use sausage::{Host, cargo_actor, console};
@@ -12,11 +13,18 @@ use sausage::{Host, cargo_actor, console};
/// The source of the agent actor. `init` builds it like any other actor.
static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
// Without a command, `sausage` runs the world in `./world` and creates it
// first (like `init`) if it does not exist yet.
#[derive(Parser)]
#[command(version, about)]
struct Cli {
/// Do not read stdin, so end of input does not stop the world. It runs
/// until SIGTERM (or SIGINT), or until an actor sends
/// `@broker {"type":"exit"}`. Also set by SAUSAGE_HEADLESS=1.
#[arg(long, global = true)]
headless: bool,
#[command(subcommand)]
command: Cmd,
command: Option<Cmd>,
}
#[derive(Subcommand)]
@@ -83,19 +91,47 @@ async fn init(dir: PathBuf) -> Result<()> {
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, false).await.context("build agent")?;
std::fs::copy(&out.wasm, world.wasm_path("agent"))?;
println!("Built in {:.1} s. Start the world with: sausage run {}", out.seconds, dir.display());
Ok(())
}
async fn run(dir: PathBuf) -> Result<()> {
/// Install the SIGTERM and SIGINT handlers now (not when first polled).
fn termination_signals() -> Result<(Signal, Signal)> {
use tokio::signal::unix::{SignalKind, signal};
let term = signal(SignalKind::terminate()).context("install SIGTERM handler")?;
let int = signal(SignalKind::interrupt()).context("install SIGINT handler")?;
Ok((term, int))
}
/// Resolves on SIGTERM or SIGINT.
async fn termination((mut term, mut int): (Signal, Signal)) {
tokio::select! {
_ = term.recv() => {}
_ = int.recv() => {}
}
}
async fn run(dir: PathBuf, headless: bool) -> Result<()> {
// Register the signal handlers before the host starts, so a signal that
// arrives early is never the default (kill) action.
let signals = if headless { Some(termination_signals()?) } else { None };
let host = Host::start(&dir).await?;
tokio::spawn(console::run_mailbox(host.mailbox(console::ID)));
let broker = host.broker.clone();
let exit = tokio::select! {
_ = console::run_input(host.broker.clone()) => false,
_ = broker.exit_requested() => true,
let exit = match signals {
Some(signals) => {
println!("Sausage headless: running until SIGTERM or an exit request.");
tokio::select! {
_ = termination(signals) => false,
_ = broker.exit_requested() => true,
}
}
None => tokio::select! {
_ = console::run_input(host.broker.clone()) => false,
_ = broker.exit_requested() => true,
},
};
println!("Stopping actors …");
host.stop().await;
@@ -111,8 +147,18 @@ async fn run(dir: PathBuf) -> Result<()> {
#[tokio::main]
async fn main() -> Result<()> {
match Cli::parse().command {
Cmd::Init { dir } => init(dir).await,
Cmd::Run { dir } => run(dir).await,
let cli = Cli::parse();
let headless = cli.headless || std::env::var("SAUSAGE_HEADLESS").is_ok_and(|v| !v.is_empty() && v != "0");
match cli.command {
Some(Cmd::Init { dir }) => init(dir).await,
Some(Cmd::Run { dir }) => run(dir, headless).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, headless).await
}
}
}
+14 -9
View File
@@ -23,7 +23,7 @@ pub type Reply = oneshot::Sender<Result<Value>>;
pub enum Control {
/// Swap to `version` (already in `versions/`, compiled as `component`).
Upgrade { version: u64, component: Component, requester: String, reply: Reply },
Upgrade { version: u64, component: Component, requester: String, preserve_stopped: bool, reply: Reply },
Rollback { version: u64, reply: Reply },
Stop { reply: Reply },
Start { reply: Reply },
@@ -221,24 +221,29 @@ impl Supervisor {
/// Make `version` current. It is compiled first (unless `component` is
/// given): if that fails, nothing changes. `data/` and `public/` stay as
/// they are.
async fn install(&mut self, version: u64, component: Option<Component>) -> Result<()> {
async fn install(&mut self, version: u64, component: Option<Component>, running: bool) -> Result<()> {
let component = match component {
Some(c) => c,
None => self.broker.load_version(&self.id, version)?,
};
self.stop().await;
self.broker.install(&self.id, version)?;
self.broker.install(&self.id, version, running)?;
self.version = version;
self.component = Some(component);
self.running = true;
self.running = running;
self.restarts = 0;
self.restart_at = None;
Ok(())
}
async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result<Value> {
async fn upgrade(&mut self, version: u64, component: Component, requester: String, preserve_stopped: bool) -> Result<Value> {
let previous = self.version;
self.install(version, Some(component)).await?;
let running = self.running || !preserve_stopped;
self.install(version, Some(component), running).await?;
if !running {
self.probation = None;
return Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous}));
}
self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester });
if let Err(e) = self.start().await {
self.rollback(previous).await?;
@@ -249,7 +254,7 @@ impl Supervisor {
async fn rollback(&mut self, version: u64) -> Result<Value> {
let from = self.version;
self.install(version, None).await?;
self.install(version, None, true).await?;
self.probation = None;
self.start().await?;
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version}))
@@ -258,8 +263,8 @@ impl Supervisor {
/// Handle a control request. Returns `true` when the supervisor ends.
async fn handle(&mut self, ctl: Control) -> bool {
match ctl {
Control::Upgrade { version, component, requester, reply } => {
let _ = reply.send(self.upgrade(version, component, requester).await);
Control::Upgrade { version, component, requester, preserve_stopped, reply } => {
let _ = reply.send(self.upgrade(version, component, requester, preserve_stopped).await);
}
Control::Rollback { version, reply } => {
let _ = reply.send(self.rollback(version).await);
+7 -6
View File
@@ -328,14 +328,15 @@ impl VfsView<'_> {
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, stream.clone()).await? {
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)?;
HostDirectoryEntryStream::drop(&mut self.inner, stream).map_err(|_| ErrorCode::Io)?;
Ok(entries)
}
@@ -587,10 +588,10 @@ impl HostDescriptor for VfsView<'_> {
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
&& HostDescriptor::get_type(&mut self.inner, opened.clone()).await? == DescriptorType::Directory
{
self.state.dirs.insert(opened.rep(), path);
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)
}
+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:
//! `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
//! <world>/
//! actor/<id>/
//! 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
//! public/ written by the actor, read-only for others
//! 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`).
#[serde(default)]
pub tapped: bool,
/// Namespace paths whose changes this actor gets as `fs_changed` (`watch`).
#[serde(default)]
pub watches: BTreeSet<String>,
}
fn yes() -> bool {
+4
View File
@@ -38,6 +38,10 @@ fn main() {
panic!("version 2 crashes on startup");
}
if actor == "stopped-upgrade" {
std::fs::write("/data/main-version", &version).unwrap();
}
let mut open = HashMap::<String, std::fs::File>::new();
// Files that this actor serves at /actor/<id>/fs/ (path -> text).
let mut served = std::collections::BTreeMap::<String, String>::new();
+85
View File
@@ -0,0 +1,85 @@
//! 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());
}
/// Start a headless host on a fresh world with stdin at EOF; returns once it
/// printed its ready line (printed after the signal handlers are installed).
fn start_headless(label: &str) -> (std::process::Child, std::io::Lines<std::io::BufReader<std::process::ChildStdout>>) {
use std::io::{BufRead, BufReader};
let dir = empty_dir(label);
let world = dir.join("world");
sausage::World::create(&world).unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_sausage"))
.args(["run", "--headless"])
.arg(&world)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.unwrap();
let mut lines = BufReader::new(child.stdout.take().unwrap()).lines();
let line = lines.next().expect("host exited before it started").unwrap();
assert!(line.contains("headless"), "unexpected first line: {line}");
(child, lines)
}
/// With stdin at EOF (as soon as the host starts), a headless host stays up.
#[test]
fn headless_ignores_stdin_eof() {
let (mut child, _lines) = start_headless("headless-eof");
std::thread::sleep(std::time::Duration::from_millis(500));
let alive = child.try_wait().unwrap().is_none();
let _ = child.kill();
let _ = child.wait();
assert!(alive, "headless host exited on stdin EOF");
}
/// SIGTERM right after the ready line stops the actors and exits with 0.
#[test]
fn headless_stops_cleanly_on_sigterm_right_after_ready() {
let (mut child, lines) = start_headless("headless-term");
let pid = child.id().to_string();
assert!(Command::new("kill").args(["-TERM", &pid]).status().unwrap().success());
let rest: Vec<String> = lines.map(|l| l.unwrap()).collect();
let status = child.wait().unwrap();
assert!(status.success(), "exit status {status:?}, output {rest:?}");
assert!(rest.iter().any(|l| l.contains("Stopping actors")), "output {rest:?}");
}
@@ -0,0 +1,6 @@
[package]
name = "cargo-test-fail"
version = "0.1.0"
edition = "2024"
[workspace]
@@ -0,0 +1,9 @@
fn main() {}
#[cfg(test)]
mod tests {
#[test]
fn deliberate_unit_test_failure() {
assert_eq!(2 + 2, 5);
}
}
+8
View File
@@ -1 +1,9 @@
fn main() {}
#[cfg(test)]
mod tests {
#[test]
fn unit_tests_run_on_the_host_target() {
assert_eq!(2 + 2, 4);
}
}
+296 -15
View File
@@ -108,15 +108,15 @@ async fn routes_replies_publishes_and_reports_undeliverable() {
async fn broker_does_not_answer_messages_with_reply_to() {
let world = create_world("broker-reply-to");
let (host, mut mailbox) = start(&world).await;
let scanned = request(&host, &mut mailbox, json!({"type": "scan"})).await;
assert_eq!(scanned.body["type"], "scanned");
let untapped = request(&host, &mut mailbox, json!({"type": "untap"})).await;
assert_eq!(untapped.body["type"], "untapped");
host.broker.send(
"console",
Some("broker"),
None,
Some("upstream-message"),
json!({"type": "scan"}),
json!({"type": "untap"}),
);
assert!(
tokio::time::timeout(Duration::from_secs(1), mailbox.recv())
@@ -436,6 +436,113 @@ async fn upgrades_test_candidates_keep_old_versions_and_queue_messages_in_order(
host.stop().await;
}
#[tokio::test]
async fn stopped_preserving_upgrade_installs_without_execution_and_keeps_all_mail() {
let world = create_world("stopped-upgrade");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let source = passing_source(&world, "initial");
spawn_type(&host, &mut mailbox, "stopped-upgrade", &wasm, &source).await;
ask_type(&host, &mut mailbox, "stopped-upgrade", json!({"type": "version"}), "version").await;
let stopped = request(&host, &mut mailbox, json!({"type": "stop", "actor": "stopped-upgrade"})).await;
assert_eq!(stopped.body["type"], "stopped");
// A distinct valid component: append an empty custom section named x.
let mut bytes = std::fs::read(&wasm).unwrap();
bytes.extend_from_slice(&[0, 2, 1, b'x']);
let new_wasm = world.public_dir("console").join("new.wasm");
std::fs::write(&new_wasm, &bytes).unwrap();
// This would fail if candidate main/tests were executed.
let unrun = source_with_test(&world, "unrun", json!({"steps": [{
"send": {"type": "echo"}, "expect": {"type": "must-not-run"}
}]}));
std::fs::write(unrun.join("new-source"), "version two").unwrap();
std::fs::write(world.public_dir("stopped-upgrade").join("keep"), "public state").unwrap();
let mut ids = Vec::new();
for n in 0..sausage::broker::MAILBOX_SIZE {
ids.push(host.broker.send("console", Some("stopped-upgrade"), None, None, json!({"type": "ordered", "n": n})));
}
let upgrade_id = host.broker.send("console", Some("broker"), None, None, json!({
"type": "upgrade", "actor": "stopped-upgrade", "wasm": new_wasm.display().to_string(),
"source": unrun.display().to_string(), "preserve_stopped": true
}));
// Do not silently discard unexpected actor mail while awaiting the reply.
let upgraded = receive_message(&mut mailbox, Duration::from_secs(10)).await;
assert_eq!(upgraded.reply_to.as_deref(), Some(upgrade_id.as_str()));
assert_eq!(upgraded.body["type"], "upgraded");
assert_eq!(upgraded.body["version"], 2);
assert_eq!(upgraded.body["previous"], 1);
assert_eq!(upgraded.body["tests"], json!({"skipped": "preserve_stopped"}));
let meta = world.read_meta("stopped-upgrade");
assert_eq!(meta.version, Some(2));
assert!(!meta.running);
assert_eq!(std::fs::read(world.wasm_path("stopped-upgrade")).unwrap(), bytes);
assert_eq!(std::fs::read(world.version_dir("stopped-upgrade", 2).join("actor.wasm")).unwrap(), bytes);
assert_ne!(std::fs::read(world.version_dir("stopped-upgrade", 1).join("actor.wasm")).unwrap(), bytes);
assert_eq!(std::fs::read_to_string(world.source_dir("stopped-upgrade").join("new-source")).unwrap(), "version two");
assert_eq!(std::fs::read_to_string(world.data_dir("stopped-upgrade").join("main-version")).unwrap(), "1");
assert_eq!(std::fs::read_to_string(world.public_dir("stopped-upgrade").join("keep")).unwrap(), "public state");
assert!(mailbox.try_recv().is_err());
// The full mailbox is still full: nothing was consumed or added by the upgrade.
let overflow_id = host.broker.send("console", Some("stopped-upgrade"), None, None, json!({"type": "echo"}));
let overflow = receive_message(&mut mailbox, Duration::from_secs(10)).await;
assert_eq!(overflow.reply_to.as_deref(), Some(overflow_id.as_str()));
assert_eq!(overflow.body["type"], "undeliverable");
assert!(overflow.body["reason"].as_str().unwrap().contains("full"));
// Only an explicit start executes v2. Every original envelope survives in order.
let start_id = host.broker.send("console", Some("broker"), None, None, json!({"type": "start", "actor": "stopped-upgrade"}));
let mut started = false;
let mut count = 0;
tokio::time::timeout(Duration::from_secs(10), async {
while !started || count < ids.len() {
let reply = mailbox.recv().await.unwrap();
if reply.reply_to.as_deref() == Some(start_id.as_str()) {
assert_eq!(reply.body["type"], "started");
started = true;
} else {
assert_eq!(reply.from, "stopped-upgrade");
assert_eq!(reply.reply_to.as_deref(), Some(ids[count].as_str()));
assert_eq!(reply.body, json!({"type": "ordered", "n": count, "version": "2"}));
count += 1;
}
}
}).await.expect("queued mail was not delivered after start");
assert_eq!(std::fs::read_to_string(world.data_dir("stopped-upgrade").join("main-version")).unwrap(), "2");
host.stop().await;
}
#[tokio::test]
async fn stopped_preserving_option_keeps_running_upgrades_tested_and_default_upgrades_start() {
let world = create_world("preserve-running");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let passing = passing_source(&world, "passing");
let failing = source_with_test(&world, "failing", json!({"steps": [{
"send": {"type": "echo"}, "expect": {"type": "wrong"}
}]}));
spawn_type(&host, &mut mailbox, "running", &wasm, &passing).await;
for (option, source, expected) in [(json!("true"), &passing, "error"), (json!(true), &failing, "error"), (json!(true), &passing, "upgraded")] {
let reply = request(&host, &mut mailbox, json!({
"type": "upgrade", "actor": "running", "wasm": wasm.display().to_string(),
"source": source.display().to_string(), "preserve_stopped": option
})).await;
assert_eq!(reply.body["type"], expected);
}
assert!(world.read_meta("running").running);
let version = ask_type(&host, &mut mailbox, "running", json!({"type": "version"}), "version").await;
assert_eq!(version.body["version"], "2");
request(&host, &mut mailbox, json!({"type": "stop", "actor": "running"})).await;
let default = upgrade_actor(&host, &mut mailbox, "running", &wasm, &passing).await;
assert_eq!(default.body["type"], "upgraded");
assert!(world.read_meta("running").running);
let version = ask_type(&host, &mut mailbox, "running", json!({"type": "version"}), "version").await;
assert_eq!(version.body["version"], "3");
host.stop().await;
}
#[tokio::test]
async fn rolls_back_a_crash_loop_to_the_previous_version_without_restoring_data() {
let world = create_world("crash-loop");
@@ -611,14 +718,18 @@ async fn unknown_broker_request_lists_the_known_ones() {
let unknown = request(&host, &mut mailbox, json!({"type": "frobnicate"})).await;
assert_eq!(unknown.body["type"], "error");
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;
}
#[tokio::test]
async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
async fn cargo_actor_runs_unit_tests_by_default_and_can_skip_them() {
let world = create_world("cargo-actor");
let (host, mut mailbox) = start(&world).await;
let valid_source = cargo_source("cargo-valid");
let valid_id = host.broker.send(
"console",
@@ -627,12 +738,42 @@ async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
None,
json!({"type": "build", "source": valid_source.display().to_string()}),
);
let built = receive_reply(&mut mailbox, &valid_id, Duration::from_secs(120)).await;
assert_eq!(built.body["type"], "built");
let built = receive_reply(&mut mailbox, &valid_id, Duration::from_secs(180)).await;
assert_eq!(built.body["type"], "built", "{built:?}");
assert_eq!(built.body["tests"]["passed"], 1);
assert_eq!(built.body["tests"]["failed"], 0);
assert!(built.body["tests"]["output"].as_str().unwrap().contains("unit_tests_run_on_the_host_target"));
let wasm = built.body["wasm"].as_str().unwrap();
assert!(wasm.ends_with(".wasm"));
assert!(host.world.resolve("console", wasm).unwrap().is_file());
let failing_test_source = cargo_source("cargo-test-fail");
let failed_test_id = host.broker.send(
"console",
Some("cargo"),
None,
None,
json!({"type": "build", "source": failing_test_source.display().to_string()}),
);
let failed_test = receive_reply(&mut mailbox, &failed_test_id, Duration::from_secs(120)).await;
assert_eq!(failed_test.body["type"], "build_failed", "{failed_test:?}");
assert_eq!(failed_test.body["tests"]["passed"], 0);
assert_eq!(failed_test.body["tests"]["failed"], 1);
assert!(failed_test.body["tests"]["output"].as_str().unwrap().contains("deliberate_unit_test_failure"));
let skipped_id = host.broker.send(
"console",
Some("cargo"),
None,
None,
json!({"type": "build", "source": failing_test_source.display().to_string(), "skip_tests": true}),
);
let skipped = receive_reply(&mut mailbox, &skipped_id, Duration::from_secs(120)).await;
assert_eq!(skipped.body["type"], "built", "{skipped:?}");
assert_eq!(skipped.body["tests"]["passed"], 0);
assert_eq!(skipped.body["tests"]["failed"], 0);
assert_eq!(skipped.body["tests"]["output"], "skipped (skip_tests=true)");
let invalid_source = cargo_source("cargo-fail");
let invalid_id = host.broker.send(
"console",
@@ -641,14 +782,10 @@ async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
None,
json!({"type": "build", "source": invalid_source.display().to_string()}),
);
let failed = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await;
assert_eq!(failed.body["type"], "build_failed");
assert!(
failed.body["errors"]
.as_str()
.unwrap()
.contains("missing_function")
);
let invalid = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await;
assert_eq!(invalid.body["type"], "build_failed", "{invalid:?}");
assert_eq!(invalid.body["tests"]["failed"], 1);
assert!(invalid.body["tests"]["output"].as_str().unwrap().contains("missing_function"));
host.stop().await;
}
@@ -974,3 +1111,147 @@ async fn a_full_tapper_mailbox_drops_copies_without_blocking() {
assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped");
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": {
"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": {
"inputs": {
"systems": "systems"
@@ -36,6 +51,7 @@
},
"root": {
"inputs": {
"crane": "crane",
"flake-utils": "flake-utils",
"nixpkgs": "nixpkgs",
"rust-overlay": "rust-overlay"
+108 -22
View File
@@ -6,9 +6,10 @@
flake-utils.url = "github:numtide/flake-utils";
rust-overlay.url = "github:oxalica/rust-overlay";
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:
let
pkgs = import nixpkgs {
@@ -20,34 +21,53 @@
targets = [ "wasm32-wasip2" ];
extensions = [ "rust-src" ];
};
rustPlatform = pkgs.makeRustPlatform { cargo = rust; rustc = rust; };
sausage = rustPlatform.buildRustPackage {
# The runtime cargo actor needs Cargo, rustc (with wasm32-wasip2), and
# the native linker, even when the host itself is started with a bare
# PATH. Keep this same runtime path in the packaged host and exercise
# the installed binary below with PATH deliberately empty.
runtimePath = pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ];
craneLib = (crane.mkLib pkgs).overrideToolchain rust;
common = {
pname = "sausage";
version = "0.1.0";
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;
doNotRemoveReferencesToRustToolchain = true;
cargoExtraArgs = "--locked -p sausage";
doCheck = false;
nativeBuildInputs = [ pkgs.makeWrapper ];
# The `cargo` actor runs cargo at runtime. Build scripts need a C linker.
postInstall = ''
wrapProgram $out/bin/sausage --prefix PATH : ${pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ]}
wrapProgram $out/bin/sausage --prefix PATH : ${runtimePath}
'';
meta.mainProgram = "sausage";
};
});
# `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,
# so the world updates itself. `/quit` (exit code 0) ends the loop.
# `--internal` (or `.#loop-internal`) uses Gitea's SSH port on the LAN.
publicUrl = "git+ssh://containeruser@gitea.surma.technology:2222/surma/sausage";
internalUrl = "git+ssh://containeruser@10.0.0.2:2222/surma/sausage";
loop = name: url: pkgs.writeShellScriptBin name ''
url="${url}"
if [ "''${1:-}" = "--internal" ]; then url="${internalUrl}"; shift; fi
if [ $# -eq 0 ]; then set -- world; fi
# SAUSAGE_FLAKE overrides where it fetches from (any flake URL,
# e.g. a fork or a local checkout).
repoUrl = "git+https://gitea.surma.technology/surma/sausage";
loop = pkgs.writeShellScriptBin "sausage-loop" ''
url="''${SAUSAGE_FLAKE:-${repoUrl}}"
# No folder: bare `sausage` (creates ./world first if needed).
if [ $# -gt 0 ]; then set -- run "$@"; fi
while true; do
nix run --refresh "$url#default" -- run "$@"
nix run --refresh "$url#default" -- "$@"
code=$?
if [ "$code" -eq 0 ]; then exit 0; fi
echo "sausage-loop: sausage exited with $code, starting the newest version in 2 s …" >&2
@@ -61,19 +81,85 @@
# `nix flake check` runs every test of the host, the SDK, and the
# agent. The tests build WASM actors with cargo, offline, from the
# vendored crates of Cargo.lock.
checks.tests = sausage.overrideAttrs (old: {
pname = "sausage-tests";
doCheck = true;
cargoTestFlags = [ "-p" "sausage" "-p" "sausage-actor" "-p" "agent" ];
# The binary the host loop runs (`nix run`), built from the same
# cached dependencies as the tests.
checks.package = sausage;
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 = ''
# Its own lock file pins crates that the vendored set may lack.
rm -f cli/tests/actors/toolbox/Cargo.lock
'';
});
# Exercise the installed host exactly as the loop launches it: the
# caller has no Cargo on PATH, so only the packaged runtime toolchain
# can run the native tests and the wasm32-wasip2 build.
checks.runtime-toolchain = pkgs.runCommand "sausage-runtime-toolchain" { } ''
set -eu
root="$TMPDIR/runtime-toolchain"
world="$root/world"
emptyPath="$root/empty-bin"
mkdir -p "$emptyPath" "$root/home" "$root/cargo-home"
for actor in broker console cargo fs; do
mkdir -p "$world/actor/$actor/public" "$world/actor/$actor/data"
done
source="$world/actor/console/public/probe"
mkdir -p "$source/src" "$source/tests"
cat > "$source/Cargo.toml" <<'TOML'
[package]
name = "runtime-probe"
version = "0.1.0"
edition = "2024"
TOML
printf 'fn main() {}\n' > "$source/src/main.rs"
cat > "$source/tests/probe.rs" <<'RUST'
#[test]
fn native_test_harness_runs() {
assert_eq!(2 + 2, 4);
}
RUST
echo "runtime-check: run packaged host with an empty PATH"
coproc host {
${pkgs.coreutils}/bin/env -i PATH="$emptyPath" HOME="$root/home" TMPDIR="$root" \
CARGO_HOME="$root/cargo-home" ${sausage}/bin/sausage run "$world"
}
hostInput=''${host[1]}
hostOutput=''${host[0]}
hostPid=$host_PID
trap 'kill "$hostPid" 2>/dev/null || true' EXIT
printf '%s\n' '@cargo {"type":"build","source":"/actor/console/public/probe"}' >&"$hostInput"
built=0
while IFS= read -r -t 300 line <&"$hostOutput"; do
echo "$line"
case "$line" in
*'"type":"build_failed"'*)
echo "packaged cargo actor build failed with an empty caller PATH" >&2
exit 1
;;
*'"type":"built"'*)
case "$line" in
*'"passed":1'*) built=1 ;;
*) echo "native test did not pass" >&2; exit 1 ;;
esac
break
;;
esac
done
if [ "$built" -ne 1 ]; then
echo "packaged host did not build the probe" >&2
exit 1
fi
printf '%s\n' '/quit' >&"$hostInput"
wait "$hostPid"
mkdir -p "$out"
touch "$out/passed"
'';
apps.default = { type = "app"; program = "${sausage}/bin/sausage"; };
apps.loop = { type = "app"; program = "${loop "sausage-loop" publicUrl}/bin/sausage-loop"; };
apps.loop-internal = { type = "app"; program = "${loop "sausage-loop-internal" internalUrl}/bin/sausage-loop-internal"; };
apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };
devShells.default = pkgs.mkShell {
packages = [ rust pkgs.wasmtime pkgs.wasm-tools ];