From d0568d7933ca1102bd2197a5809ffb3345236baf Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 01:53:31 +0100 Subject: [PATCH] relay: sink failure defers + pauses w/ backoff, no attempt spent --- crates/bootstrap/src/outbox_relay.rs | 64 ++++++++++++++++++--- crates/bootstrap/tests/outbox_relay_test.rs | 64 +++++++++++++++++++-- 2 files changed, 114 insertions(+), 14 deletions(-) diff --git a/crates/bootstrap/src/outbox_relay.rs b/crates/bootstrap/src/outbox_relay.rs index 86d25152..5d3f99cd 100644 --- a/crates/bootstrap/src/outbox_relay.rs +++ b/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, sink: Arc, + backoff: fn(u32) -> Duration, } impl OutboxRelay { pub fn new(source: Arc, sink: Arc) -> 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) { 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 { 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}"); - if let Err(e) = envelope.nack().await { - tracing::error!("outbox relay: could not nack: {e}"); + *sink_failures = sink_failures.saturating_add(1); + let wait = backoff(*sink_failures); + 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) } } } diff --git a/crates/bootstrap/tests/outbox_relay_test.rs b/crates/bootstrap/tests/outbox_relay_test.rs index e424af6a..857a6ab5 100644 --- a/crates/bootstrap/tests/outbox_relay_test.rs +++ b/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, 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, SqliteEventQueue) { + outbox_with_attempts(config::EventsConfig::default().queue_max_attempts).await +} + +async fn outbox_with_attempts( + max_attempts: i32, +) -> (sqlx::SqlitePool, Arc, 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) { } 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 = 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 = sqlx::query_scalar("SELECT last_error FROM event_queue") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(last_error.as_deref(), Some("deferred")); } #[tokio::test]