feat(consumers): less aggressive retries (#7317)

To avoid too many retries at once, add:

1. A bigger initial delay
2. a jitter to all retry attempts

Author: volokluev

rust_snuba/Cargo.lock

@@ -36,6 +36,7 @@ parking_lot = "0.12.1"
 procspawn = { version = "1.0.0", features = ["json"] }
 prost = "0.13"
 pyo3 = { version = "0.24.1", features = ["chrono"] }
+rand = "0.9.2"
 regex = "1.11.1"
 reqwest = { version = "0.11.11", features = ["stream"] }
 schemars = { version = "0.8.16", features = ["uuid1"] }

rust_snuba/Cargo.toml

@@ -46,7 +46,7 @@ fn clickhouse_task_runner(
                 tracing::debug!("performing write");
 
                 let response = client
-                    .send(encoded_rows)
+                    .send(encoded_rows, RetryConfig::default())
                     .await
                     .map_err(RunTaskError::Other)?;
 
@@ -121,6 +121,22 @@ where
     }
 }
 
+pub struct RetryConfig {
+    initial_backoff_ms: f64,
+    max_retries: usize,
+    jitter_factor: f64, // between 0 and 1
+}
+
+impl Default for RetryConfig {
+    fn default() -> Self {
+        Self {
+            initial_backoff_ms: 500.0,
+            max_retries: 4,
+            jitter_factor: 0.2,
+        }
+    }
+}
+
 #[derive(Clone)]
 pub struct ClickhouseClient {
     client: Client,
@@ -163,15 +179,12 @@ impl ClickhouseClient {
         }
     }
 
-    pub async fn send(&self, body: Vec<u8>) -> anyhow::Result<Response> {
-        const MAX_RETRIES: usize = 4;
-        const INITIAL_BACKOFF_MS: u64 = 50;
-
+    pub async fn send(&self, body: Vec<u8>, retry_config: RetryConfig) -> anyhow::Result<Response> {
         // Convert to Bytes once for efficient cloning since sending the request
         // moves the body into the request body.
         let body_bytes = bytes::Bytes::from(body);
 
-        for attempt in 0..=MAX_RETRIES {
+        for attempt in 0..=retry_config.max_retries {
             let res = self
                 .client
                 .post(&self.url)
@@ -192,11 +205,11 @@ impl ClickhouseClient {
                             .await
                             .unwrap_or_else(|_| "unknown error".to_string());
 
-                        if attempt == MAX_RETRIES {
+                        if attempt == retry_config.max_retries {
                             counter!("rust_consumer.clickhouse_insert_error", 1, "status" => status, "retried" => "false");
                             anyhow::bail!(
                                 "error writing to clickhouse after {} attempts: {}",
-                                MAX_RETRIES + 1,
+                                retry_config.max_retries + 1,
                                 error_text
                             );
                         }
@@ -205,18 +218,18 @@ impl ClickhouseClient {
                         tracing::warn!(
                             "ClickHouse write failed (attempt {}/{}): status={}, error={}",
                             attempt + 1,
-                            MAX_RETRIES + 1,
+                            retry_config.max_retries + 1,
                             status,
                             error_text
                         );
                     }
                 }
                 Err(e) => {
-                    if attempt == MAX_RETRIES {
+                    if attempt == retry_config.max_retries {
                         counter!("rust_consumer.clickhouse_insert_error", 1, "status" => "network_error", "retried" => "false");
                         anyhow::bail!(
                             "error writing to clickhouse after {} attempts: {}",
-                            MAX_RETRIES + 1,
+                            retry_config.max_retries + 1,
                             e
                         );
                     }
@@ -225,21 +238,26 @@ impl ClickhouseClient {
                     tracing::warn!(
                         "ClickHouse write failed (attempt {}/{}): {}",
                         attempt + 1,
-                        MAX_RETRIES + 1,
+                        retry_config.max_retries + 1,
                         e
                     );
                 }
             }
 
             // Calculate exponential backoff delay
-            if attempt < MAX_RETRIES {
-                let backoff_ms = INITIAL_BACKOFF_MS * (2_u64.pow(attempt as u32));
-                let delay = Duration::from_millis(backoff_ms);
+            if attempt < retry_config.max_retries {
+                let backoff_ms =
+                    retry_config.initial_backoff_ms * (2_u64.pow(attempt as u32) as f64);
+                // add/subtract up to 10% jitter (by default) to avoid every consumer retrying at the same time
+                // causing too many simultaneous queries
+                let jitter = rand::random::<f64>() * retry_config.jitter_factor
+                    - retry_config.jitter_factor / 2.0; // Random value between (-jitter_factor/2, jitter_factor/2)
+                let delay = Duration::from_millis((backoff_ms * (1.0 + jitter)).round() as u64);
                 tracing::debug!(
                     "Retrying in {:?} (attempt {}/{})",
                     delay,
                     attempt + 1,
-                    MAX_RETRIES
+                    retry_config.max_retries
                 );
                 tokio::time::sleep(delay).await;
             }
@@ -280,7 +298,7 @@ mod tests {
         assert!(client.url.contains("load_balancing"));
         assert!(client.url.contains("insert_distributed_sync"));
         println!("running test");
-        let res = client.send(b"[]".to_vec()).await;
+        let res = client.send(b"[]".to_vec(), RetryConfig::default()).await;
         println!("Response status {}", res.unwrap().status());
         Ok(())
     }
@@ -302,15 +320,24 @@ mod tests {
         let client = ClickhouseClient::new(&config, "test_table");
 
         let start_time = Instant::now();
-        let result = client.send(b"test data".to_vec()).await;
+        let result = client
+            .send(
+                b"test data".to_vec(),
+                RetryConfig {
+                    initial_backoff_ms: 100.0,
+                    max_retries: 4,
+                    jitter_factor: 0.1,
+                },
+            )
+            .await;
         let elapsed = start_time.elapsed();
 
         // Should fail after all retries
         assert!(result.is_err());
 
         // Should have taken at least the sum of our backoff delays
-        // 50ms + 100ms + 200ms + 400ms = 750ms minimum
-        assert!(elapsed >= Duration::from_millis(750));
+        // 90ms + 180ms + 360ms + 720ms = 1350ms minimum
+        assert!(elapsed >= Duration::from_millis(1350));
 
         // Error message should mention the number of attempts
         let error_msg = result.unwrap_err().to_string();