Skip to content

central_systest_blobs: generalize the GCS helpers to fetch and upload arbitrary objects #14920

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
itamar-starkware merged 1 commit into from
Aug 5, 2026
Merged

central_systest_blobs: generalize the GCS helpers to fetch and upload arbitrary objects #14920

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 14 additions & 13 deletions crates/central_systest_blobs/src/cende_blob_regression_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -791,7 +791,7 @@ async fn gcs_client() -> Client {
async fn find_next_available_blobs_generation(client: &Client) -> usize {
let mut next_generation = current_generation() + 1;
loop {
match fetch_raw_blobs_at_generation(client, next_generation).await {
match fetch_raw_object(client, blobs_object_path(next_generation)).await {
Ok(_) => next_generation += 1,
Err(GcsError::Response(ErrorResponse { code: GCS_ERROR_CODE_NOT_FOUND, .. })) => break,
Err(GcsError::HttpClient(error))
Expand All @@ -805,26 +805,20 @@ async fn find_next_available_blobs_generation(client: &Client) -> usize {
next_generation
}

async fn fetch_raw_blobs_at_generation(
client: &Client,
generation: usize,
) -> Result<Vec<u8>, GcsError> {
async fn fetch_raw_object(client: &Client, object: String) -> Result<Vec<u8>, GcsError> {
client
.download_object(
&GetObjectRequest {
bucket: BLOBS_BUCKET_NAME.to_string(),
object: blobs_object_path(generation),
object,
..Default::default()
},
&Range::default(),
)
.await
}

/// Pushes the blobs to GCS.
async fn bump_generation_and_store_blob_file(blobs: Vec<AerospikeBlob>, client: &Client) {
let blobs_json = to_normalized_json(&blobs);
let next_generation = find_next_available_blobs_generation(client).await;
async fn upload_new_object(client: &Client, object: String, content: String) {
client
.upload_object(
&UploadObjectRequest {
Expand All @@ -833,17 +827,24 @@ async fn bump_generation_and_store_blob_file(blobs: Vec<AerospikeBlob>, client:
if_generation_match: Some(0),
..Default::default()
},
blobs_json.into_bytes(),
&UploadType::Simple(Media::new(blobs_object_path(next_generation))),
content.into_bytes(),
&UploadType::Simple(Media::new(object)),
)
.await
.unwrap();
}

/// Pushes the blobs to GCS.
async fn bump_generation_and_store_blob_file(blobs: Vec<AerospikeBlob>, client: &Client) {
let next_generation = find_next_available_blobs_generation(client).await;
upload_new_object(client, blobs_object_path(next_generation), to_normalized_json(&blobs)).await;
fs::write(&*BLOBS_GENERATION_FILE, next_generation.to_string()).unwrap();
}

/// Fetches the blobs from GCS.
async fn fetch_blob_file(client: &Client) -> Vec<AerospikeBlob> {
let blobs_json = fetch_raw_blobs_at_generation(client, current_generation()).await.unwrap();
let blobs_json =
fetch_raw_object(client, blobs_object_path(current_generation())).await.unwrap();
serde_json::from_slice(&blobs_json).unwrap()
}

Expand Down
Loading