fix(logs): restore retry logic and update flush API signature

- Change LogFlusher::flush() to accept retry builders (Vec<RequestBuilder>)
- Restore redrive loop in FlushingService for transient log flush failures
- Update log_flush_handles type to return retry builders instead of bool
- Suppress clippy::result_large_err in test modules across config and otlp

Author: litianningdatadog

bottlecap/src/config/env.rs

@@ -711,6 +711,7 @@ impl ConfigSource for EnvConfigSource {
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
 mod tests {
+    #![allow(clippy::result_large_err)]
     use std::time::Duration;
 
     use super::*;

bottlecap/src/config/mod.rs

@@ -811,6 +811,7 @@ pub fn deserialize_optional_duration_from_seconds_ignore_zero<'de, D: Deserializ
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
 pub mod tests {
+    #![allow(clippy::result_large_err)]
     use libdd_trace_obfuscation::replacer::parse_rules_from_string;
 
     use super::*;

bottlecap/src/config/yaml.rs

@@ -742,6 +742,7 @@ impl ConfigSource for YamlConfigSource {
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
 mod tests {
+    #![allow(clippy::result_large_err)]
     use std::path::Path;
     use std::time::Duration;
 

bottlecap/src/flushing/handles.rs

@@ -27,8 +27,8 @@ pub struct MetricsRetryBatch {
 pub struct FlushHandles {
     /// Handles for trace flush operations. Returns failed traces for retry.
     pub trace_flush_handles: Vec<JoinHandle<Vec<SendData>>>,
-    /// Handles for log flush operations. Returns true on success, false on any failure.
-    pub log_flush_handles: Vec<JoinHandle<bool>>,
+    /// Handles for log flush operations. Returns builders for transient failures to retry next invocation.
+    pub log_flush_handles: Vec<JoinHandle<Vec<reqwest::RequestBuilder>>>,
     /// Handles for metrics flush operations. Returns batch info for retry.
     pub metric_flush_handles: Vec<JoinHandle<MetricsRetryBatch>>,
     /// Handles for proxy flush operations. Returns failed request builders for retry.

bottlecap/src/flushing/service.rs

@@ -77,7 +77,7 @@ impl FlushingService {
         let lf = self.logs_flusher.clone();
         self.handles
             .log_flush_handles
-            .push(tokio::spawn(async move { lf.flush().await }));
+            .push(tokio::spawn(async move { lf.flush(vec![]).await }));
 
         // Spawn traces flush
         let tf = self.trace_flusher.clone();
@@ -192,18 +192,32 @@ impl FlushingService {
             }
         }
 
-        // Await log handles — retries are handled internally by LogFlusher
+        // Await log handles with retry
         for handle in self.handles.log_flush_handles.drain(..) {
             match handle.await {
-                Ok(success) => {
-                    if !success {
-                        debug!("FLUSHING_SERVICE | log flush reported a failure");
-                        flush_error = true;
+                Ok(retry) => {
+                    if !retry.is_empty() {
+                        debug!(
+                            "FLUSHING_SERVICE | redriving {:?} log payloads",
+                            retry.len()
+                        );
+                    }
+                    for item in retry {
+                        let lf = self.logs_flusher.clone();
+                        match item.try_clone() {
+                            Some(item_clone) => {
+                                joinset.spawn(async move {
+                                    lf.flush(vec![item_clone]).await;
+                                });
+                            }
+                            None => {
+                                error!("FLUSHING_SERVICE | Can't clone redrive log payloads");
+                            }
+                        }
                     }
                 }
                 Err(e) => {
-                    error!("FLUSHING_SERVICE | log flush task error {e:?}");
-                    flush_error = true;
+                    error!("FLUSHING_SERVICE | redrive log error {e:?}");
                 }
             }
         }
@@ -312,7 +326,7 @@ impl FlushingService {
             .collect();
 
         tokio::join!(
-            self.logs_flusher.flush(),
+            self.logs_flusher.flush(vec![]),
             futures::future::join_all(metrics_futures),
             self.trace_flusher.flush(None),
             self.stats_flusher.flush(force_stats, None),

bottlecap/src/otlp/mod.rs

@@ -16,6 +16,7 @@ pub fn should_enable_otlp_agent(config: &Arc<Config>) -> bool {
 
 #[cfg(test)]
 mod tests {
+    #![allow(clippy::result_large_err)]
     use super::*;
 
     use std::path::Path;

bottlecap/tests/logs_integration_test.rs

@@ -103,7 +103,7 @@ async fn test_logs() {
 
     logs_agent.sync_consume().await;
 
-    let _ = logs_flusher.flush().await;
+    let _ = logs_flusher.flush(vec![]).await;
 
     hello_mock.assert();
 }