diff --git a/PROTOCOL.md b/PROTOCOL.md index 1fd6f83..41faba1 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), `source/`, `public/`, `data/`, +(installed version, running, subscriptions, tapped), `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,7 +110,7 @@ Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`. 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`, `subscribed`, `unsubscribed`, -`scanned`, `exiting`), or `{"type": "error", "reason": "..."}`. +`tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`. | `type` | Fields | Effect | | --- | --- | --- | @@ -119,11 +119,30 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`, | `rollback` | `actor`, `version` | go back to an older version | | `stop`, `start` | `actor` | stop or start an actor | | `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) | Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`. +**Taps.** After `tap`, the sender gets a copy of every message that the +broker routes (direct messages, replies, and publishes, between any actors, +including its own), for debugging, logging, or a live view of the world: + +```json +{"type": "tapped", "message": {"id": "m-17", "from": "console", "to": "agent", "reply_to": null, "body": {...}}, "ts": 1792132579067} +``` + +- `to` is `null` for a publish: one copy per publish, not per subscriber. +- `ts` is the time of routing, in milliseconds since the Unix epoch. +- Copies come from `broker`. They are not routed themselves, so they are + never tapped, and each message is tapped once per tapper. +- A copy that does not fit into the tapper's mailbox is dropped; routing + never waits for a tapper, and nobody gets `undeliverable` for a copy. +- Never answer a copy: the answer would be tapped again, forever. +- Like subscriptions, a tap is saved in `actor.json` (`"tapped": true`) and + survives restarts and upgrades. `untap` 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//` diff --git a/cli/src/broker.rs b/cli/src/broker.rs index c90f6e1..28a13c5 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -1,6 +1,7 @@ -//! The broker: message routing, mailboxes, subscriptions, the version store, -//! and the requests to `broker` (spawn, upgrade, rollback, stop, start, -//! subscribe, unsubscribe, scan, exit). `supervisor.rs` runs each WASM actor. +//! The broker: message routing, mailboxes, subscriptions, taps, the version +//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start, +//! subscribe, unsubscribe, tap, untap, scan, 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 @@ -133,17 +134,23 @@ impl Broker { Some(own) => format!("{from}/{own}"), None => format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)), }; - let recipients: Vec = match to { - Some(to) => vec![to.to_string()], - None => self - .metas - .lock() - .unwrap() - .iter() - .filter(|(_, m)| m.subscriptions.contains(from)) - .map(|(id, _)| id.clone()) - .collect(), + let (recipients, tappers) = { + let metas = self.metas.lock().unwrap(); + let recipients: Vec = match to { + Some(to) => vec![to.to_string()], + None => metas + .iter() + .filter(|(_, m)| m.subscriptions.contains(from)) + .map(|(id, _)| id.clone()) + .collect(), + }; + let tappers: Vec = metas.iter().filter(|(_, m)| m.tapped).map(|(id, _)| id.clone()).collect(); + (recipients, tappers) }; + if !tappers.is_empty() { + let message = json!({"id": id, "from": from, "to": to, "reply_to": reply_to, "body": body}); + self.tap(&tappers, message); + } for to in recipients { self.deliver(Envelope { id: id.clone(), @@ -156,6 +163,25 @@ impl Broker { id } + /// Give every tapper a copy of one routed message. Copies skip `send`, so + /// 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}); + let mailboxes = self.mailboxes.lock().unwrap(); + for tapper in tappers { + if let Some(tx) = mailboxes.get(tapper) { + let _ = tx.try_send(Envelope { + id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)), + from: "broker".into(), + to: tapper.clone(), + reply_to: None, + body: body.clone(), + }); + } + } + } + /// Route one stdout line of a WASM actor. pub fn route_line(&self, from: &str, line: &str) { let line = line.trim(); @@ -467,6 +493,12 @@ impl Broker { })?; Ok(json!({"type": format!("{ty}d"), "actor": publisher})) } + "tap" | "untap" => { + let on = ty == "tap"; + self.update_meta(&env.from, |m| m.tapped = on)?; + let answer = if on { "tapping" } else { "untapped" }; + Ok(json!({"type": answer})) + } "scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})), "exit" => { self.log("broker", &format!("exit requested by {}", env.from)); @@ -474,7 +506,7 @@ impl Broker { self.exit.notify_one(); Ok(json!({"type": "exiting"})) } - other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, scan, exit"), + other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, tap, untap, scan, exit"), } } diff --git a/cli/src/world.rs b/cli/src/world.rs index e7d78af..7d16a4b 100644 --- a/cli/src/world.rs +++ b/cli/src/world.rs @@ -5,7 +5,7 @@ //! / //! actor// //! actor.wasm the current version -//! actor.json version, running, subscriptions (broker) +//! actor.json version, running, subscriptions, tapped (broker) //! source/ source of the current version //! public/ written by the actor, read-only for others //! data/ state, writable only by the actor @@ -43,6 +43,9 @@ pub struct ActorMeta { /// The actors whose published messages this actor receives. #[serde(default)] pub subscriptions: BTreeSet, + /// Whether this actor gets a copy of every routed message (`tap`). + #[serde(default)] + pub tapped: bool, } fn yes() -> bool { diff --git a/cli/tests/host.rs b/cli/tests/host.rs index 81ee5e6..b6f8b78 100644 --- a/cli/tests/host.rs +++ b/cli/tests/host.rs @@ -742,3 +742,114 @@ async fn running_actors_have_sockets_and_candidates_do_not() { assert_eq!(answer.body, json!({"type": "tcp", "line": "echo: hi"})); host.stop().await; } + +/// 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 +} + +#[tokio::test] +async fn tap_copies_every_routed_message_once() { + let world = create_world("tap"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + spawn_type(&host, &mut mailbox, "tapped", &wasm, &source).await; + let subscribe = json!({"type": "subscribe", "actor": "tapped"}); + assert_eq!(request(&host, &mut mailbox, subscribe).await.body["type"], "subscribed"); + assert_eq!(request(&host, &mut mailbox, json!({"type": "tap"})).await.body["type"], "tapping"); + + let echo = json!({"type": "echo", "value": "hi"}); + let echo_id = host.broker.send("console", Some("tapped"), None, None, echo.clone()); + let publish = json!({"type": "publish", "value": "news"}); + host.broker.send("console", Some("tapped"), None, None, publish); + let messages = drain(&mut mailbox, Duration::from_secs(2)).await; + let copies: Vec<&Value> = messages + .iter() + .filter(|m| m.from == "broker" && m.body["type"] == "tapped") + .map(|m| &m.body) + .collect(); + // 4 copies (2 requests, 1 reply, 1 publish) + the reply + the publish itself. + assert_eq!((copies.len(), messages.len()), (4, 6), "{messages:?}"); + for copy in &copies { + assert!(copy["ts"].as_u64().unwrap() > 0, "{copy}"); + } + let count = |want: Value| copies.iter().filter(|c| c["message"] == want).count(); + let request_copy = json!({"id": echo_id, "from": "console", "to": "tapped", "reply_to": null, "body": echo}); + assert_eq!(count(request_copy), 1, "{copies:?}"); + let reply_copies = copies + .iter() + .filter(|c| c["message"]["reply_to"] == echo_id.as_str() && c["message"]["from"] == "tapped" && c["message"]["to"] == "console") + .count(); + assert_eq!(reply_copies, 1, "{copies:?}"); + let publish_copy = copies.iter().find(|c| c["message"]["body"]["type"] == "published").expect("publish not tapped"); + assert_eq!(publish_copy["message"]["from"], "tapped"); + assert_eq!(publish_copy["message"]["to"], Value::Null); + + assert_eq!(request(&host, &mut mailbox, json!({"type": "untap"})).await.body["type"], "untapped"); + let echo_id = host.broker.send("console", Some("tapped"), None, None, json!({"type": "echo", "value": "quiet"})); + let messages = drain(&mut mailbox, Duration::from_secs(1)).await; + assert_eq!(messages.len(), 1, "{messages:?}"); + assert_eq!(messages[0].reply_to.as_deref(), Some(echo_id.as_str())); + host.stop().await; +} + +#[tokio::test] +async fn tap_survives_a_restart() { + let world = create_world("tap-restart"); + let (host, mut mailbox) = start(&world).await; + assert_eq!(request(&host, &mut mailbox, json!({"type": "tap"})).await.body["type"], "tapping"); + host.stop().await; + assert!(world.read_meta("console").tapped); + + let (host, mut mailbox) = start(&world).await; + let id = host.broker.send("console", Some("nobody"), None, None, json!({"type": "hello"})); + let copy = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let m = mailbox.recv().await.expect("console mailbox closed"); + if m.body["type"] == "tapped" && m.body["message"]["id"] == id.as_str() { + return m; + } + } + }) + .await + .expect("no tapped copy after the restart"); + assert_eq!(copy.body["message"]["to"], "nobody"); + host.stop().await; +} + +#[tokio::test] +async fn a_full_tapper_mailbox_drops_copies_without_blocking() { + let world = create_world("tap-full"); + let (host, mut mailbox) = start(&world).await; + // "ghost" taps and never reads its mailbox. + let mut ghost = host.mailbox("ghost"); + let mut sink = host.mailbox("sink"); + host.broker.send("ghost", Some("broker"), None, None, json!({"type": "tap"})); + tokio::time::timeout(Duration::from_secs(10), async { + while !world.read_meta("ghost").tapped { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("tap was not saved"); + + let total = sausage::broker::MAILBOX_SIZE + 500; + for n in 0..total { + host.broker.send("console", Some("sink"), None, None, json!({"type": "n", "n": n})); + assert_eq!(sink.try_recv().unwrap().body["n"], n); + } + assert!(mailbox.try_recv().is_err(), "a dropped copy reported something to the sender"); + let mut held = 0; + while ghost.try_recv().is_ok() { + held += 1; + } + assert_eq!(held, sausage::broker::MAILBOX_SIZE); + host.broker.send("console", Some("sink"), None, None, json!({"type": "after"})); + assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped"); + host.stop().await; +}