From f0a80c98126f7cc8c34a44b67d8e476b0faf03cf Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 24 Oct 2026 23:12:35 +0200 Subject: [PATCH] notify router: bound the deliveries log append, note doubled retry rows --- Cargo.lock | 1 + docs/swarm/notify-router.md | 5 +- .../src/pages/notifications/Deliveries.tsx | 15 ++- swarm-notify-router/Cargo.toml | 1 + swarm-notify-router/src/log.rs | 120 +++++++++++++++++- 5 files changed, 129 insertions(+), 13 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e32c3c40..7ed609cb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5057,6 +5057,7 @@ dependencies = [ "tempfile", "tokio", "tracing", + "tracing-subscriber", ] [[package]] diff --git a/docs/swarm/notify-router.md b/docs/swarm/notify-router.md index a06c134b..f4ec6f71 100644 --- a/docs/swarm/notify-router.md +++ b/docs/swarm/notify-router.md @@ -188,8 +188,9 @@ controller, whether it went through or not. A row has: A review the controller skipped (`skipped: `) is `ok`: nothing was wrong. An event the router retries after a failed step writes its rows again, -so a retry shows twice. When the router fails to write a row, it logs the failure and the delivery -stands. +so a retry shows twice, and the page says so under the table. A row that +takes longer than 2 seconds to write counts as failed: the router logs a +warning and the delivery goes on. **The dry run.** The router runs a flows document over the newest 50 events of the `SWARM_EVENTS` stream and returns, for each flow the document turns on, diff --git a/frontend/packages/swarm-ui/src/pages/notifications/Deliveries.tsx b/frontend/packages/swarm-ui/src/pages/notifications/Deliveries.tsx index e4834d50..8bb2dd11 100644 --- a/frontend/packages/swarm-ui/src/pages/notifications/Deliveries.tsx +++ b/frontend/packages/swarm-ui/src/pages/notifications/Deliveries.tsx @@ -51,11 +51,14 @@ export function Deliveries({ rows }: { rows: Delivery[] }) { if (rows.length === 0) return

no deliveries yet

; const placed = rows.map((d, place) => ({ ...d, place })); return ( - String(d.place)} - storageKey="swarm-ui:notifications:deliveries-filters" - /> + <> +
String(d.place)} + storageKey="swarm-ui:notifications:deliveries-filters" + /> +

a retried event writes its rows again, so a retry shows twice

+ ); } diff --git a/swarm-notify-router/Cargo.toml b/swarm-notify-router/Cargo.toml index 684b20ef..ea67415e 100644 --- a/swarm-notify-router/Cargo.toml +++ b/swarm-notify-router/Cargo.toml @@ -33,6 +33,7 @@ tracing.workspace = true # A fresh directory per test `nats-server`, for its store and ports file. [dev-dependencies] tempfile = "3" +tracing-subscriber.workspace = true [lints] workspace = true diff --git a/swarm-notify-router/src/log.rs b/swarm-notify-router/src/log.rs index 7b846934..1675c47c 100644 --- a/swarm-notify-router/src/log.rs +++ b/swarm-notify-router/src/log.rs @@ -5,7 +5,7 @@ //! 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 std::time::{Duration, SystemTime, UNIX_EPOCH}; use anyhow::{Context as _, Result}; use async_nats::jetstream::stream::{Config, DiscardPolicy, Stream}; @@ -13,6 +13,9 @@ 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 long an append may take before it is given up on. +const APPEND_WITHIN: Duration = Duration::from_secs(2); + /// How many messages are fetched at once. const FETCH_AT_ONCE: usize = 16; @@ -77,8 +80,14 @@ impl Cause { } } -/// Append `row`; a failure is logged and dropped. +/// Append `row`; a failure, or an append that takes longer than +/// `APPEND_WITHIN`, is logged and dropped. It is awaited in place, so rows +/// keep the order of the deliveries. pub async fn append(js: &jetstream::Context, row: &Row) { + append_within(js, row, APPEND_WITHIN).await; +} + +async fn append_within(js: &jetstream::Context, row: &Row, within: Duration) { let body = serde_json::to_vec(row).expect("a row encodes"); let sent = async { js.publish(LOG_SUBJECT, body.into()) @@ -87,9 +96,12 @@ pub async fn append(js: &jetstream::Context, row: &Row) { .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"); - } + let failure = match tokio::time::timeout(within, sent).await { + Ok(Ok(_)) => return, + Ok(Err(e)) => format!("{e:#}"), + Err(_) => format!("no ack within {within:?}"), + }; + tracing::warn!(error = failure, flow = %row.flow, "notify router: could not log a delivery"); } /// The payloads of the newest `limit` messages of `stream`, newest first, @@ -132,3 +144,101 @@ pub async fn recent(js: &jetstream::Context, limit: usize) -> Result> { .filter_map(|(_, body)| serde_json::from_slice(&body).ok()) .collect()) } + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + use std::time::Instant; + + use tracing_subscriber::layer::SubscriberExt as _; + + use super::*; + use crate::test_nats::TestNats; + + /// The level and message of every event. + #[derive(Clone, Default)] + struct Messages(Arc>>); + + impl tracing_subscriber::Layer for Messages { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + let mut message = String::new(); + event.record( + &mut |field: &tracing::field::Field, value: &dyn std::fmt::Debug| { + if field.name() == "message" { + message = format!("{value:?}"); + } + }, + ); + self.0 + .lock() + .expect("unpoisoned") + .push((*event.metadata().level(), message)); + } + } + + /// The messages logged at warn or above. + fn warnings(messages: &Messages) -> Vec { + messages + .0 + .lock() + .expect("unpoisoned") + .iter() + .filter(|(level, _)| *level <= tracing::Level::WARN) + .map(|(_, m)| m.clone()) + .collect() + } + + fn a_row() -> Row { + Cause { + event: "issue_assign".to_owned(), + repo: "o/r".to_owned(), + number: 1, + } + .row("f", "n", Kind::Todo, "agent", Ok("upsert k".to_owned())) + } + + /// A subscriber on the log subject with no stream behind it: the publish + /// goes out and no ack ever comes back. + #[tokio::test] + async fn an_append_nothing_acks_gives_up_and_warns() { + let nats = TestNats::start(); + let client = nats.client().await; + let js = jetstream::new(client.clone()); + let _silent = client.subscribe(LOG_SUBJECT).await.expect("subscribed"); + let messages = Messages::default(); + let _guard = + tracing::subscriber::set_default(tracing_subscriber::registry().with(messages.clone())); + + let started = Instant::now(); + append_within(&js, &a_row(), Duration::from_millis(300)).await; + assert!( + started.elapsed() < Duration::from_secs(2), + "{:?}", + started.elapsed() + ); + assert_eq!( + warnings(&messages), + ["notify router: could not log a delivery"] + ); + } + + /// Control: with the stream in place the same call acks and logs nothing. + #[tokio::test] + async fn an_append_the_stream_acks_logs_nothing() { + let nats = TestNats::start(); + let client = nats.client().await; + let js = jetstream::new(client); + open_or_create(&js).await.expect("the log stream"); + let messages = Messages::default(); + let _guard = + tracing::subscriber::set_default(tracing_subscriber::registry().with(messages.clone())); + + append_within(&js, &a_row(), Duration::from_secs(5)).await; + assert!(warnings(&messages).is_empty()); + assert_eq!(recent(&js, 10).await.expect("rows").len(), 1); + } +}