diff --git a/components/clp-rust-utils/src/job_config/ingestion.rs b/components/clp-rust-utils/src/job_config/ingestion.rs index 015bbb2bdc..1eaaba53bd 100644 --- a/components/clp-rust-utils/src/job_config/ingestion.rs +++ b/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, } + /// 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> { diff --git a/components/log-ingestor/src/ingestion_job.rs b/components/log-ingestor/src/ingestion_job.rs index 94c3b70a7d..edfb4a4734 100644 --- a/components/log-ingestor/src/ingestion_job.rs +++ b/components/log-ingestor/src/ingestion_job.rs @@ -1,3 +1,4 @@ +mod retry; mod s3_scanner; mod sqs_listener; mod state; diff --git a/components/log-ingestor/src/ingestion_job/retry.rs b/components/log-ingestor/src/ingestion_job/retry.rs new file mode 100644 index 0000000000..08d497518a --- /dev/null +++ b/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); + } +} diff --git a/components/log-ingestor/src/ingestion_job/s3_scanner.rs b/components/log-ingestor/src/ingestion_job/s3_scanner.rs index 61232beedd..872f2e35bf 100644 --- a/components/log-ingestor/src/ingestion_job/s3_scanner.rs +++ b/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, 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, 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; + } } } } diff --git a/components/log-ingestor/src/ingestion_job/sqs_listener.rs b/components/log-ingestor/src/ingestion_job/sqs_listener.rs index 1a8749236c..84be76dec1 100644 --- a/components/log-ingestor/src/ingestion_job/sqs_listener.rs +++ b/components/log-ingestor/src/ingestion_job/sqs_listener.rs @@ -14,6 +14,7 @@ use clp_rust_utils::{ use tokio::select; use tokio_util::sync::CancellationToken; +use super::retry::RetryState; use crate::{ aws_client_manager::AwsClientManagerType, ingestion_job::{IngestionJobId, IngestionJobState, SqsListenerState}, @@ -43,7 +44,8 @@ impl, State: IngestionJobState + Task { /// Runs the SQS listener task to listen to SQS messages and extract S3 object metadata. The - /// extracted metadata is ingested into the provided state. + /// extracted metadata is ingested into the provided state. Transient failures are retried with + /// exponential backoff up to the configured maximum consecutive failures. /// /// # Returns /// @@ -53,36 +55,66 @@ impl, State: IngestionJobState + /// /// Returns an error if: /// - /// * Forwards [`AwsClientManagerType::get`]'s return values on failure. - /// * Forwards [`Self::process_sqs_response`]'s return values on failure. - /// * Forwards - /// [`aws_sdk_sqs::operation::receive_message::builders::ReceiveMessageFluentBuilder::send`]'s - /// return values on failure. + /// * The maximum number of consecutive transient failures is exceeded. pub async fn run(self, cancel_token: CancellationToken) -> Result<()> { - const MAX_NUM_MESSAGES_TO_FETCH: i32 = 10; - const MAX_WAIT_TIME_SEC: u16 = 20; - let config = self.config.get(); - let wait_time_sec = i32::from(min(config.wait_time_sec, MAX_WAIT_TIME_SEC)); + let mut retry_state = + RetryState::new(self.config.get().base.retry_config.clone()); loop { select! { - // Cancellation requested. () = cancel_token.cancelled() => { return Ok(()); } - // Listen to SQS messages. - result = self.sqs_client_manager.get().await? - .receive_message() - .queue_url(config.queue_url.as_str()) - .max_number_of_messages(MAX_NUM_MESSAGES_TO_FETCH) - .wait_time_seconds(wait_time_sec).send() => { - self.process_sqs_response(result?).await?; + result = self.receive_and_process_once() => { + match result { + Ok(()) => { + retry_state.record_success(); + } + Err(err) => { + if !retry_state.record_failure() { + return Err(err.context(format!( + "Exceeded maximum consecutive failures ({})", + retry_state.consecutive_failures() + ))); + } + tracing::warn!( + error = ?err, + job_id = ?self.job_id, + task_id = ?self.id, + attempt = retry_state.consecutive_failures(), + max_attempts = self.config.get().base.retry_config.max_consecutive_failures, + "SQS listener transient failure, retrying after backoff." + ); + if retry_state.backoff_sleep(&cancel_token).await { + return Ok(()); + } + } + } } } } } + /// Receives messages from SQS and processes them in a single iteration. + async fn receive_and_process_once(&self) -> Result<()> { + const MAX_NUM_MESSAGES_TO_FETCH: i32 = 10; + const MAX_WAIT_TIME_SEC: u16 = 20; + let config = self.config.get(); + let wait_time_sec = i32::from(min(config.wait_time_sec, MAX_WAIT_TIME_SEC)); + + let client = self.sqs_client_manager.get().await?; + let response = client + .receive_message() + .queue_url(config.queue_url.as_str()) + .max_number_of_messages(MAX_NUM_MESSAGES_TO_FETCH) + .wait_time_seconds(wait_time_sec) + .send() + .await?; + self.process_sqs_response(response).await?; + Ok(()) + } + /// Processes the SQS response to extract S3 object metadata and ingests it into the provided /// state. /// diff --git a/components/log-ingestor/tests/test_ingestion_job.rs b/components/log-ingestor/tests/test_ingestion_job.rs index fb81895261..19d70fe576 100644 --- a/components/log-ingestor/tests/test_ingestion_job.rs +++ b/components/log-ingestor/tests/test_ingestion_job.rs @@ -10,6 +10,7 @@ use clp_rust_utils::{ job_config::ingestion::s3::{ BaseConfig, BufferConfig, + RetryConfig, S3ScannerConfig, SqsListenerConfig, ValidatedSqsListenerConfig, @@ -330,6 +331,7 @@ async fn test_sqs_listener() -> Result<()> { timestamp_key: None, unstructured: false, buffer_config: BufferConfig::default(), + retry_config: RetryConfig::default(), }, }, ) @@ -372,6 +374,7 @@ async fn test_s3_scanner() -> Result<()> { timestamp_key: None, unstructured: false, buffer_config: BufferConfig::default(), + retry_config: RetryConfig::default(), }, scanning_interval_sec: 1, start_after: None,