feat(BA-4127): Add exponential backoff for Redis consumer reconnection (#8387)

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: HyeockJinKim

changes/8387.feature.md

@@ -0,0 +1 @@
+Add exponential backoff for Redis consumer reconnection

src/ai/backend/common/message_queue/redis_queue/consumer.py

@@ -3,7 +3,9 @@
 import asyncio
 import hashlib
 import logging
+import random
 import socket
+import time
 from collections.abc import AsyncGenerator, Iterable
 from dataclasses import dataclass
 from typing import Optional, Self, override
@@ -28,6 +30,24 @@
 _DEFAULT_READ_COUNT = 64
 
 
+@dataclass
+class _BackoffState:
+    """Tracks exponential backoff state for a stream."""
+
+    attempt: int = 0
+    last_error_time: float = 0.0
+
+    def increment(self) -> None:
+        """Increment attempt counter and record error time."""
+        self.attempt += 1
+        self.last_error_time = time.monotonic()
+
+    def reset(self) -> None:
+        """Reset backoff state after successful operation."""
+        self.attempt = 0
+        self.last_error_time = 0.0
+
+
 @dataclass
 class RedisConsumerArgs:
     stream_keys: Iterable[str]
@@ -36,6 +56,9 @@ class RedisConsumerArgs:
     db: int = 0
     autoclaim_idle_timeout: int = _DEFAULT_AUTOCLAIM_IDLE_TIMEOUT
     autoclaim_start_id: Optional[str] = None
+    backoff_initial_delay: float = 0.1  # 100ms first retry
+    backoff_max_delay: float = 30.0  # cap at 30 seconds
+    backoff_max_attempts: Optional[int] = None  # None = infinite retry
 
 
 class RedisConsumer(AbstractConsumer):
@@ -55,6 +78,10 @@ class RedisConsumer(AbstractConsumer):
     _autoclaim_idle_timeout: int
     _closed: bool
     _loop_tasks: list[asyncio.Task]
+    _backoff_initial_delay: float
+    _backoff_max_delay: float
+    _backoff_max_attempts: Optional[int]
+    _backoff_state: dict[str, _BackoffState]
 
     def __init__(
         self,
@@ -83,6 +110,12 @@ def __init__(
         self._autoclaim_idle_timeout = args.autoclaim_idle_timeout
         self._closed = False
 
+        # Backoff configuration
+        self._backoff_initial_delay = args.backoff_initial_delay
+        self._backoff_max_delay = args.backoff_max_delay
+        self._backoff_max_attempts = args.backoff_max_attempts
+        self._backoff_state = {}
+
         start_id = args.autoclaim_start_id or "0-0"
         self._loop_tasks = []
 
@@ -221,6 +254,7 @@ async def _read_messages_loop(self, stream_key: str) -> None:
             while not self._closed:
                 try:
                     await self._read_messages(client, stream_key)
+                    self._reset_backoff(stream_key)
                 except glide.ClosingError:
                     log.info(
                         "Client connection closed, stopping read messages loop for stream {}",
@@ -229,8 +263,10 @@ async def _read_messages_loop(self, stream_key: str) -> None:
                     break
                 except glide.GlideError as e:
                     await self._failover_consumer(stream_key, e)
+                    await self._handle_backoff(stream_key)
                 except Exception as e:
                     log.error("Error while reading messages from stream {}: {}", stream_key, e)
+                    await self._handle_backoff(stream_key)
         finally:
             await client.close()
 
@@ -277,6 +313,7 @@ async def _auto_claim_loop(
                 )
                 if claimed:
                     autoclaim_start_id = next_start_id
+                    self._reset_backoff(stream_key)
                     continue
             except glide.TimeoutError:
                 # If the auto claim times out, we just continue to the next iteration
@@ -288,10 +325,14 @@ async def _auto_claim_loop(
                 break
             except glide.GlideError as e:
                 await self._failover_consumer(stream_key, e)
+                await self._handle_backoff(stream_key)
+                continue
             except Exception as e:
                 log.exception(
                     "Error while auto claiming messages from stream {}: {}", stream_key, e
                 )
+                await self._handle_backoff(stream_key)
+                continue
 
             await asyncio.sleep(_DEFAULT_AUTOCLAIM_INTERVAL / 1000)
 
@@ -342,6 +383,42 @@ async def _retry_message(self, stream_key: str, message: MQMessage) -> None:
             stream_key, self._group_name, message.msg_id, message.payload
         )
 
+    async def _handle_backoff(self, stream_key: str) -> None:
+        """
+        Handle exponential backoff for a stream.
+
+        Increments attempt counter, calculates delay with jitter, and sleeps.
+
+        Args:
+            stream_key: The Redis stream key experiencing connection issues
+        """
+        if stream_key not in self._backoff_state:
+            self._backoff_state[stream_key] = _BackoffState()
+
+        state = self._backoff_state[stream_key]
+        state.increment()
+
+        # Calculate delay with exponential backoff
+        delay = min(
+            self._backoff_initial_delay * (2 ** (state.attempt - 1)),
+            self._backoff_max_delay,
+        )
+
+        # Add jitter (50-100% of calculated delay)
+        actual_delay = delay * (0.5 + random.random() * 0.5)
+
+        await asyncio.sleep(actual_delay)
+
+    def _reset_backoff(self, stream_key: str) -> None:
+        """
+        Reset backoff state for a stream after successful operation.
+
+        Args:
+            stream_key: The Redis stream key that successfully completed operation
+        """
+        if stream_key in self._backoff_state:
+            self._backoff_state[stream_key].reset()
+
     async def _failover_consumer(self, stream_key: str, e: Exception) -> None:
         """
         Handle consumer failover scenarios.