feat(upload): split profile archive uploads into concurrent parts

S3 rejects single uploads above 5 GiB, and a single connection to S3
only reaches about 20-25 MiB/s on GitHub-hosted runners, so large
profile archives were slow or impossible to upload.

Archives of 64 MiB and more are now sent as an S3 multipart upload, in
about two parts per concurrent upload, each between 16 MiB and 256 MiB.
The md5 of every part is computed in the same pass as the archive md5
and sent in the upload metadata (version 12) as `profileMultipart`. The
API then answers with `multipartUploadUrls`: the parts are uploaded 8 at
a time (overridable with `CODSPEED_UPLOAD_CONCURRENCY`), each with its
own retries, then the upload is completed with the part ETags in order.
This applies to both on-disk and in-memory (gzip) archives.

Walltime profile folders above 5 GiB are no longer gzipped on disk to
fit in a single request, and the maximum archive size goes from 5 GiB to
15 GiB. Archives are now hashed while streaming on the blocking thread
pool, instead of being read whole into memory on the async runtime.

Closes COD-3700
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: lvaroqui

Cargo.lock

@@ -69,6 +69,7 @@ uuid = { version = "1.23.1", features = ["v4"] }
 which = "8.0.2"
 crc32fast = "1.5.0"
 samply = { path = "crates/samply-codspeed/samply" }
+bytes = "1"
 
 # Memory profiling (memtrack) and the capability handling around it are Linux-only.
 [target.'cfg(target_os = "linux")'.dependencies]

Cargo.toml

@@ -136,8 +136,9 @@ pub trait RunEnvironmentProvider {
             tokenless: api_client.token().is_none(),
             repository_provider: self.get_repository_provider(),
             run_environment_metadata,
-            profile_md5: profile_archive.hash.clone(),
+            profile_md5: profile_archive.md5.clone(),
             profile_encoding: profile_archive.content.encoding(),
+            profile_multipart: profile_archive.multipart.clone(),
             commit_hash,
             allow_empty: config.allow_empty,
             runner: Runner {

src/run_environment/provider.rs

@@ -5,7 +5,7 @@ use crate::instruments::InstrumentName;
 use crate::run_environment::{RepositoryProvider, RunEnvironment, RunEnvironmentMetadata, RunPart};
 use crate::system::SystemInfo;
 
-pub const LATEST_UPLOAD_METADATA_VERSION: u32 = 11;
+pub const LATEST_UPLOAD_METADATA_VERSION: u32 = 12;
 
 #[derive(Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
@@ -15,6 +15,8 @@ pub struct UploadMetadata {
     pub tokenless: bool,
     pub profile_md5: String,
     pub profile_encoding: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub profile_multipart: Option<ProfileMultipart>,
     pub runner: Runner,
     pub run_environment: RunEnvironment,
     pub run_part: Option<RunPart>,
@@ -24,6 +26,18 @@ pub struct UploadMetadata {
     pub run_environment_metadata: RunEnvironmentMetadata,
 }
 
+/// Layout of a profile archive uploaded as an S3 multipart upload, in consecutive
+/// `part_size` chunks, the last one holding the remainder. S3 requires parts of 5 MiB
+/// to 5 GiB (the last one excepted from the minimum), and at most 10,000 of them.
+#[derive(Serialize, Debug, Clone, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct ProfileMultipart {
+    pub size: u64,
+    pub part_size: u64,
+    /// Base64 md5 of each part, in upload order
+    pub part_md5s: Vec<String>,
+}
+
 #[derive(Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Runner {
@@ -46,12 +60,75 @@ pub struct Runner {
 #[serde(rename_all = "camelCase")]
 pub struct UploadData {
     pub status: String,
-    pub upload_url: String,
+    #[serde(flatten)]
+    pub target: UploadTarget,
     pub run_id: String,
 }
 
+#[derive(Deserialize, Serialize, Debug)]
+#[serde(rename_all = "camelCase")]
+pub enum UploadTarget {
+    /// Set for archives uploaded in a single request
+    UploadUrl(String),
+    /// Set for archives uploaded as an S3 multipart upload
+    MultipartUploadUrls(MultipartUploadUrls),
+}
+
+#[derive(Deserialize, Serialize, Debug)]
+#[serde(rename_all = "camelCase")]
+pub struct MultipartUploadUrls {
+    /// Presigned S3 `UploadPart` URLs, one per part, in upload order
+    pub part_urls: Vec<String>,
+    /// Presigned S3 `CompleteMultipartUpload` URL
+    pub complete_url: String,
+}
+
 #[derive(Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct UploadError {
     pub error: String,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_single_upload_response() {
+        let upload_data: UploadData = serde_json::from_str(
+            r#"{"status": "success", "uploadUrl": "https://upload", "runId": "run-id"}"#,
+        )
+        .unwrap();
+
+        assert_eq!(upload_data.run_id, "run-id");
+        let UploadTarget::UploadUrl(upload_url) = upload_data.target else {
+            panic!("expected a single upload URL, got {:?}", upload_data.target);
+        };
+        assert_eq!(upload_url, "https://upload");
+    }
+
+    #[test]
+    fn parses_multipart_upload_response() {
+        let upload_data: UploadData = serde_json::from_str(
+            r#"{
+                "status": "success",
+                "runId": "run-id",
+                "multipartUploadUrls": {
+                    "partUrls": ["https://part/1", "https://part/2"],
+                    "completeUrl": "https://complete"
+                }
+            }"#,
+        )
+        .unwrap();
+
+        assert_eq!(upload_data.run_id, "run-id");
+        let UploadTarget::MultipartUploadUrls(urls) = upload_data.target else {
+            panic!(
+                "expected multipart upload URLs, got {:?}",
+                upload_data.target
+            );
+        };
+        assert_eq!(urls.part_urls, ["https://part/1", "https://part/2"]);
+        assert_eq!(urls.complete_url, "https://complete");
+    }
+}

src/upload/interfaces.rs

@@ -3,6 +3,7 @@ mod interfaces;
 pub mod poll_results;
 mod profile_archive;
 mod run_index_state;
+mod s3;
 mod upload_metadata;
 mod uploader;