diff --git a/Cargo.lock b/Cargo.lock index 58e4cee..ca4874c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/PROTOCOL.md b/PROTOCOL.md index 940dec4..ddf9cb7 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -68,7 +68,7 @@ Each instance sees three folders: | `/actor` | read-only | the `actor/` folder of the world | `/actor//` 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//`. `ls /actor` lists all actors. Files are plain POSIX files: a write is visible at once. To publish a file @@ -110,14 +110,15 @@ Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`. - No reply within 10 seconds, or an actor that does not exist, makes the 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 | | --- | --- | --- | @@ -128,7 +129,7 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`, | `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor//` (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 | +| `watch`, `unwatch` | `path` | receive (or not) `fs_changed` for every change of `path` or anything under it; answer `{"type": "watching" \| "unwatched", "path"}` with the path as an `/actor/...` path | | `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) | Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`. @@ -151,6 +152,31 @@ including its own), for debugging, logging, or a live view of the world: - Like subscriptions, a tap is saved in `actor.json` (`"tapped": true`) and 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//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. Only when its tests pass does the broker copy it to `versions//` @@ -189,11 +215,15 @@ the delay up to 30 s. A new version that crashes 3 times within 30 seconds after an upgrade goes back to the previous version. The actor that asked for 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//actor.wasm`. If the file differs from the installed version, it -is a new version, tested with the tests in `actor//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//actor.wasm`. Afterwards, the file watcher reports a new or +changed `actor//actor.wasm` (or a new folder `actor//`) 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//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 diff --git a/README.md b/README.md index c3eeb6c..678c710 100644 --- a/README.md +++ b/README.md @@ -72,8 +72,9 @@ versions// 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 diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 29943cf..033c2ee 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -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"] } diff --git a/cli/src/broker.rs b/cli/src/broker.rs index b8a343e..b8a5095 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -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//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//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(), }); @@ -406,31 +412,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) -> Vec { let mut report = Vec::new(); for id in self.world.actor_ids().unwrap_or_default() { - if NATIVE.contains(&id.as_str()) || !self.world.wasm_path(&id).is_file() { - continue; - } - let lock = self.op_lock(&id); - let _guard = lock.lock().await; - if let Some(r) = self.scan_actor(&id).await { + if let Some(r) = self.boot_actor(&id).await { report.push(r); } } report } + /// The file watcher saw `actor//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, 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) -> Vec { + self.host_error("the file watcher dropped events; watchers get a rescan, actors a full pass"); + let watchers: Vec = + 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 = { + 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, id: &str) -> Option { + 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, id: &str) -> Option { - 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) { @@ -529,14 +594,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 +646,11 @@ fn same_file(a: &Path, b: &Path) -> bool { matches!((std::fs::read(a), std::fs::read(b)), (Ok(x), Ok(y)) if x == y) } +/// 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(); diff --git a/cli/src/lib.rs b/cli/src/lib.rs index 3ca15ab..042f4c8 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -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, native: Vec>, + watcher: Option, _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 { 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(); } diff --git a/cli/src/watch.rs b/cli/src/watch.rs new file mode 100644 index 0000000..e8ec96f --- /dev/null +++ b/cli/src/watch.rs @@ -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//actor.wasm` (or folder +//! `actor/`) 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) -> notify::Result { + 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, + first: Instant, + last: Instant, +} + +impl Pending { + fn due(&self) -> Instant { + (self.last + QUIET).min(self.first + MAX_DELAY) + } +} + +async fn run(broker: Arc, mut events: mpsc::UnboundedReceiver>) { + let root = broker.world.actors_root(); + let mut pending: HashMap = 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 = 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) { + 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 { + 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); + +/// 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: ¬ify::Event) -> Vec { + let path = |i: usize| event.paths.get(i).and_then(|p| namespace(root, p)); + let one = |op: &'static str| -> Vec { 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, path: String, op: &'static str, to: Option, 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//actor.wasm` or the folder `actor/` +/// appeared or changed. +fn new_wasm(path: &str, op: &str, to: Option<&str>) -> Option { + 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)> = + 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())), + ] + ); + } +} diff --git a/cli/src/world.rs b/cli/src/world.rs index 7d16a4b..7164256 100644 --- a/cli/src/world.rs +++ b/cli/src/world.rs @@ -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 //! / //! actor// //! 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, } fn yes() -> bool { diff --git a/cli/tests/host.rs b/cli/tests/host.rs index 526f5ce..296b92f 100644 --- a/cli/tests/host.rs +++ b/cli/tests/host.rs @@ -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()) @@ -611,7 +611,10 @@ 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; } @@ -974,3 +977,156 @@ 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, + 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")) +} + +/// Every message that arrives until the mailbox is quiet for `quiet`. +async fn drain(mailbox: &mut mpsc::Receiver, quiet: Duration) -> Vec { + let mut messages = Vec::new(); + while let Ok(Some(message)) = tokio::time::timeout(quiet, mailbox.recv()).await { + messages.push(message); + } + messages +} + +/// The `fs_changed` bodies among `messages` (all from `fs`). +fn fs_events(messages: Vec) -> Vec { + 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::(&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::::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::::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; +}