notify router: deliveries log, dry-run hit counts, api and page

swarm-notify-router/Cargo.toml

@@ -30,5 +30,9 @@ swarm-wire.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 
+# A fresh directory per test `nats-server`, for its store and ports file.
+[dev-dependencies]
+tempfile = "3"
+
 [lints]
 workspace = true

swarm-notify-router/README.md

@@ -4,5 +4,8 @@ Routes forge events to agents' todo lists. It drains its consumer on the
 forge events stream that swarm-controller fills from its webhooks, runs its
 flows over each delivery, and queues the todo ops they emit on each agent's
 subject of the `ROUTER_TODOS` stream. Each agent's harness applies them.
+It keeps the newest 500 deliveries in `NOTIFY_ROUTER_LOG` and answers
+swarm-controller's requests for them and for a dry run of the flows over the
+newest 50 events.
 
 Design: [`docs/swarm/notify-router.md`](../docs/swarm/notify-router.md).

swarm-notify-router/src/api.rs

@@ -0,0 +1,288 @@
+//! The router's two answers to swarm-controller: the newest deliveries and a
+//! dry run (`swarm_wire::router_api`).
+//!
+//! Each is a NATS request. A dry run replays the newest forge events from the
+//! events stream through [`dry_run::run`] and delivers nothing, writes no
+//! snapshot and sends no review request. The router hears these on their own
+//! task, so a slow replay never holds up the events.
+
+use anyhow::Result;
+use async_nats::jetstream;
+use futures_util::StreamExt as _;
+use serde::Serialize;
+use swarm_wire::forge_events;
+use swarm_wire::router_api::{
+    DEFAULT_ROWS, DELIVERIES_SUBJECT, DRY_RUN_EVENTS, DRY_RUN_SUBJECT, Deliveries,
+    DeliveriesRequest, DryRunRequest, Refused,
+};
+
+use crate::{dry_run, log};
+
+/// Answer requests on both subjects until the connection closes.
+pub async fn serve(client: async_nats::Client) {
+    let js = jetstream::new(client.clone());
+    let (mut deliveries, mut dry) = match (
+        client.subscribe(DELIVERIES_SUBJECT).await,
+        client.subscribe(DRY_RUN_SUBJECT).await,
+    ) {
+        (Ok(a), Ok(b)) => (a, b),
+        (a, b) => {
+            let why = a
+                .err()
+                .map(|e| e.to_string())
+                .or(b.err().map(|e| e.to_string()));
+            tracing::error!(error = ?why, "notify router: cannot hear api requests; the controller will report them unavailable");
+            return;
+        }
+    };
+    tracing::info!("notify router: answering deliveries and dry-run requests");
+    loop {
+        let (request, dry_run) = tokio::select! {
+            Some(m) = deliveries.next() => (m, false),
+            Some(m) = dry.next() => (m, true),
+            else => return,
+        };
+        let (client, js) = (client.clone(), js.clone());
+        tokio::spawn(async move {
+            let Some(reply) = request.reply else { return };
+            let answer = if dry_run {
+                dry_run_answer(&js, &request.payload).await
+            } else {
+                deliveries_answer(&js, &request.payload).await
+            };
+            if let Err(e) = client.publish(reply, answer.into()).await {
+                tracing::warn!(error = %e, "notify router: could not answer an api request");
+            }
+        });
+    }
+}
+
+/// The reply body: `value`, or a [`Refused`] saying why not.
+fn body<T: Serialize>(answer: Result<T>) -> Vec<u8> {
+    match answer {
+        Ok(value) => serde_json::to_vec(&value).expect("an answer encodes"),
+        Err(e) => {
+            tracing::warn!(
+                error = format!("{e:#}"),
+                "notify router: an api request failed"
+            );
+            serde_json::to_vec(&Refused {
+                error: format!("{e:#}"),
+            })
+            .expect("a refusal encodes")
+        }
+    }
+}
+
+async fn deliveries_answer(js: &jetstream::Context, payload: &[u8]) -> Vec<u8> {
+    body(
+        async {
+            let request: DeliveriesRequest = if payload.is_empty() {
+                DeliveriesRequest::default()
+            } else {
+                serde_json::from_slice(payload)?
+            };
+            let limit = request.limit.unwrap_or(DEFAULT_ROWS);
+            Ok(Deliveries {
+                rows: log::recent(js, limit).await?,
+            })
+        }
+        .await,
+    )
+}
+
+async fn dry_run_answer(js: &jetstream::Context, payload: &[u8]) -> Vec<u8> {
+    body(
+        async {
+            let request: DryRunRequest = serde_json::from_slice(payload)?;
+            request
+                .document
+                .check()
+                .map_err(|why| anyhow::anyhow!("the document fails its check: {why}"))?;
+            let (flows, _) = request.document.active();
+            let mut events: Vec<(String, Vec<u8>)> =
+                log::newest(js, forge_events::STREAM, DRY_RUN_EVENTS)
+                    .await?
+                    .into_iter()
+                    .filter_map(|(subject, body)| {
+                        Some((forge_events::event_type(&subject)?.to_owned(), body))
+                    })
+                    .collect();
+            events.reverse();
+            Ok(dry_run::run(&flows, &events))
+        }
+        .await,
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use swarm_wire::notify_flows::Document;
+    use swarm_wire::router_api::{DryRun, Kind, LOG_STREAM, Row};
+    use swarm_wire::router_todo;
+
+    use super::*;
+    use crate::test_nats::TestNats;
+
+    /// Send a request to the router's api, retrying while its subscription
+    /// is not yet in place.
+    async fn ask(client: &async_nats::Client, subject: &str, body: Vec<u8>) -> Vec<u8> {
+        for _ in 0..50 {
+            if let Ok(m) = client
+                .request(subject.to_owned(), body.clone().into())
+                .await
+            {
+                return m.payload.to_vec();
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        panic!("nothing answered {subject}");
+    }
+
+    fn row(n: u64) -> Row {
+        log::Cause {
+            event: "issue_assign".to_owned(),
+            repo: "o/r".to_owned(),
+            number: n,
+        }
+        .row(
+            "assigned",
+            "sink",
+            Kind::Todo,
+            "atlas",
+            Ok("upsert k".to_owned()),
+        )
+    }
+
+    async fn publish_event(js: &jetstream::Context, event: &str, n: i64) {
+        let body = serde_json::json!({
+            "action": "assigned",
+            "repository": { "full_name": "o/r", "id": 1 },
+            "issue": { "number": n, "title": "t", "html_url": "u", "assignees": [{ "login": "atlas" }] },
+            "sender": { "login": "mara" },
+        });
+        js.publish(
+            forge_events::subject(event),
+            serde_json::to_vec(&body).unwrap().into(),
+        )
+        .await
+        .expect("published")
+        .await
+        .expect("acked");
+    }
+
+    /// The newest rows come back newest first, no more than asked for, and
+    /// the stream keeps 500 of them.
+    #[tokio::test]
+    async fn the_deliveries_are_the_newest_rows_and_the_log_keeps_500() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let js = jetstream::new(client.clone());
+        log::open_or_create(&js).await.expect("the log stream");
+        tokio::spawn(serve(client.clone()));
+        for n in 1..=3 {
+            log::append(&js, &row(n)).await;
+        }
+
+        let all: Deliveries =
+            serde_json::from_slice(&ask(&client, DELIVERIES_SUBJECT, Vec::new()).await)
+                .expect("an answer");
+        let numbers: Vec<u64> = all.rows.iter().map(|r| r.number).collect();
+        assert_eq!(numbers, [3, 2, 1]);
+        let first = Row {
+            at: all.rows[0].at,
+            ..row(3)
+        };
+        assert_eq!(all.rows[0], first);
+
+        let two: Deliveries = serde_json::from_slice(
+            &ask(&client, DELIVERIES_SUBJECT, br#"{"limit":2}"#.to_vec()).await,
+        )
+        .expect("an answer");
+        assert_eq!(
+            two.rows.iter().map(|r| r.number).collect::<Vec<_>>(),
+            [3, 2]
+        );
+
+        for n in 4..=503 {
+            log::append(&js, &row(n)).await;
+        }
+        let mut stream = js.get_stream(LOG_STREAM).await.expect("the stream");
+        assert_eq!(stream.info().await.expect("info").state.messages, 500);
+        let newest: Deliveries = serde_json::from_slice(
+            &ask(&client, DELIVERIES_SUBJECT, br#"{"limit":9999}"#.to_vec()).await,
+        )
+        .expect("an answer");
+        assert_eq!(newest.rows.len(), 500);
+        assert_eq!(newest.rows[0].number, 503);
+        assert_eq!(newest.rows[499].number, 4);
+    }
+
+    /// A dry run counts the newest 50 events, and delivers nothing: no todo
+    /// op, no log row.
+    #[tokio::test]
+    async fn a_dry_run_counts_the_newest_fifty_events_and_delivers_nothing() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let js = jetstream::new(client.clone());
+        swarm_queue_client::forge_events::open_or_create(&client)
+            .await
+            .expect("events");
+        swarm_queue_client::router_todo::open_or_create(&client)
+            .await
+            .expect("todos");
+        log::open_or_create(&js).await.expect("the log stream");
+        tokio::spawn(serve(client.clone()));
+        // 60 assign events on distinct issues, then one on a type no flow
+        // starts from.
+        for n in 1..=60 {
+            publish_event(&js, "issue_assign", n).await;
+        }
+        publish_event(&js, "issue_comment", 61).await;
+
+        let request = serde_json::to_vec(&DryRunRequest {
+            document: Document::default(),
+        })
+        .unwrap();
+        let reply = ask(&client, DRY_RUN_SUBJECT, request).await;
+        let done: DryRun = serde_json::from_slice(&reply).expect("a dry run");
+        assert_eq!(done.events, 50);
+        let hits = |flow: &str, node: &str| {
+            done.flows
+                .iter()
+                .find(|f| f.flow == flow)
+                .and_then(|f| f.nodes.iter().find(|n| n.node == node))
+                .map(|n| n.hits)
+        };
+        // 49 of the newest 50 are assign events; the 50th is the comment.
+        assert_eq!(hits("assigned", "source"), Some(49));
+        assert_eq!(hits("assigned", "sink"), Some(49));
+        assert_eq!(hits("unassigned", "sink"), Some(0));
+
+        for (stream, what) in [(router_todo::STREAM, "todo op"), (LOG_STREAM, "log row")] {
+            let mut s = js.get_stream(stream).await.expect("the stream");
+            assert_eq!(s.info().await.expect("info").state.messages, 0, "a {what}");
+        }
+    }
+
+    /// A document that fails its check is refused, and says why.
+    #[tokio::test]
+    async fn a_document_that_fails_its_check_is_refused() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        tokio::spawn(serve(client.clone()));
+        let doc = Document {
+            disabled_builtins: vec!["no-such-flow".to_owned()],
+            ..Document::default()
+        };
+        let request = serde_json::to_vec(&DryRunRequest { document: doc }).unwrap();
+        let refused: Refused =
+            serde_json::from_slice(&ask(&client, DRY_RUN_SUBJECT, request).await)
+                .expect("a refusal");
+        assert!(refused.error.contains("no-such-flow"), "{}", refused.error);
+        let garbled = ask(&client, DRY_RUN_SUBJECT, b"nonsense".to_vec()).await;
+        assert!(serde_json::from_slice::<Refused>(&garbled).is_ok());
+    }
+}

swarm-notify-router/src/dry_run.rs

@@ -0,0 +1,183 @@
+//! Running a flows document over past events without delivering anything.
+//!
+//! Each event goes through the walk the router delivers with ([`flow::reached`]
+//! and [`route::route`]), counting the nodes it reaches. The per-issue
+//! snapshots start empty and carry from one event to the next, as the router's
+//! do, so an assign event's `assignees_added` is the difference to the
+//! previous replayed one. Every recipient counts as an agent.
+
+use std::collections::BTreeMap;
+
+use swarm_wire::notify_flows::Flow;
+use swarm_wire::router_api::{DryRun, FlowHits, NodeHits};
+
+use crate::event::{Delivery, Snapshot, fields};
+use crate::{flow, route};
+
+/// `flows` over `events`, oldest first, each a forge event type and body.
+/// An event that names no issue or PR is not counted.
+pub fn run(flows: &[Flow], events: &[(String, Vec<u8>)]) -> DryRun {
+    let mut hits: Vec<BTreeMap<&str, u64>> = flows
+        .iter()
+        .map(|f| f.nodes.iter().map(|n| (n.id.as_str(), 0)).collect())
+        .collect();
+    let mut snapshots: BTreeMap<String, Snapshot> = BTreeMap::new();
+    let mut run = 0;
+    for (event_type, body) in events {
+        let Some(delivery) = Delivery::parse(body) else {
+            continue;
+        };
+        run += 1;
+        let key = delivery.state_key();
+        let before = snapshots.remove(&key).unwrap_or_default();
+        let event_fields = fields(event_type, &delivery, &before);
+        for (flow, counts) in flows.iter().zip(&mut hits) {
+            for id in flow::reached(flow, event_type, &event_fields) {
+                if let Some(count) = counts.get_mut(id) {
+                    *count += 1;
+                }
+            }
+        }
+        let routed = route::route(event_type, &delivery, &before, flows, |_| true);
+        snapshots.insert(key, routed.next);
+    }
+    DryRun {
+        events: run,
+        flows: flows
+            .iter()
+            .zip(hits)
+            .map(|(flow, counts)| FlowHits {
+                flow: flow.id.clone(),
+                nodes: flow
+                    .nodes
+                    .iter()
+                    .map(|n| NodeHits {
+                        node: n.id.clone(),
+                        hits: counts.get(n.id.as_str()).copied().unwrap_or(0),
+                    })
+                    .collect(),
+            })
+            .collect(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use swarm_wire::notify_flows::{Document, builtins};
+
+    use super::*;
+
+    fn body(action: &str, number: i64, assignees: &[&str]) -> Vec<u8> {
+        let people: Vec<_> = assignees
+            .iter()
+            .map(|a| serde_json::json!({ "login": a }))
+            .collect();
+        serde_json::to_vec(&serde_json::json!({
+            "action": action,
+            "repository": { "full_name": "o/r", "id": 1 },
+            "issue": { "number": number, "title": "t", "html_url": "u", "assignees": people },
+            "sender": { "login": "mara" },
+        }))
+        .expect("encodes")
+    }
+
+    fn count(r: &DryRun, flow: &str, node: &str) -> u64 {
+        r.flows
+            .iter()
+            .find(|f| f.flow == flow)
+            .and_then(|f| f.nodes.iter().find(|n| n.node == node))
+            .unwrap_or_else(|| panic!("no node {flow}/{node}"))
+            .hits
+    }
+
+    /// The built-ins over an assign, a second assign that adds nobody, an
+    /// unassign and a close: each node counts the events that reached it.
+    #[test]
+    fn each_node_counts_the_events_that_reached_it() {
+        let (flows, _) = Document::default().active();
+        let events = vec![
+            ("issue_assign".to_owned(), body("assigned", 5, &["atlas"])),
+            ("issue_assign".to_owned(), body("assigned", 5, &["atlas"])),
+            ("issue_assign".to_owned(), body("unassigned", 5, &[])),
+            ("issues".to_owned(), body("closed", 5, &[])),
+            ("issues".to_owned(), b"not json".to_vec()),
+        ];
+        let r = run(&flows, &events);
+        assert_eq!(r.events, 4);
+        // The source sees all three assign events, the match and the sink both
+        // assigned ones, though the second adds nobody.
+        assert_eq!(count(&r, "assigned", "source"), 3);
+        assert_eq!(count(&r, "assigned", "match"), 2);
+        assert_eq!(count(&r, "assigned", "sink"), 2);
+        // Only the unassign reaches the unassigned flow's sink.
+        assert_eq!(count(&r, "unassigned", "source"), 3);
+        assert_eq!(count(&r, "unassigned", "sink"), 1);
+        // The close is an `issues` event, which these flows start from only
+        // in `closed`.
+        assert_eq!(count(&r, "closed", "source"), 1);
+        assert_eq!(count(&r, "closed", "sink"), 1);
+        // No pull request push: the classifier flow counts zero.
+        assert_eq!(count(&r, "head-classifier", "source"), 0);
+        assert_eq!(count(&r, "head-classifier", "request"), 0);
+    }
+
+    /// A condition on a field that depends on the previous event of the
+    /// issue (`assignees_added`) holds for the first assign event and not for
+    /// an identical second one.
+    #[test]
+    fn the_issues_snapshot_carries_from_one_replayed_event_to_the_next() {
+        use swarm_wire::notify_flows::{Condition, Edge, Node, NodeKind};
+
+        let node = |id: &str, kind| Node {
+            id: id.to_owned(),
+            kind,
+        };
+        let edge = |from: &str, to: &str| Edge {
+            from: from.to_owned(),
+            to: to.to_owned(),
+        };
+        let flow = Flow {
+            id: "added".to_owned(),
+            name: "added".to_owned(),
+            enabled: true,
+            nodes: vec![
+                node(
+                    "source",
+                    NodeKind::Source {
+                        events: vec!["issue_assign".to_owned()],
+                    },
+                ),
+                node(
+                    "atlas",
+                    NodeKind::Match {
+                        when: vec![Condition {
+                            field: "assignees_added".to_owned(),
+                            equals: "atlas".to_owned(),
+                        }],
+                    },
+                ),
+            ],
+            edges: vec![edge("source", "atlas")],
+        };
+        let assign = ("issue_assign".to_owned(), body("assigned", 5, &["atlas"]));
+        let r = run(&[flow], &[assign.clone(), assign]);
+        assert_eq!(count(&r, "added", "source"), 2);
+        assert_eq!(count(&r, "added", "atlas"), 1);
+    }
+
+    /// Only the flows the document turns on are run.
+    #[test]
+    fn a_switched_off_built_in_is_not_listed() {
+        let doc = Document {
+            disabled_builtins: vec!["closed".to_owned()],
+            ..Document::default()
+        };
+        let (flows, _) = doc.active();
+        let r = run(&flows, &[]);
+        assert_eq!(r.events, 0);
+        let listed: Vec<&str> = r.flows.iter().map(|f| f.flow.as_str()).collect();
+        assert!(!listed.contains(&"closed"));
+        assert!(listed.contains(&"assigned"));
+        assert_eq!(listed.len(), builtins().len() - 1);
+    }
+}

swarm-notify-router/src/flow.rs

@@ -43,10 +43,10 @@ pub struct ReviewRequest {
     pub reviewer: String,
 }
 
-/// Walk `flow` for one event of type `event_type`, calling `leaf` on each
-/// `todo` and `request_reviewer` node the event reaches. Assumes
-/// [`Flow::validate`] passed.
-fn walk<'f>(flow: &'f Flow, event_type: &str, fields: &Fields, mut leaf: impl FnMut(&'f Node)) {
+/// Walk `flow` for one event of type `event_type`, calling `visit` on each node
+/// the event reaches: the `source`, each `match` it passes, and the `todo` and
+/// `request_reviewer` nodes below. Assumes [`Flow::validate`] passed.
+fn walk<'f>(flow: &'f Flow, event_type: &str, fields: &Fields, mut visit: impl FnMut(&'f Node)) {
     let Some(root) = flow
         .nodes
         .iter()
@@ -58,17 +58,30 @@ fn walk<'f>(flow: &'f Flow, event_type: &str, fields: &Fields, mut leaf: impl Fn
     while let Some(id) = stack.pop() {
         let Some(node) = flow.node(id) else { continue };
         match &node.kind {
-            NodeKind::Source { .. } => stack.extend(flow.children(id).into_iter().rev()),
+            NodeKind::Source { .. } => {
+                visit(node);
+                stack.extend(flow.children(id).into_iter().rev());
+            }
             NodeKind::Match { when } => {
                 if when.iter().all(|c| holds(c, fields)) {
+                    visit(node);
                     stack.extend(flow.children(id).into_iter().rev());
                 }
             }
-            NodeKind::Todo(_) | NodeKind::RequestReviewer { .. } => leaf(node),
+            NodeKind::Todo(_) | NodeKind::RequestReviewer { .. } => visit(node),
         }
     }
 }
 
+/// The ids of the nodes of `flow` that one event of type `event_type` reaches,
+/// a `match` node counting only when it passes. Emits nothing: this is the
+/// dry run's view of the walk that [`evaluate`] and [`requests`] make.
+pub fn reached<'f>(flow: &'f Flow, event_type: &str, fields: &Fields) -> Vec<&'f str> {
+    let mut ids = Vec::new();
+    walk(flow, event_type, fields, |node| ids.push(node.id.as_str()));
+    ids
+}
+
 /// Walk `flow` for one event of type `event_type` and return the todo ops it
 /// emits. Assumes [`Flow::validate`] passed.
 pub fn evaluate(flow: &Flow, event_type: &str, fields: &Fields) -> Vec<Action> {

swarm-notify-router/src/log.rs

@@ -0,0 +1,134 @@
+//! The deliveries log: one row on the `NOTIFY_ROUTER_LOG` stream for each
+//! delivery the router makes (`swarm_wire::router_api`), and reading the
+//! newest rows back.
+//!
+//! The stream keeps the newest 500 rows. Appending is best effort: a log that
+//! cannot be written is reported and never holds up a delivery.
+
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use anyhow::{Context as _, Result};
+use async_nats::jetstream::stream::{Config, DiscardPolicy, Stream};
+use async_nats::jetstream::{self};
+use futures_util::{StreamExt as _, stream};
+use swarm_wire::router_api::{Kind, LOG_MAX_MSGS, LOG_STREAM, LOG_SUBJECT, Row};
+
+/// How many messages are fetched at once.
+const FETCH_AT_ONCE: usize = 16;
+
+/// Open the stream, creating it if nothing has yet. The router is its one
+/// writer and its one creator.
+pub async fn open_or_create(js: &jetstream::Context) -> Result<Stream> {
+    match js.get_stream(LOG_STREAM).await {
+        Ok(stream) => Ok(stream),
+        Err(e) => {
+            tracing::info!(stream = LOG_STREAM, reason = %e, "notify router: log stream not available, creating it");
+            js.create_stream(Config {
+                name: LOG_STREAM.to_owned(),
+                description: Some("Notification router deliveries, newest 500".to_owned()),
+                subjects: vec![LOG_SUBJECT.to_owned()],
+                max_messages: LOG_MAX_MSGS,
+                discard: DiscardPolicy::Old,
+                ..Default::default()
+            })
+            .await
+            .context("creating the log stream")
+        }
+    }
+}
+
+/// The event a delivery answers.
+#[derive(Clone, Debug)]
+pub struct Cause {
+    pub event: String,
+    pub repo: String,
+    pub number: u64,
+}
+
+impl Cause {
+    /// A row of `kind` made at `flow`'s `node` for `target`, which went
+    /// through when `result` is `Ok`.
+    pub fn row(
+        &self,
+        flow: &str,
+        node: &str,
+        kind: Kind,
+        target: &str,
+        result: Result<String, String>,
+    ) -> Row {
+        let (ok, detail) = match result {
+            Ok(detail) => (true, detail),
+            Err(why) => (false, why),
+        };
+        Row {
+            at: SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .map_or(0, |d| d.as_secs()),
+            flow: flow.to_owned(),
+            node: node.to_owned(),
+            event: self.event.clone(),
+            repo: self.repo.clone(),
+            number: self.number,
+            kind,
+            target: target.to_owned(),
+            ok,
+            detail,
+        }
+    }
+}
+
+/// Append `row`; a failure is logged and dropped.
+pub async fn append(js: &jetstream::Context, row: &Row) {
+    let body = serde_json::to_vec(row).expect("a row encodes");
+    let sent = async {
+        js.publish(LOG_SUBJECT, body.into())
+            .await
+            .context("publishing")?
+            .await
+            .context("waiting for the ack")
+    };
+    if let Err(e) = sent.await {
+        tracing::warn!(error = format!("{e:#}"), flow = %row.flow, "notify router: could not log a delivery");
+    }
+}
+
+/// The payloads of the newest `limit` messages of `stream`, newest first,
+/// with their subjects. A sequence that no longer exists is skipped.
+pub async fn newest(
+    js: &jetstream::Context,
+    stream: &str,
+    limit: usize,
+) -> Result<Vec<(String, Vec<u8>)>> {
+    let stream = js
+        .get_stream(stream)
+        .await
+        .with_context(|| format!("opening {stream}"))?;
+    let state = &stream.cached_info().state;
+    let (first, last) = (state.first_sequence, state.last_sequence);
+    let count = u64::try_from(limit).unwrap_or(u64::MAX);
+    let from = first.max((last + 1).saturating_sub(count));
+    let seqs = (from..=last).rev();
+    let fetched: Vec<_> = stream::iter(seqs)
+        .map(|seq| {
+            let stream = &stream;
+            async move { stream.get_raw_message(seq).await }
+        })
+        .buffered(FETCH_AT_ONCE)
+        .collect()
+        .await;
+    Ok(fetched
+        .into_iter()
+        .filter_map(Result::ok)
+        .map(|m| (m.subject.to_string(), m.payload.to_vec()))
+        .collect())
+}
+
+/// The newest `limit` rows of the log, newest first. A row that does not
+/// decode is skipped.
+pub async fn recent(js: &jetstream::Context, limit: usize) -> Result<Vec<Row>> {
+    Ok(newest(js, LOG_STREAM, limit)
+        .await?
+        .into_iter()
+        .filter_map(|(_, body)| serde_json::from_slice(&body).ok())
+        .collect())
+}

swarm-notify-router/src/main.rs

@@ -9,10 +9,15 @@
 //!
 //! A `request_reviewer` node is a request to swarm-controller, which holds the
 //! forge token (`swarm_wire::router_sink`). It is sent once, after the event's
-//! ops are queued and its snapshot is written, and the answer is only logged:
+//! ops are queued and its snapshot is written, and the answer is only logged and
+//! recorded:
 //! a controller that is down or slow must not hold up every agent's todos, and
 //! the next push asks again.
 //!
+//! Each delivery, made or failed, is a row of the deliveries log ([`log`]), and
+//! the router answers swarm-controller's requests for the newest rows and for a
+//! dry run of a flows document ([`api`], [`dry_run`]).
+//!
 //! Per issue it keeps a snapshot in its state bucket ([`event::Snapshot`]):
 //! the assignees it last saw and the agents it queued a todo for. An event is
 //! acked once its ops are queued and the snapshot is written, so a crash
@@ -32,14 +37,20 @@ use async_nats::jetstream::{self, consumer::pull, kv, message::PublishMessage};
 use futures_util::{StreamExt as _, TryStreamExt as _};
 use swarm_queue_client::{ClientSecret, QueueConfig};
 use swarm_secret_client::stored_secret::StoredSecret;
+use swarm_wire::router_api::Kind;
 use swarm_wire::router_sink::{self, Outcome, RequestReviewer};
 use swarm_wire::{forge_events, notify_router, router_todo};
 
+mod api;
+mod dry_run;
 mod event;
 mod flow;
+mod log;
 mod route;
 mod settle;
 mod stored;
+#[cfg(test)]
+mod test_nats;
 
 /// The cert-auth role this service logs in to the store as. `swarm-bao.nix`
 /// writes it, matching the leaf's subject.
@@ -93,6 +104,7 @@ async fn main() -> Result<()> {
     .await?;
 
     let mut router = Router::open(&queue).await?;
+    tokio::spawn(api::serve(queue.clone()));
     tracing::info!(
         stream = forge_events::STREAM,
         "notify router: routing forge events"
@@ -126,6 +138,9 @@ impl Router {
         swarm_queue_client::router_todo::open_or_create(queue)
             .await
             .context("opening the router todo stream")?;
+        log::open_or_create(&js)
+            .await
+            .context("opening the deliveries log stream")?;
         let events = events
             .create_consumer(pull::Config {
                 durable_name: Some(forge_events::ROUTER_CONSUMER.to_owned()),
@@ -199,8 +214,13 @@ impl Router {
         let routed = route::route(&event_type, &delivery, &snapshot, &self.flows.active, |a| {
             roster.contains(a)
         });
+        let cause = log::Cause {
+            event: event_type.clone(),
+            repo: delivery.repo.clone(),
+            number: u64::try_from(delivery.number).unwrap_or_default(),
+        };
         for action in &routed.actions {
-            self.queue(seq, action).await.map_err(Skip::Retry)?;
+            self.queue(seq, &cause, action).await.map_err(Skip::Retry)?;
         }
         if !routed.skipped.is_empty() {
             tracing::info!(%event_type, repo = %delivery.repo, number = delivery.number, skipped = ?routed.skipped, "notify router: recipients that are not agents got nothing");
@@ -210,14 +230,14 @@ impl Router {
             self.state.put(&key, bytes.into()).await.map_err(retry)?;
         }
         for request in &routed.requests {
-            self.request_review(request).await;
+            self.request_review(&cause, request).await;
         }
         Ok(())
     }
 
-    /// Ask swarm-controller for one review and log what it answered. One
-    /// attempt: nothing here fails the event.
-    async fn request_review(&self, request: &flow::ReviewRequest) {
+    /// Ask swarm-controller for one review, log what it answered and record
+    /// the delivery. One attempt: nothing here fails the event.
+    async fn request_review(&self, cause: &log::Cause, request: &flow::ReviewRequest) {
         let wire = RequestReviewer {
             repo: request.repo.clone(),
             number: request.number,
@@ -233,7 +253,7 @@ impl Router {
             request.number,
             &request.reviewer,
         );
-        match self
+        let result = match self
             .nats
             .send_request(router_sink::REQUEST_REVIEWER_SUBJECT, sent)
             .await
@@ -241,22 +261,46 @@ impl Router {
             Ok(reply) => match Outcome::decode(&String::from_utf8_lossy(&reply.payload)) {
                 Outcome::Requested => {
                     tracing::info!(%flow, %repo, number, %reviewer, "notify router: review requested");
+                    Ok("requested".to_owned())
                 }
                 Outcome::Skipped(why) => {
                     tracing::info!(%flow, %repo, number, %reviewer, %why, "notify router: review not requested");
+                    Ok(format!("skipped: {why}"))
                 }
                 Outcome::Error(why) => {
                     tracing::warn!(%flow, %repo, number, %reviewer, %why, "notify router: the controller could not request the review");
+                    Err(why)
                 }
             },
             Err(e) => {
                 tracing::warn!(%flow, %repo, number, %reviewer, error = %e, "notify router: no answer to a review request; not retried");
+                Err(format!("no answer: {e}"))
             }
-        }
+        };
+        let row = cause.row(flow, &request.node, Kind::RequestReviewer, reviewer, result);
+        log::append(&self.js, &row).await;
     }
 
-    /// Queue one op on its agent's subject and wait for the stream's ack.
-    async fn queue(&self, seq: u64, action: &flow::Action) -> Result<()> {
+    /// Queue one op on its agent's subject, wait for the stream's ack and
+    /// record the delivery, failed or not.
+    async fn queue(&self, seq: u64, cause: &log::Cause, action: &flow::Action) -> Result<()> {
+        let sent = self.publish(seq, action).await;
+        let result = match &sent {
+            Ok(()) => Ok(format!("{} {}", op_name(action.op.op), action.op.key)),
+            Err(e) => Err(format!("{e:#}")),
+        };
+        let row = cause.row(
+            &action.flow,
+            &action.node,
+            Kind::Todo,
+            &action.agent,
+            result,
+        );
+        log::append(&self.js, &row).await;
+        sent
+    }
+
+    async fn publish(&self, seq: u64, action: &flow::Action) -> Result<()> {
         let subject = router_todo::subject(&action.agent);
         let body = serde_json::to_vec(&action.op).expect("an op encodes");
         let id = format!("{seq}:{}:{}:{}", action.flow, action.node, action.agent);
@@ -319,6 +363,14 @@ fn retry(e: impl std::error::Error + Send + Sync + 'static) -> Skip {
     Skip::Retry(e.into())
 }
 
+/// The word the log uses for an op.
+const fn op_name(op: router_todo::Op) -> &'static str {
+    match op {
+        router_todo::Op::Upsert => "upsert",
+        router_todo::Op::Clear => "clear",
+    }
+}
+
 async fn client_secret(client_id: &str) -> Result<String> {
     use swarm_secret_client::builder::{ClientSecret, client_secret_path};
 
@@ -338,3 +390,148 @@ async fn client_secret(client_id: &str) -> Result<String> {
         ),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use swarm_wire::notify_flows::builtins;
+    use swarm_wire::router_api::Row;
+
+    use super::*;
+    use crate::test_nats::TestNats;
+
+    fn cause() -> log::Cause {
+        log::Cause {
+            event: "issue_assign".to_owned(),
+            repo: "o/r".to_owned(),
+            number: 9,
+        }
+    }
+
+    fn action() -> flow::Action {
+        let flow = builtins().remove(0);
+        let (agent, op) = (
+            "atlas".to_owned(),
+            router_todo::RouterTodoOp {
+                op: router_todo::Op::Upsert,
+                key: "forge:o/r#9:assignment".to_owned(),
+                prefix: false,
+                summary: "s".to_owned(),
+                source: "router:assigned".to_owned(),
+            },
+        );
+        flow::Action {
+            flow: flow.id,
+            node: "sink".to_owned(),
+            agent,
+            op,
+        }
+    }
+
+    fn review() -> flow::ReviewRequest {
+        flow::ReviewRequest {
+            flow: "head-classifier".to_owned(),
+            node: "request".to_owned(),
+            repo: "o/r".to_owned(),
+            number: 9,
+            reviewer: "review-classifier".to_owned(),
+        }
+    }
+
+    async fn rows(router: &Router) -> Vec<Row> {
+        log::recent(&router.js, 10).await.expect("the log")
+    }
+
+    /// A todo op that was queued is a row that went through; one the stream
+    /// would not take is a row with the error, and the delivery still fails.
+    #[tokio::test]
+    async fn a_queued_todo_is_a_row_and_a_failed_one_is_a_row_with_the_error() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let router = Router::open(&client).await.expect("opens");
+
+        router.queue(5, &cause(), &action()).await.expect("queued");
+        let [row] = &rows(&router).await[..] else {
+            panic!("one row")
+        };
+        assert!(row.ok, "{row:?}");
+        assert_eq!(
+            (row.flow.as_str(), row.node.as_str(), row.target.as_str()),
+            ("assigned", "sink", "atlas")
+        );
+        assert_eq!(
+            (row.event.as_str(), row.repo.as_str(), row.number),
+            ("issue_assign", "o/r", 9)
+        );
+        assert_eq!(row.kind, Kind::Todo);
+        assert_eq!(row.detail, "upsert forge:o/r#9:assignment");
+
+        router
+            .js
+            .delete_stream(router_todo::STREAM)
+            .await
+            .expect("deleted");
+        router
+            .queue(6, &cause(), &action())
+            .await
+            .expect_err("no stream takes it");
+        let rows = rows(&router).await;
+        assert_eq!(rows.len(), 2);
+        assert!(!rows[0].ok && !rows[0].detail.is_empty(), "{:?}", rows[0]);
+    }
+
+    /// Each answer the controller can give is a row: requested and skipped
+    /// went through, an error did not, and no answer at all did not either.
+    #[tokio::test]
+    async fn a_review_request_is_a_row_whatever_the_controller_answers() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let router = Router::open(&client).await.expect("opens");
+
+        // Nobody answers.
+        router.request_review(&cause(), &review()).await;
+        let first = rows(&router).await.remove(0);
+        assert!(
+            !first.ok && first.detail.starts_with("no answer"),
+            "{first:?}"
+        );
+        assert_eq!(first.kind, Kind::RequestReviewer);
+        assert_eq!(first.target, "review-classifier");
+
+        let mut hears = client
+            .subscribe(router_sink::REQUEST_REVIEWER_SUBJECT)
+            .await
+            .expect("subscribed");
+        let answering = client.clone();
+        tokio::spawn(async move {
+            for line in [
+                "requested",
+                "skipped:a review stands",
+                "error:the forge said no",
+            ] {
+                let m = hears.next().await.expect("a request");
+                answering
+                    .publish(m.reply.expect("a reply subject"), line.into())
+                    .await
+                    .expect("answered");
+            }
+        });
+        for _ in 0..3 {
+            router.request_review(&cause(), &review()).await;
+        }
+        let got: Vec<(bool, String)> = rows(&router)
+            .await
+            .into_iter()
+            .take(3)
+            .rev()
+            .map(|r| (r.ok, r.detail))
+            .collect();
+        assert_eq!(
+            got,
+            [
+                (true, "requested".to_owned()),
+                (true, "skipped: a review stands".to_owned()),
+                (false, "the forge said no".to_owned()),
+            ]
+        );
+    }
+}

swarm-notify-router/src/test_nats.rs

@@ -0,0 +1,77 @@
+//! A throwaway `nats-server` with `JetStream`, for the tests whose subject is
+//! the server's own behaviour: what a stream keeps and what a request gets.
+//!
+//! The same shape as `swarm-queue-client`'s: the binary comes from
+//! [`NATS_SERVER_ENV`], which `nix/checks.nix` and `nix/devshell.nix` export,
+//! and a test that needs it panics when it is unset rather than skipping.
+
+use std::path::Path;
+use std::process::{Child, Command, Stdio};
+use std::time::Duration;
+
+/// The variable naming the `nats-server` binary.
+const NATS_SERVER_ENV: &str = "HIVE_NATS_SERVER";
+
+/// The subdirectory holding nothing but the server's ports file.
+const PORTS_DIR: &str = "ports";
+
+/// A running server on a loopback port it bound itself. Dropping it kills
+/// the server and removes its directory.
+pub struct TestNats {
+    child: Child,
+    dir: tempfile::TempDir,
+}
+
+impl TestNats {
+    pub fn start() -> Self {
+        let bin = std::env::var(NATS_SERVER_ENV).unwrap_or_else(|_| {
+            panic!("{NATS_SERVER_ENV} is exported by nix/checks.nix and nix/devshell.nix")
+        });
+        let dir = tempfile::Builder::new()
+            .prefix("test-nats-")
+            .tempdir()
+            .expect("a fresh directory for the server");
+        std::fs::create_dir(dir.path().join(PORTS_DIR)).expect("the ports file directory");
+        let child = Command::new(&bin)
+            .args(["-a", "127.0.0.1", "-p", "-1", "-js", "-sd"])
+            .arg(dir.path().join("store"))
+            .arg("--ports_file_dir")
+            .arg(dir.path().join(PORTS_DIR))
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .spawn()
+            .unwrap_or_else(|e| panic!("starting {bin}: {e}"));
+        Self { child, dir }
+    }
+
+    /// A client connected to the server, once it accepts connections.
+    pub async fn client(&self) -> async_nats::Client {
+        for _ in 0..100 {
+            if let Some(url) = announced_url(&self.dir.path().join(PORTS_DIR))
+                && let Ok(client) = async_nats::connect(&url).await
+            {
+                return client;
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        panic!(
+            "nats-server in {} never accepted a connection",
+            self.dir.path().display()
+        );
+    }
+}
+
+/// The client URL in the ports file the server writes once it listens.
+fn announced_url(dir: &Path) -> Option<String> {
+    let ports = std::fs::read_dir(dir).ok()?.next()?.ok()?;
+    let ports: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(ports.path()).ok()?).ok()?;
+    ports["nats"][0].as_str().map(str::to_owned)
+}
+
+impl Drop for TestNats {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}