Skip to content

fix(gcs): Retry multipart operations #575

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
lcian merged 4 commits into from
Jul 27, 2026
Merged

fix(gcs): Retry multipart operations #575

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e374cc0
fix(service): Retry GCS multipart initiate/list/abort
lcian Jul 24, 2026
ff2e220
improve
lcian Jul 24, 2026
88aa47d
improve
lcian Jul 24, 2026
0b6fca4
dont need to handle 404; retry complete too
lcian Jul 27, 2026
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
181 changes: 113 additions & 68 deletions objectstore-service/src/backend/gcs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -898,35 +898,47 @@ impl MultipartUploadBackend for GcsBackend {
let mut url = self.xml_object_url(id)?;
url.set_query(Some("uploads"));

let mut builder = self
.request(Method::POST, url)
.await?
.header(header::CONTENT_TYPE, metadata.content_type.as_ref())
.header(header::CONTENT_LENGTH, "0");

let meta_headers = metadata_to_gcs_headers(metadata)?;
for (name, value) in &meta_headers {
builder = builder.header(name, value);
}
let mut headers = metadata_to_gcs_headers(metadata)?;
headers.insert(
header::CONTENT_TYPE,
metadata.content_type.parse().map_err(|e| Error::Generic {
context: "GCS: invalid content-type header value".into(),
cause: Some(Box::new(e)),
})?,
);
headers.insert(
header::CONTENT_LENGTH,
header::HeaderValue::from_static("0"),
);

let resp = builder
.send_traced()
.await
.check_error("GCS: initiate multipart upload")
.await?;
self.with_retry("initiate_multipart", || {
let url = url.clone();
let headers = headers.clone();
async move {
let resp = self
.request(Method::POST, url)
.await?
.headers(headers)
.send_traced()
.await
.check_error("GCS: initiate multipart upload")
.await?;

let body = resp
.bytes()
.await
.map_err(|e| Error::reqwest("GCS: read initiate multipart body", e))?;
let body = resp
.bytes()
.await
.map_err(|e| Error::reqwest("GCS: read initiate multipart body", e))?;

let xml: XmlInitiateMultipartUploadResponse = quick_xml::de::from_reader(body.as_ref())
.map_err(|e| Error::Generic {
context: "GCS: failed to parse initiate multipart response".to_owned(),
cause: Some(Box::new(e)),
})?;
let xml: XmlInitiateMultipartUploadResponse =
quick_xml::de::from_reader(body.as_ref()).map_err(|e| Error::Generic {
context: "GCS: failed to parse initiate multipart response".to_owned(),
cause: Some(Box::new(e)),
})?;

Ok(xml.try_into()?)
xml.try_into()
}
})
.await
}

#[tracing::instrument(level = "debug", skip(self, content_md5, body))]
Expand Down Expand Up @@ -994,26 +1006,32 @@ impl MultipartUploadBackend for GcsBackend {
}
}

let resp = self
.request(Method::GET, url)
.await?
.send_traced()
.await
.check_error("GCS: list parts")
.await?;
self.with_retry("list_parts", || {
let url = url.clone();
async move {
let resp = self
.request(Method::GET, url)
.await?
.send_traced()
.await
.check_error("GCS: list parts")
.await?;

let body = resp
.bytes()
.await
.map_err(|e| Error::reqwest("GCS: read list parts body", e))?;
let body = resp
.bytes()
.await
.map_err(|e| Error::reqwest("GCS: read list parts body", e))?;

let xml: XmlListPartsResponse =
quick_xml::de::from_reader(body.as_ref()).map_err(|e| Error::Generic {
context: "GCS: failed to parse list parts response".to_owned(),
cause: Some(Box::new(e)),
})?;
let xml: XmlListPartsResponse =
quick_xml::de::from_reader(body.as_ref()).map_err(|e| Error::Generic {
context: "GCS: failed to parse list parts response".to_owned(),
cause: Some(Box::new(e)),
})?;

Ok(xml.into())
Ok(xml.into())
}
})
.await
}

#[tracing::instrument(level = "debug", skip(self))]
Expand All @@ -1026,16 +1044,29 @@ impl MultipartUploadBackend for GcsBackend {
let mut url = self.xml_object_url(id)?;
url.query_pairs_mut().append_pair("uploadId", upload_id);

self.request(Method::DELETE, url)
.await?
.send_traced()
.await
.check_error("GCS: abort multipart upload")
.await?
.drain_body()
.await;
self.with_retry("abort_multipart", || {
let url = url.clone();
async move {
let resp = self
.request(Method::DELETE, url)
.await?
.send_traced()
.await
.map_err(|e| Error::reqwest("GCS: abort multipart upload", e))?;

Ok(())
// XXX: real S3 would return 404 here if the upload has been recently completed and we
// would have to handle it. It turns out GCS returns 204 instead, so we don't need to
// handle that case.

resp.check_error("GCS: abort multipart upload")
.await?
.drain_body()
.await;

Ok(())
}
})
.await
}

#[tracing::instrument(level = "debug", skip(self, parts))]
Expand All @@ -1055,26 +1086,37 @@ impl MultipartUploadBackend for GcsBackend {
cause: Some(Box::new(e)),
})?;

let resp = self
.request(Method::POST, url)
.await?
.header(header::CONTENT_TYPE, "application/xml")
.body(xml)
.send_traced()
.await
.check_error("GCS: complete multipart upload")
.await?;
self.with_retry("complete_multipart", || {
let url = url.clone();
let xml = xml.clone();
async move {
let resp = self
.request(Method::POST, url)
.await?
.header(header::CONTENT_TYPE, "application/xml")
.body(xml)
.send_traced()
.await
.check_error("GCS: complete multipart upload")
.await?;

let body = resp
.bytes()
.await
.map_err(|e| Error::reqwest("GCS: read complete multipart body", e))?;
// XXX: real S3 would return 404 here if the upload has been recently completed and we
// would have to handle it. It turns out GCS returns 200 instead, so we don't need to
// handle that case.

let error = quick_xml::de::from_reader::<_, XmlError>(body.as_ref())
.ok()
.map(Into::into);
let body = resp
.bytes()
.await
.map_err(|e| Error::reqwest("GCS: read complete multipart body", e))?;

Ok(error)
let error = quick_xml::de::from_reader::<_, XmlError>(body.as_ref())
.ok()
.map(Into::into);

Ok(error)
}
})
.await
}
}

Expand Down Expand Up @@ -1761,6 +1803,9 @@ mod tests {
let result = backend.get_object(&id, None).await?;
assert!(result.is_none(), "object should not exist after abort");

// A second abort should still succeed (idempotent 404 handling).
backend.abort_multipart(&id, &upload_id).await?;

Ok(())
}

Expand Down
Loading