notify router: bound the deliveries log append, note doubled retry rows

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<Vec<Row>> {
         .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<Mutex<Vec<(tracing::Level, String)>>>);
+
+    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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<String> {
+        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);
+    }
+}