diff --git a/src/bus/dispatcher.rs b/src/bus/dispatcher.rs index afe0d7a..35927c9 100644 --- a/src/bus/dispatcher.rs +++ b/src/bus/dispatcher.rs @@ -153,7 +153,7 @@ impl OutboundDispatcher { let result = tokio::time::timeout(SEND_TIMEOUT, channel.send(msg.clone())).await; match result { Ok(Ok(())) => return Ok(()), - Ok(Err(error)) if attempt < DELAYS.len() - 1 => { + Ok(Err(error)) if attempt < DELAYS.len() - 1 && error.is_transient() => { tracing::warn!(attempt = attempt + 1, delay, error = %error, "Send failed, retrying"); } Ok(Err(error)) => return Err(error), @@ -177,6 +177,7 @@ impl OutboundDispatcher { mod tests { use super::*; use async_trait::async_trait; + use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::{Mutex, Notify}; struct RecordingChannel { @@ -184,6 +185,30 @@ mod tests { notify: Notify, } + struct PermanentFailureChannel { + attempts: AtomicUsize, + } + + #[async_trait] + impl Channel for PermanentFailureChannel { + fn name(&self) -> &str { + "permanent-failure" + } + fn is_running(&self) -> bool { + true + } + async fn start(&self, _bus: Arc) -> Result<(), ChannelError> { + Ok(()) + } + async fn stop(&self) -> Result<(), ChannelError> { + Ok(()) + } + async fn send(&self, _msg: OutboundMessage) -> Result<(), ChannelError> { + self.attempts.fetch_add(1, Ordering::SeqCst); + Err(ChannelError::Other("invalid destination".to_string())) + } + } + #[async_trait] impl Channel for RecordingChannel { fn name(&self) -> &str { @@ -312,4 +337,18 @@ mod tests { task.abort(); supervisor.shutdown(Duration::from_secs(1)).await; } + + #[tokio::test] + async fn permanent_send_failure_is_not_retried() { + let channel = PermanentFailureChannel { + attempts: AtomicUsize::new(0), + }; + + let error = OutboundDispatcher::send_with_retry(&channel, &outbound("invalid", "message")) + .await + .unwrap_err(); + + assert!(matches!(error, ChannelError::Other(_))); + assert_eq!(channel.attempts.load(Ordering::SeqCst), 1); + } } diff --git a/src/channels/base.rs b/src/channels/base.rs index d91afea..73e164b 100644 --- a/src/channels/base.rs +++ b/src/channels/base.rs @@ -26,6 +26,12 @@ impl std::fmt::Display for ChannelError { impl std::error::Error for ChannelError {} +impl ChannelError { + pub fn is_transient(&self) -> bool { + matches!(self, Self::ConnectionError(_) | Self::SendError(_)) + } +} + impl From for ChannelError { fn from(e: BusError) -> Self { ChannelError::BusError(e.to_string())