fix(channels): retry only transient delivery errors

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<MessageBus>) -> 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);
+    }
 }

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<BusError> for ChannelError {
     fn from(e: BusError) -> Self {
         ChannelError::BusError(e.to_string())