Watch the world folder: auto spawn/upgrade, broker watch/unwatch, retire scan

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).

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/<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(),
                 });
@@ -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<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)
         {
@@ -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();