feat(log-ingestor): Add retry with exponential backoff for transient ingestion failures

components/clp-rust-utils/src/job_config/ingestion.rs

@@ -69,6 +69,10 @@ pub mod s3 {
         /// Per-job ingestion buffer config.
         #[serde(default)]
         pub buffer_config: BufferConfig,
+
+        /// Configuration for retry behavior on transient ingestion failures.
+        #[serde(default)]
+        pub retry_config: RetryConfig,
     }
 
     /// Configuration for a SQS listener job.
@@ -164,6 +168,39 @@ pub mod s3 {
         pub start_after: Option<NonEmptyString>,
     }
 
+    /// Configuration for retry behavior on transient failures.
+    #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+    #[serde(default)]
+    pub struct RetryConfig {
+        /// Maximum number of consecutive transient failures before the job is marked as failed.
+        /// Defaults to 10.
+        #[schema(minimum = 1)]
+        pub max_consecutive_failures: u32,
+
+        /// Initial backoff duration in milliseconds before the first retry.
+        /// Defaults to 1000 (1 second).
+        pub initial_backoff_ms: u64,
+
+        /// Maximum backoff duration in milliseconds.
+        /// Defaults to 60000 (60 seconds).
+        pub max_backoff_ms: u64,
+
+        /// Backoff multiplier applied after each consecutive failure.
+        /// Defaults to 2.
+        pub backoff_multiplier: u32,
+    }
+
+    impl Default for RetryConfig {
+        fn default() -> Self {
+            Self {
+                max_consecutive_failures: 10,
+                initial_backoff_ms: 1_000,
+                max_backoff_ms: 60_000,
+                backoff_multiplier: 2,
+            }
+        }
+    }
+
     /// Configuration for buffer behavior.
     #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
     #[serde(default)]
