Merge pull request #118 from seanwevans/codex/replace-multiplication-with-clamped-calculation

Clamp messaging retry backoff delay

Author: seanwevans

src/an_node.rs

@@ -121,6 +121,10 @@ pub async fn run() -> Result<(), Box<dyn Error>> {
         .unwrap_or_else(|_| "500".into())
         .parse()
         .unwrap_or(500);
+    let max_backoff_ms: u64 = std::env::var("AMQP_RECONNECT_MAX_BACKOFF_MS")
+        .unwrap_or_else(|_| "5000".into())
+        .parse()
+        .unwrap_or(5_000);
 
     let queue_name = "an_task_queue";
     let consumer_tag = "an_consumer";
@@ -132,6 +136,7 @@ pub async fn run() -> Result<(), Box<dyn Error>> {
             consumer_tag,
             max_retries,
             backoff_ms,
+            max_backoff_ms,
         )
         .await?;
 
@@ -212,8 +217,14 @@ mod tests {
     #[cfg(feature = "integration-tests")]
     #[tokio::test]
     async fn test_setup_consumer_workflow() {
-        let (channel, mut consumer) =
-            messaging::connect_with_retries(AMQP_ADDR, "test_queue", "test_consumer", 1, 10)
+        let (channel, mut consumer) = messaging::connect_with_retries(
+            AMQP_ADDR,
+            "test_queue",
+            "test_consumer",
+            1,
+            10,
+            10_000,
+        )
                 .await
                 .expect("setup");
 

src/ki_node.rs

@@ -65,6 +65,10 @@ pub async fn run() -> Result<(), Box<dyn Error>> {
         .unwrap_or_else(|_| "500".into())
         .parse()
         .unwrap_or(500);
+    let max_backoff_ms: u64 = std::env::var("AMQP_RECONNECT_MAX_BACKOFF_MS")
+        .unwrap_or_else(|_| "5000".into())
+        .parse()
+        .unwrap_or(5_000);
 
     let queue_name = "ki_task_queue";
     let consumer_tag = "ki_consumer";
@@ -80,6 +84,7 @@ pub async fn run() -> Result<(), Box<dyn Error>> {
             consumer_tag,
             max_retries,
             backoff_ms,
+            max_backoff_ms,
         )
         .await?;
 
@@ -89,6 +94,7 @@ pub async fn run() -> Result<(), Box<dyn Error>> {
             model_consumer_tag,
             max_retries,
             backoff_ms,
+            max_backoff_ms,
         )
         .await?;
 

src/messaging.rs

@@ -82,6 +82,7 @@ pub async fn connect_with_retries(
     consumer_tag: &str,
     max_retries: u32,
     backoff_ms: u64,
+    max_delay_ms: u64,
 ) -> Result<(Channel, lapin::Consumer), Box<dyn Error>> {
     let mut attempts = 0u32;
     loop {
@@ -105,7 +106,11 @@ pub async fn connect_with_retries(
                     error!("Exceeded maximum reconnection attempts.");
                     return Err(e);
                 }
-                let delay = backoff_ms * 2u64.pow(attempts - 1);
+                let attempt_factor = attempts.saturating_sub(1);
+                let multiplier = 2u64.saturating_pow(attempt_factor);
+                let delay = backoff_ms
+                    .saturating_mul(multiplier)
+                    .min(max_delay_ms);
                 sleep(Duration::from_millis(delay)).await;
             }
         }
@@ -165,14 +170,23 @@ mod tests {
 
     #[tokio::test]
     async fn test_connect_with_retries_failure() {
-        let result = connect_with_retries("amqp://invalid:5672/%2f", "queue", "tag", 3, 10).await;
+        let result = connect_with_retries("amqp://invalid:5672/%2f", "queue", "tag", 3, 10, 20)
+            .await;
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_connect_with_retries_hits_max_delay_without_panic() {
+        let result = connect_with_retries("amqp://invalid:5672/%2f", "queue", "tag", 5, 4, 8)
+            .await;
         assert!(result.is_err());
     }
 
     #[cfg(feature = "integration-tests")]
     #[tokio::test]
     async fn test_connect_with_retries_success() {
-        let res = connect_with_retries(AMQP_ADDR, "test_queue_retry", "test_tag", 1, 10).await;
+        let res =
+            connect_with_retries(AMQP_ADDR, "test_queue_retry", "test_tag", 1, 10, 10_000).await;
         match res {
             Ok((channel, consumer)) => {
                 // drop consumer to close