relay: sink failure defers + pauses w/ backoff, no attempt spent

crates/bootstrap/src/outbox_relay.rs

@@ -6,16 +6,21 @@
 //! and dead-letter path. A row is marked done only after the sink reports the
 //! bus accepted it.
 //!
+//! A sink (transport) failure is the bus's fault, not the row's: the row is
+//! deferred without spending an attempt, and the relay pauses with a growing
+//! backoff instead of churning through the rest of the outbox. Attempts, and
+//! so dead-lettering, are only ever spent by consumers on handler failures.
+//!
 //! Delivery is at-least-once. A crash between publish and mark-done leaves the
 //! row `processing`; when its claim lease lapses it is claimed and relayed
 //! again. The bus drops that repeat while it is inside the broker's dedup
 //! window (same `Nats-Msg-Id`); beyond the window, or in core mode, the
 //! consumer ledger makes the handlers skip it.
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use domain::{
-    events::EventEnvelope,
+    events::{EventEnvelope, retry_backoff},
     ports::{EventConsumer, EventRelaySink},
 };
 use futures::StreamExt;
@@ -24,11 +29,22 @@ use tokio::sync::watch;
 pub struct OutboxRelay {
     source: Arc<dyn EventConsumer>,
     sink: Arc<dyn EventRelaySink>,
+    backoff: fn(u32) -> Duration,
 }
 
 impl OutboxRelay {
     pub fn new(source: Arc<dyn EventConsumer>, sink: Arc<dyn EventRelaySink>) -> Self {
-        Self { source, sink }
+        Self {
+            source,
+            sink,
+            backoff: retry_backoff,
+        }
+    }
+
+    /// Overrides the pause/defer schedule, keyed by consecutive sink failures.
+    pub fn with_backoff(mut self, backoff: fn(u32) -> Duration) -> Self {
+        self.backoff = backoff;
+        self
     }
 
     /// Runs until `shutdown` fires or the source ends. A shutdown never
@@ -37,6 +53,7 @@ impl OutboxRelay {
     /// `processing` and come back when their lease lapses.
     pub async fn run(self, mut shutdown: watch::Receiver<bool>) {
         let mut rows = self.source.consume();
+        let mut sink_failures: u32 = 0;
         loop {
             tokio::select! {
                 biased;
@@ -45,7 +62,24 @@ impl OutboxRelay {
                     return;
                 }
                 next = rows.next() => match next {
-                    Some(Ok(envelope)) => relay_envelope(envelope, self.sink.as_ref()).await,
+                    Some(Ok(envelope)) => {
+                        let pause = relay_envelope(
+                            envelope,
+                            self.sink.as_ref(),
+                            &mut sink_failures,
+                            self.backoff,
+                        )
+                        .await;
+                        if let Some(pause) = pause {
+                            tokio::select! {
+                                _ = tokio::time::sleep(pause) => {}
+                                _ = shutdown.changed() => {
+                                    tracing::info!("outbox relay: shutdown signal received, stopping");
+                                    return;
+                                }
+                            }
+                        }
+                    }
                     // The queue already dead-lettered it and logged the row id.
                     Some(Err(e)) => tracing::error!("outbox relay: skipped undecodable row: {e}"),
                     None => {
@@ -58,18 +92,32 @@ impl OutboxRelay {
     }
 }
 
-async fn relay_envelope(envelope: EventEnvelope, sink: &dyn EventRelaySink) {
+/// Returns how long the relay should pause when the sink failed.
+async fn relay_envelope(
+    envelope: EventEnvelope,
+    sink: &dyn EventRelaySink,
+    sink_failures: &mut u32,
+    backoff: fn(u32) -> Duration,
+) -> Option<Duration> {
     match sink.relay(envelope.id(), &envelope.event).await {
         Ok(()) => {
+            *sink_failures = 0;
             if let Err(e) = envelope.ack().await {
                 tracing::error!("outbox relay: published but could not mark done: {e}");
             }
+            None
         }
         Err(e) => {
-            tracing::warn!("outbox relay: publish failed, will retry: {e}");
+            *sink_failures = sink_failures.saturating_add(1);
-            if let Err(e) = envelope.nack().await {
+            let wait = backoff(*sink_failures);
-                tracing::error!("outbox relay: could not nack: {e}");
+            tracing::warn!(
-            }
+                consecutive_failures = *sink_failures,
+                "outbox relay: bus unavailable, deferring row {wait:?} without spending an attempt: {e}"
+            );
+            if let Err(e) = envelope.defer(wait).await {
+                tracing::error!("outbox relay: could not defer: {e}");
+            }
+            Some(wait)
         }
     }
 }

crates/bootstrap/tests/outbox_relay_test.rs

@@ -20,6 +20,8 @@ use uuid::Uuid;
 #[derive(Default)]
 struct FakeSink {
     fail: bool,
+    /// Fail this many calls, then succeed (a bus outage that ends).
+    fail_first: std::sync::atomic::AtomicUsize,
     /// When set, `relay` signals `started` then takes this long.
     slow: Option<Duration>,
     started: Notify,
@@ -34,7 +36,15 @@ impl EventRelaySink for FakeSink {
             self.started.notify_one();
             tokio::time::sleep(delay).await;
         }
-        if self.fail {
+        let outage = self
+            .fail_first
+            .fetch_update(
+                std::sync::atomic::Ordering::SeqCst,
+                std::sync::atomic::Ordering::SeqCst,
+                |n| n.checked_sub(1),
+            )
+            .is_ok();
+        if self.fail || outage {
             Err(DomainError::InfrastructureError("bus down".into()))
         } else {
             Ok(())
@@ -43,10 +53,17 @@ impl EventRelaySink for FakeSink {
 }
 
 async fn outbox() -> (sqlx::SqlitePool, Arc<dyn EventPublisher>, SqliteEventQueue) {
+    outbox_with_attempts(config::EventsConfig::default().queue_max_attempts).await
+}
+
+async fn outbox_with_attempts(
+    max_attempts: i32,
+) -> (sqlx::SqlitePool, Arc<dyn EventPublisher>, SqliteEventQueue) {
     let pool = support::memory_pool().await;
     sqlite::migrate(&pool).await.unwrap();
     let config = config::EventsConfig {
         queue_poll_interval_ms: 10,
+        queue_max_attempts: max_attempts,
         ..Default::default()
     };
     let queue = SqliteEventQueue::create(
@@ -64,7 +81,7 @@ async fn row(pool: &sqlx::SqlitePool) -> (String, i32, Option<String>) {
 }
 
 async fn wait_for(pool: &sqlx::SqlitePool, status: &str) {
-    for _ in 0..200 {
+    for _ in 0..800 {
         if row(pool).await.0 == status {
             return;
         }
@@ -81,7 +98,8 @@ fn start(
     tokio::task::JoinHandle<()>,
 ) {
     let (stop, rx) = tokio::sync::watch::channel(false);
-    let relay = bootstrap::outbox_relay::OutboxRelay::new(Arc::new(queue), sink);
+    let relay = bootstrap::outbox_relay::OutboxRelay::new(Arc::new(queue), sink)
+        .with_backoff(|_| Duration::from_millis(10));
     (stop, tokio::spawn(relay.run(rx)))
 }
 
@@ -114,7 +132,32 @@ async fn relay_marks_rows_done_only_after_publish_succeeds() {
 }
 
 #[tokio::test]
-async fn a_failed_publish_leaves_the_row_pending_with_an_attempt_counted() {
+async fn a_failing_bus_never_spends_the_rows_attempts_and_the_row_relays_on_recovery() {
+    // Budget of 2 attempts, 3 consecutive bus failures: were failures counted
+    // as attempts the row would be dead-lettered before the bus recovers.
+    let (pool, publisher, queue) = outbox_with_attempts(2).await;
+    publisher.publish(&event()).await.unwrap();
+    let sink = Arc::new(FakeSink {
+        fail_first: std::sync::atomic::AtomicUsize::new(3),
+        ..Default::default()
+    });
+
+    let (stop, relay) = start(queue, sink.clone());
+    wait_for(&pool, "done").await;
+    relay.abort();
+    drop(stop);
+
+    let (status, attempts, _) = row(&pool).await;
+    assert_eq!((status.as_str(), attempts), ("done", 0));
+    assert_eq!(
+        sink.seen.lock().unwrap().len(),
+        4,
+        "3 failures then success"
+    );
+}
+
+#[tokio::test]
+async fn a_failed_publish_defers_the_row_without_counting_an_attempt() {
     let (pool, publisher, queue) = outbox().await;
     publisher.publish(&event()).await.unwrap();
     let sink = Arc::new(FakeSink {
@@ -124,7 +167,11 @@ async fn a_failed_publish_leaves_the_row_pending_with_an_attempt_counted() {
 
     let (stop, relay) = start(queue, sink.clone());
     for _ in 0..200 {
-        if row(&pool).await.1 == 1 {
+        let last_error: Option<String> = sqlx::query_scalar("SELECT last_error FROM event_queue")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        if last_error.as_deref() == Some("deferred") {
             break;
         }
         tokio::time::sleep(Duration::from_millis(10)).await;
@@ -133,7 +180,12 @@ async fn a_failed_publish_leaves_the_row_pending_with_an_attempt_counted() {
     drop(stop);
 
     let (status, attempts, _) = row(&pool).await;
-    assert_eq!((status.as_str(), attempts), ("pending", 1));
+    assert_eq!((status.as_str(), attempts), ("pending", 0));
+    let last_error: Option<String> = sqlx::query_scalar("SELECT last_error FROM event_queue")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+    assert_eq!(last_error.as_deref(), Some("deferred"));
 }
 
 #[tokio::test]