@@ -230,7 +267,28 @@ pub type JobId = u64;
 
 #[cfg(test)]
 mod tests {
-    use super::s3::BufferConfig;
+    use super::s3::{BufferConfig, RetryConfig};
+
+    #[test]
+    fn test_retry_config_defaults() -> Result<(), serde_json::Error> {
+        let config: RetryConfig = serde_json::from_str("{}")?;
+        assert_eq!(config.max_consecutive_failures, 10);
+        assert_eq!(config.initial_backoff_ms, 1_000);
+        assert_eq!(config.max_backoff_ms, 60_000);
+        assert_eq!(config.backoff_multiplier, 2);
+        Ok(())
+    }
+
+    #[test]
+    fn test_retry_config_partial_override() -> Result<(), serde_json::Error> {
+        let config: RetryConfig =
+            serde_json::from_str(r#"{"max_consecutive_failures": 5}"#)?;
+        assert_eq!(config.max_consecutive_failures, 5);
+        assert_eq!(config.initial_backoff_ms, 1_000);
+        assert_eq!(config.max_backoff_ms, 60_000);
+        assert_eq!(config.backoff_multiplier, 2);
+        Ok(())
+    }
 
     #[test]
     fn test_buffer_config_partial_override() -> Result<(), serde_json::Error> {

components/log-ingestor/src/ingestion_job.rs

@@ -1,3 +1,4 @@
+mod retry;
 mod s3_scanner;
 mod sqs_listener;
 mod state;

components/log-ingestor/src/ingestion_job/retry.rs

@@ -0,0 +1,175 @@
+use std::time::Duration;
+
+use clp_rust_utils::job_config::ingestion::s3::RetryConfig;
+use tokio::select;
+use tokio_util::sync::CancellationToken;
+
+/// Tracks consecutive failure state and computes exponential backoff durations.
+pub struct RetryState {
+    config: RetryConfig,
+    consecutive_failures: u32,
+}
+
+impl RetryState {
+    pub const fn new(config: RetryConfig) -> Self {
+        Self {
+            config,
+            consecutive_failures: 0,
+        }
+    }
+
+    /// Records a successful operation, resetting the consecutive failure counter.
+    pub const fn record_success(&mut self) {
+        self.consecutive_failures = 0;
+    }
+
+    /// Records a failure. Returns `true` if retry is allowed (under the max),
+    /// `false` if the max consecutive failures has been reached.
+    pub const fn record_failure(&mut self) -> bool {
+        self.consecutive_failures = self.consecutive_failures.saturating_add(1);
+        self.consecutive_failures < self.config.max_consecutive_failures
+    }
+
+    /// Returns the current backoff duration based on the number of consecutive failures.
+    pub fn current_backoff(&self) -> Duration {
+        let exponent = self.consecutive_failures.saturating_sub(1);
+        let backoff_ms = self
+            .config
+            .initial_backoff_ms
+            .saturating_mul(u64::from(self.config.backoff_multiplier).saturating_pow(exponent))
+            .min(self.config.max_backoff_ms);
+        Duration::from_millis(backoff_ms)
+    }
+
+    /// Returns the current number of consecutive failures.
+    pub const fn consecutive_failures(&self) -> u32 {
+        self.consecutive_failures
+    }
+
+    /// Sleeps for the current backoff duration, respecting cancellation.
+    /// Returns `true` if cancelled during sleep, `false` if sleep completed normally.
+    pub async fn backoff_sleep(&self, cancel_token: &CancellationToken) -> bool {
+        let backoff = self.current_backoff();
+        select! {
+            () = cancel_token.cancelled() => true,
+            () = tokio::time::sleep(backoff) => false,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn default_config() -> RetryConfig {
+        RetryConfig::default()
+    }
+
+    #[test]
+    fn test_record_success_resets_counter() {
+        let mut state = RetryState::new(default_config());
+        state.record_failure();
+        state.record_failure();
+        assert_eq!(state.consecutive_failures(), 2);
+        state.record_success();
+        assert_eq!(state.consecutive_failures(), 0);
+    }
+
+    #[test]
+    fn test_record_failure_returns_true_until_max() {
+        let config = RetryConfig {
+            max_consecutive_failures: 3,
+            ..Default::default()
+        };
+        let mut state = RetryState::new(config);
+        assert!(state.record_failure()); // 1 < 3
+        assert!(state.record_failure()); // 2 < 3
+        assert!(!state.record_failure()); // 3 == 3, exhausted
+    }
+
+    #[test]
+    fn test_backoff_exponential_growth() {
+        let config = RetryConfig {
+            initial_backoff_ms: 1000,
+            backoff_multiplier: 2,
+            max_backoff_ms: 60_000,
+            max_consecutive_failures: 10,
+        };
+        let mut state = RetryState::new(config);
+
+        state.record_failure(); // consecutive_failures = 1
+        assert_eq!(state.current_backoff(), Duration::from_millis(1000)); // 1000 * 2^0
+
+        state.record_failure(); // consecutive_failures = 2
+        assert_eq!(state.current_backoff(), Duration::from_millis(2000)); // 1000 * 2^1
+
+        state.record_failure(); // consecutive_failures = 3
+        assert_eq!(state.current_backoff(), Duration::from_millis(4000)); // 1000 * 2^2
+
+        state.record_failure(); // consecutive_failures = 4
+        assert_eq!(state.current_backoff(), Duration::from_millis(8000)); // 1000 * 2^3
+    }
+
+    #[test]
+    fn test_backoff_caps_at_max() {
+        let config = RetryConfig {
+            initial_backoff_ms: 1000,
+            backoff_multiplier: 2,
+            max_backoff_ms: 5000,
+            max_consecutive_failures: 20,
+        };
+        let mut state = RetryState::new(config);
+
+        for _ in 0..10 {
+            state.record_failure();
+        }
+        assert_eq!(state.current_backoff(), Duration::from_millis(5000));
+    }
+
+    #[test]
+    fn test_backoff_zero_failures_returns_initial() {
+        let state = RetryState::new(default_config());
+        // With 0 failures, exponent would be saturating_sub(1) = 0 on underflow,
+        // but this state shouldn't normally be queried before a failure.
+        // Still, verify it doesn't panic.
+        let _ = state.current_backoff();
+    }
+
+    #[tokio::test]
+    async fn test_backoff_sleep_completes_normally() {
+        let config = RetryConfig {
+            initial_backoff_ms: 10,
+            max_backoff_ms: 100,
+            backoff_multiplier: 2,
+            max_consecutive_failures: 5,
+        };
+        let mut state = RetryState::new(config);
+        state.record_failure();
+
+        let cancel_token = CancellationToken::new();
+        let cancelled = state.backoff_sleep(&cancel_token).await;
+        assert!(!cancelled);
+    }
+
+    #[tokio::test]
+    async fn test_backoff_sleep_respects_cancellation() {
+        let config = RetryConfig {
+            initial_backoff_ms: 60_000,
+            max_backoff_ms: 60_000,
+            backoff_multiplier: 2,
+            max_consecutive_failures: 5,
+        };
+        let mut state = RetryState::new(config);
+        state.record_failure();
+
+        let cancel_token = CancellationToken::new();
+        let cancel_token_clone = cancel_token.clone();
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+            cancel_token_clone.cancel();
+        });
+
+        let cancelled = state.backoff_sleep(&cancel_token).await;
+        assert!(cancelled);
+    }
+}

components/log-ingestor/src/ingestion_job/s3_scanner.rs

@@ -7,6 +7,7 @@ use non_empty_string::NonEmptyString;
 use tokio::select;
 use tokio_util::sync::CancellationToken;
 
+use super::retry::RetryState;
 use crate::{
     aws_client_manager::AwsClientManagerType,
     ingestion_job::{IngestionJobId, IngestionJobState, S3ScannerState},
@@ -37,7 +38,8 @@ impl<S3ClientManager: AwsClientManagerType<Client>, State: IngestionJobState + S
     /// Runs the S3 scanner task to scan the given bucket.
     ///
     /// This is a wrapper of [`Self::scan_once`] that supports cancellation via the provided
-    /// cancellation token.
+    /// cancellation token. Transient failures are retried with exponential backoff up to the
+    /// configured maximum consecutive failures.
     ///
     /// # Returns
     ///
@@ -47,24 +49,45 @@ impl<S3ClientManager: AwsClientManagerType<Client>, State: IngestionJobState + S
     ///
     /// Returns an error if:
     ///
-    /// * Forwards [`Self::scan_once`]'s return values on failure.
+    /// * The maximum number of consecutive transient failures is exceeded.
     pub async fn run(mut self, cancel_token: CancellationToken) -> Result<()> {
+        let mut retry_state = RetryState::new(self.config.base.retry_config.clone());
+
         loop {
             select! {
-                // Cancellation requested.
                 () = cancel_token.cancelled() => {
                     return Ok(());
                 }
 
-                // Scanner execution
                 is_truncated_result = self.scan_once() => {
-                    if is_truncated_result? {
-                        // The results are truncated. Keep going until all objects are listed.
-                        // Ideally, we can use the continuation token to continue listing objects,
-                        // but since we may refresh the client in the next scan cycle, we will use
-                        // `start_after` to send a new request to resume the scanning progress for
-                        // simplicity.
-                        continue;
+                    match is_truncated_result {
+                        Ok(is_truncated) => {
+                            retry_state.record_success();
+                            if is_truncated {
+                                // The results are truncated. Keep going until all objects are
+                                // listed. We use `start_after` to send a new request to resume
+                                // the scanning progress.
+                                continue;
+                            }
+                        }
+                        Err(err) => {
+                            if !retry_state.record_failure() {
+                                return Err(err.context(format!(
+                                    "Exceeded maximum consecutive failures ({})",
+                                    retry_state.consecutive_failures()
+                                )));
+                            }
+                            tracing::warn!(
+                                error = ?err,
+                                attempt = retry_state.consecutive_failures(),
+                                max_attempts = self.config.base.retry_config.max_consecutive_failures,
+                                "S3 scanner transient failure, retrying after backoff."
+                            );
+                            if retry_state.backoff_sleep(&cancel_token).await {
+                                return Ok(());
+                            }
+                            continue;
+                        }
                     }
                 }
             }