Skip to content

Commit a45c0ec

Browse files
lvaroquiclaude
andauthored
feat(upload): split profile archive uploads into concurrent parts (#565)
* feat(upload): upload profile archives as concurrent multipart uploads 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. Every archive is now sent as an S3 multipart upload, replacing the single-request upload: about two parts per concurrent upload, each between 16 MiB and 256 MiB, so a small archive is a single part. 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 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. This requires an upload endpoint that accepts metadata version 12 and answers with `multipartUploadUrls`. Walltime profile folders above 5 GiB are no longer gzipped on disk to fit in a single request, and the runner no longer caps the archive size itself: the upload endpoint rejects archives above its limit, with the reason shown in the runner output, so the limit can change without a runner release. 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> * feat(upload): check multipart uploads against CRC64NVME checksums Replace the md5 of each part and of the archive with CRC64NVME checksums, and describe the profile archive with a single `profileArchive` field in the upload metadata (version 12): its encoding, size and CRC64NVME, and the size and CRC64NVME of its parts. It replaces `profileEncoding`, `profileMd5` and `profileMultipart`. Each part is hashed once, and the part CRCs are combined into the archive's. S3 can check a CRC64NVME on the whole archive of a multipart upload, which md5 does not support, so a completion that leaves out a part or assembles another archive is rejected, and not only a corrupted part. It is also much cheaper to compute: about 0.5s for a 6 GiB archive on a GitHub-hosted runner, against 24s for the part and archive md5s. The upload endpoint now answers with `multipartUpload`, replacing `multipartUploadUrls`: a presigned request per part and one to complete the upload, each with the headers to send as is, as they are part of the signature. The runner sends them without knowing which ones S3 checks, so the endpoint can change them without a runner release. `crc-fast` is pinned to 1.9, the last release supporting Rust 1.88, which requires `crc` 3.3. Refs COD-3700 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * review changes (will squash merge) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 90cc8e9 commit a45c0ec

11 files changed

Lines changed: 1000 additions & 250 deletions

‎Cargo.lock‎

Lines changed: 22 additions & 9 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ sha256 = "1.6"
3737
tokio = { version = "1", features = ["macros", "rt"] }
3838
tokio-tar = { package = "astral-tokio-tar", version = "0.6.2" }
3939
tokio-util = "0.7.18"
40-
md5 = "0.8"
40+
crc-fast = { version = "1.9.0", default-features = false, features = ["std"] }
4141
base64 = "0.22.1"
4242
async-compression = { version = "0.4.42", features = ["tokio", "gzip"] }
4343
schemars = "1.2.1"
@@ -77,6 +77,7 @@ uuid = { version = "1.23.1", features = ["v4"] }
7777
which = "8.0.2"
7878
crc32fast = "1.5.0"
7979
samply = { path = "crates/samply-codspeed/samply" }
80+
bytes = "1"
8081

8182
# Memory profiling (memtrack) and the capability handling around it are Linux-only.
8283
[target.'cfg(target_os = "linux")'.dependencies]

‎src/run_environment/provider.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -136,8 +136,7 @@ pub trait RunEnvironmentProvider {
136136
tokenless: api_client.token().is_none(),
137137
repository_provider: self.get_repository_provider(),
138138
run_environment_metadata,
139-
profile_md5: profile_archive.hash.clone(),
140-
profile_encoding: profile_archive.content.encoding(),
139+
profile_archive_metadata: profile_archive.metadata.clone(),
141140
commit_hash,
142141
allow_empty: config.allow_empty,
143142
runner: Runner {

‎src/upload/interfaces.rs‎

Lines changed: 78 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,21 @@
1+
use std::collections::BTreeMap;
2+
13
use serde::{Deserialize, Serialize};
24

35
use crate::executor::ExecutorName;
46
use crate::instruments::InstrumentName;
57
use crate::run_environment::{RepositoryProvider, RunEnvironment, RunEnvironmentMetadata, RunPart};
68
use crate::system::SystemInfo;
79

8-
pub const LATEST_UPLOAD_METADATA_VERSION: u32 = 11;
10+
pub const LATEST_UPLOAD_METADATA_VERSION: u32 = 12;
911

1012
#[derive(Serialize, Debug)]
1113
#[serde(rename_all = "camelCase")]
1214
pub struct UploadMetadata {
1315
pub repository_provider: RepositoryProvider,
1416
pub version: Option<u32>,
1517
pub tokenless: bool,
16-
pub profile_md5: String,
17-
pub profile_encoding: Option<String>,
18+
pub profile_archive_metadata: ProfileArchiveMetadata,
1819
pub runner: Runner,
1920
pub run_environment: RunEnvironment,
2021
pub run_part: Option<RunPart>,
@@ -24,6 +25,22 @@ pub struct UploadMetadata {
2425
pub run_environment_metadata: RunEnvironmentMetadata,
2526
}
2627

28+
/// Metadata of the profile archive, uploaded as an S3 multipart upload in consecutive
29+
/// `part_size` chunks, the last one holding the remainder. S3 requires parts of 5 MiB to 5 GiB
30+
/// (the last one excepted from the minimum), and at most 10,000 of them.
31+
#[derive(Serialize, Debug, Clone, PartialEq)]
32+
#[serde(rename_all = "camelCase")]
33+
pub struct ProfileArchiveMetadata {
34+
/// `Content-Encoding` of the archive, such as `gzip`
35+
pub encoding: Option<String>,
36+
pub size: u64,
37+
/// Base64 big-endian CRC64NVME of the whole archive
38+
pub crc64nvme: String,
39+
pub part_size: u64,
40+
/// Base64 big-endian CRC64NVME of each part, in upload order
41+
pub part_crc64nvmes: Vec<String>,
42+
}
43+
2744
#[derive(Serialize, Debug)]
2845
#[serde(rename_all = "camelCase")]
2946
pub struct Runner {
@@ -46,12 +63,69 @@ pub struct Runner {
4663
#[serde(rename_all = "camelCase")]
4764
pub struct UploadData {
4865
pub status: String,
49-
pub upload_url: String,
66+
pub multipart_upload: MultipartUpload,
5067
pub run_id: String,
5168
}
5269

70+
#[derive(Deserialize, Serialize, Debug)]
71+
#[serde(rename_all = "camelCase")]
72+
pub struct MultipartUpload {
73+
/// Presigned S3 `UploadPart` requests, one per part, in upload order
74+
pub parts: Vec<PresignedRequest>,
75+
/// Presigned S3 `CompleteMultipartUpload` request
76+
pub complete: PresignedRequest,
77+
}
78+
79+
/// Request presigned by the API, to send to `url` with `headers` as is: they are
80+
/// part of the signature
81+
#[derive(Deserialize, Serialize, Debug)]
82+
pub struct PresignedRequest {
83+
pub url: String,
84+
pub headers: BTreeMap<String, String>,
85+
}
86+
5387
#[derive(Deserialize, Debug)]
5488
#[serde(rename_all = "camelCase")]
5589
pub struct UploadError {
5690
pub error: String,
5791
}
92+
93+
#[cfg(test)]
94+
mod tests {
95+
use super::*;
96+
97+
#[test]
98+
fn parses_multipart_upload_response() {
99+
let upload_data: UploadData = serde_json::from_str(
100+
r#"{
101+
"status": "success",
102+
"runId": "run-id",
103+
"multipartUpload": {
104+
"parts": [
105+
{ "url": "https://part/1", "headers": { "x-amz-checksum-crc64nvme": "nq48tdaL2no=" } },
106+
{ "url": "https://part/2", "headers": { "x-amz-checksum-crc64nvme": "XMXoclwBfLo=" } }
107+
],
108+
"complete": {
109+
"url": "https://complete",
110+
"headers": { "x-amz-checksum-type": "FULL_OBJECT" }
111+
}
112+
}
113+
}"#,
114+
)
115+
.unwrap();
116+
117+
assert_eq!(upload_data.run_id, "run-id");
118+
let upload = upload_data.multipart_upload;
119+
let part_urls: Vec<_> = upload.parts.iter().map(|part| part.url.as_str()).collect();
120+
assert_eq!(part_urls, ["https://part/1", "https://part/2"]);
121+
assert_eq!(
122+
upload.parts[1].headers,
123+
BTreeMap::from([("x-amz-checksum-crc64nvme".into(), "XMXoclwBfLo=".into())])
124+
);
125+
assert_eq!(upload.complete.url, "https://complete");
126+
assert_eq!(
127+
upload.complete.headers,
128+
BTreeMap::from([("x-amz-checksum-type".into(), "FULL_OBJECT".into())])
129+
);
130+
}
131+
}

‎src/upload/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ mod interfaces;
33
pub mod poll_results;
44
mod profile_archive;
55
mod run_index_state;
6+
mod s3;
67
mod upload_metadata;
78
mod uploader;
89

0 commit comments

Comments
 (0)