From dfa5e1a4dbc8083ca788e95d8929bf6edda1507e Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 01:52:28 +0100 Subject: [PATCH] nats nak w/ shared retry_backoff, exhausted log; no 'will retry' lie --- crates/adapters/nats/src/consumer.rs | 59 ++++++++++++++++++---- crates/adapters/nats/src/tests/consumer.rs | 15 ++++++ crates/application/src/worker.rs | 7 ++- crates/domain/src/events.rs | 8 +++ crates/domain/tests/events/events_test.rs | 10 ++++ 5 files changed, 88 insertions(+), 11 deletions(-) diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index 1879beed..261e6a06 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -5,7 +5,7 @@ use async_nats::{ use async_trait::async_trait; use domain::{ errors::DomainError, - events::{AckHandle, DomainEvent, EventEnvelope, EventId}, + events::{AckHandle, DomainEvent, EventEnvelope, EventId, retry_backoff}, ports::EventConsumer, }; use futures::{ @@ -21,6 +21,23 @@ use crate::{config::NatsConfig, payload::NatsEventMessage, subject::consumer_sub struct NatsJetStreamAckHandle { message: async_nats::jetstream::Message, + event_id: Option, + max_deliver: i64, +} + +/// How to nak a failed delivery: the same backoff schedule as the SQLite +/// queue, keyed by JetStream's delivery count (1 on first delivery). +struct NakPlan { + delay: std::time::Duration, + /// This was the last delivery `max_deliver` allows. + exhausted: bool, +} + +fn nak_plan(delivered: i64, max_deliver: i64) -> NakPlan { + NakPlan { + delay: retry_backoff(u32::try_from(delivered.max(0)).unwrap_or(u32::MAX)), + exhausted: max_deliver > 0 && delivered >= max_deliver, + } } #[async_trait] @@ -38,13 +55,29 @@ impl AckHandle for NatsJetStreamAckHandle { } async fn nack(&self) -> Result<(), DomainError> { - tracing::debug!( - "negatively acknowledging message with sequence {}", - self.message.info().unwrap().stream_sequence - ); + let info = self + .message + .info() + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + let plan = nak_plan(info.delivered, self.max_deliver); + if plan.exhausted { + tracing::error!( + event_id = ?self.event_id.map(|id| id.value()), + sequence = info.stream_sequence, + delivered = info.delivered, + "event exhausted max deliveries; it remains in the stream — see JetStream max-deliveries advisory" + ); + } else { + tracing::debug!( + sequence = info.stream_sequence, + delivered = info.delivered, + "nak with backoff {:?}", + plan.delay + ); + } self.message - .ack_with(AckKind::Nak(None)) + .ack_with(AckKind::Nak(Some(plan.delay))) .await .map_err(|e| DomainError::InfrastructureError(e.to_string())) } @@ -94,12 +127,19 @@ fn decode_message(bytes: &[u8]) -> Result<(Option, DomainEvent), Domain Ok((id, DomainEvent::try_from(message.payload)?)) } -fn decode_js(msg: async_nats::jetstream::Message) -> Result { +fn decode_js( + msg: async_nats::jetstream::Message, + max_deliver: i64, +) -> Result { let (id, event) = decode_message(&msg.payload)?; Ok(EventEnvelope::new( id, event, - Box::new(NatsJetStreamAckHandle { message: msg }), + Box::new(NatsJetStreamAckHandle { + message: msg, + event_id: id, + max_deliver, + }), )) } @@ -227,6 +267,7 @@ impl NatsJetStreamConsumer { let (tx, rx) = mpsc::channel(128); + let max_deliver = cfg.max_deliver; tokio::spawn(async move { loop { let mut messages = match consumer.messages().await { @@ -243,7 +284,7 @@ impl NatsJetStreamConsumer { while let Some(result) = messages.next().await { let envelope = result .map_err(|e| DomainError::InfrastructureError(e.to_string())) - .and_then(decode_js); + .and_then(|m| decode_js(m, max_deliver)); if tx.send(envelope).await.is_err() { tracing::info!("consumer channel closed, stopping message processing"); diff --git a/crates/adapters/nats/src/tests/consumer.rs b/crates/adapters/nats/src/tests/consumer.rs index 73899f58..098e3b1c 100644 --- a/crates/adapters/nats/src/tests/consumer.rs +++ b/crates/adapters/nats/src/tests/consumer.rs @@ -91,3 +91,18 @@ fn consumer_max_deliver_drift_is_updated() { }; assert!(consumer_config_update(&same, 5).is_none()); } + +#[test] +fn nak_delay_follows_the_shared_backoff_by_delivery_count() { + let secs = |d| nak_plan(d, 5).delay.as_secs(); + assert_eq!((secs(1), secs(2), secs(3), secs(4)), (10, 20, 40, 80)); + assert_eq!(nak_plan(50, 100).delay.as_secs(), 300); +} + +#[test] +fn only_the_final_delivery_is_exhausted() { + assert!(!nak_plan(4, 5).exhausted); + assert!(nak_plan(5, 5).exhausted); + assert!(nak_plan(6, 5).exhausted); + assert!(!nak_plan(1_000, -1).exhausted, "-1 means unlimited"); +} diff --git a/crates/application/src/worker.rs b/crates/application/src/worker.rs index f01d64b9..9288c2e8 100644 --- a/crates/application/src/worker.rs +++ b/crates/application/src/worker.rs @@ -119,7 +119,7 @@ async fn dispatch( } Err(e) => { // Running anyway could repeat a side effect; retry the delivery instead. - tracing::warn!(handler = name, "ledger claim failed, will retry: {e}"); + tracing::warn!(handler = name, "ledger claim failed, nacking delivery: {e}"); any_transient = true; continue; } @@ -146,7 +146,10 @@ async fn dispatch( tracing::warn!(handler = name, "releasing ledger claim failed: {le}"); } if e.is_transient() { - tracing::warn!(handler = name, "transient handler error, will retry: {e}"); + tracing::warn!( + handler = name, + "transient handler error, nacking delivery (the backend logs if attempts are exhausted): {e}" + ); any_transient = true; } else { tracing::warn!( diff --git a/crates/domain/src/events.rs b/crates/domain/src/events.rs index 3543cf9e..25218ff7 100644 --- a/crates/domain/src/events.rs +++ b/crates/domain/src/events.rs @@ -155,6 +155,14 @@ pub trait AckHandle: Send + Sync { async fn defer(&self, delay: std::time::Duration) -> Result<(), DomainError>; } +/// Wait before retrying an event whose `attempt`-th delivery just failed: +/// 5s doubling per attempt, capped at 5 minutes. Every backend (SQLite queue, +/// JetStream nak, outbox relay) uses this one schedule. +pub fn retry_backoff(attempt: u32) -> std::time::Duration { + let secs = 5u64 * (1u64 << attempt.min(6)); + std::time::Duration::from_secs(secs.min(300)) +} + /// Stable identity of one queued event; the key for de-duplicating deliveries. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub struct EventId(uuid::Uuid); diff --git a/crates/domain/tests/events/events_test.rs b/crates/domain/tests/events/events_test.rs index bde7014d..482f2a8f 100644 --- a/crates/domain/tests/events/events_test.rs +++ b/crates/domain/tests/events/events_test.rs @@ -53,3 +53,13 @@ fn follow_requested_with_handle() { } )); } + +#[test] +fn retry_backoff_doubles_from_ten_seconds_and_caps_at_five_minutes() { + use domain::events::retry_backoff; + use std::time::Duration; + let secs: Vec = (1..=8).map(|n| retry_backoff(n).as_secs()).collect(); + assert_eq!(secs, [10, 20, 40, 80, 160, 300, 300, 300]); + assert_eq!(retry_backoff(0), Duration::from_secs(5)); + assert_eq!(retry_backoff(u32::MAX), Duration::from_secs(300)); +}