fix(upload): retry streamed uploads on transient failures

The on-disk/streamed upload path (WallTime/Memory) used STREAMING_CLIENT,
which has no retry middleware, so a transient failure killed the upload
after a single attempt. The retry middleware can't help here because it
retries by cloning the request, and a streamed body isn't cloneable.

Add send_streamed_with_retry, which rebuilds the stream from disk on each
attempt while reusing reqwest_retry's ExponentialBackoff policy and
DefaultRetryableStrategy, so the streamed path retries with the same
backoff and transient classification as the middleware-backed path.

Author: not-matthias

src/request_client.rs

@@ -4,7 +4,7 @@ use reqwest::ClientBuilder;
 use reqwest_middleware::{ClientBuilder as ClientWithMiddlewareBuilder, ClientWithMiddleware};
 use reqwest_retry::{RetryTransientMiddleware, policies::ExponentialBackoff};
 
-const UPLOAD_RETRY_COUNT: u32 = 3;
+pub const UPLOAD_RETRY_COUNT: u32 = 3;
 const OIDC_RETRY_COUNT: u32 = 10;
 const USER_AGENT: &str = "codspeed-runner";
 

src/upload/uploader.rs

@@ -6,13 +6,18 @@ use crate::run_environment::RunEnvironment;
 use crate::upload::{UploadError, profile_archive::ProfileArchiveContent};
 use crate::{
     prelude::*,
-    request_client::{REQUEST_CLIENT, STREAMING_CLIENT},
+    request_client::{REQUEST_CLIENT, STREAMING_CLIENT, UPLOAD_RETRY_COUNT},
 };
 use async_compression::tokio::write::GzipEncoder;
 use console::style;
 use reqwest::StatusCode;
+use reqwest_retry::{
+    DefaultRetryableStrategy, RetryDecision, RetryPolicy, Retryable, RetryableStrategy,
+    policies::ExponentialBackoff,
+};
 use serde_json::Value;
 use std::collections::BTreeMap;
+use std::time::SystemTime;
 use tokio::fs::File;
 use tokio::io::AsyncWriteExt;
 use tokio_tar::Builder;
@@ -182,6 +187,64 @@ async fn retrieve_upload_data(
     }
 }
 
+/// Streams a file-backed body with retries. The retry middleware can't replay a
+/// stream body (a consumed stream isn't cloneable), so we rebuild it from disk on
+/// each attempt and reuse reqwest_retry's backoff policy + transient classification
+/// to match the behavior of the middleware-backed in-memory path.
+async fn send_streamed_with_retry(
+    upload_data: &UploadData,
+    path: &std::path::Path,
+    archive_size: u64,
+    archive_hash: &str,
+    encoding: Option<String>,
+) -> Result<reqwest::Response> {
+    let policy = ExponentialBackoff::builder().build_with_max_retries(UPLOAD_RETRY_COUNT);
+    let start = SystemTime::now();
+    let mut n_past_retries = 0;
+
+    loop {
+        let file = File::open(path)
+            .await
+            .context(format!("Failed to open file at path: {}", path.display()))?;
+        let stream = tokio_util::io::ReaderStream::new(file);
+        let body = reqwest::Body::wrap_stream(stream);
+
+        let mut request = STREAMING_CLIENT
+            .put(upload_data.upload_url.clone())
+            .header("Content-Type", "application/x-tar")
+            .header("Content-Length", archive_size)
+            .header("Content-MD5", archive_hash);
+        if let Some(encoding) = &encoding {
+            request = request.header("Content-Encoding", encoding);
+        }
+
+        // Wrap into reqwest_middleware::Error so the default classifier can be reused.
+        let result = request
+            .body(body)
+            .send()
+            .await
+            .map_err(reqwest_middleware::Error::Reqwest);
+
+        let is_transient =
+            matches!(DefaultRetryableStrategy.handle(&result), Some(Retryable::Transient));
+        if is_transient {
+            if let RetryDecision::Retry { execute_after } =
+                policy.should_retry(start, n_past_retries)
+            {
+                let wait = execute_after
+                    .duration_since(SystemTime::now())
+                    .unwrap_or_default();
+                debug!("Streamed upload attempt failed (transient), retrying in {wait:?}");
+                tokio::time::sleep(wait).await;
+                n_past_retries += 1;
+                continue;
+            }
+        }
+
+        return Ok(result?);
+    }
+}
+
 async fn upload_profile_archive(
     upload_data: &UploadData,
     profile_archive: ProfileArchive,
@@ -206,24 +269,14 @@ async fn upload_profile_archive(
         }
         content @ ProfileArchiveContent::UncompressedOnDisk { path }
         | content @ ProfileArchiveContent::CompressedOnDisk { path } => {
-            // Use streaming client without retry middleware for file streams
-            let file = File::open(path)
-                .await
-                .context(format!("Failed to open file at path: {}", path.display()))?;
-            let stream = tokio_util::io::ReaderStream::new(file);
-            let body = reqwest::Body::wrap_stream(stream);
-
-            let mut request = STREAMING_CLIENT
-                .put(upload_data.upload_url.clone())
-                .header("Content-Type", "application/x-tar")
-                .header("Content-Length", archive_size)
-                .header("Content-MD5", archive_hash);
-
-            if let Some(encoding) = content.encoding() {
-                request = request.header("Content-Encoding", encoding);
-            }
-
-            request.body(body).send().await?
+            send_streamed_with_retry(
+                upload_data,
+                path,
+                archive_size,
+                &archive_hash,
+                content.encoding(),
+            )
+            .await?
         }
     };