From c0841bb7b82d7cff6082737e74955742c831f60b Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Tue, 21 Jul 2026 14:47:10 +0200 Subject: [PATCH 1/3] fix(service): Retry GCS multipart initiate/complete Wrap initiate, list, abort, and complete in the existing GCS request retry helper so transient 5xx/transport failures are absorbed internally. Leave put-part unretried because the body is streamed and not buffered. --- objectstore-service/docs/architecture.md | 7 + objectstore-service/src/backend/gcs.rs | 175 ++++++++++++++--------- 2 files changed, 115 insertions(+), 67 deletions(-) diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 96e51980..4b8adf63 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -198,6 +198,13 @@ delegate to the [`MultipartUploadBackend`](backend::common::MultipartUploadBacke trait, accessed via [`Backend::as_multipart_upload_backend`](backend::common::Backend::as_multipart_upload_backend). Multipart operations share the same concurrency limiter as regular operations. +On the GCS backend, initiate/list/abort/complete reuse the same internal request +retry helper as the JSON API (`408`/`429`/`5xx` and transient transport errors). +Part upload is not retried because the part body is streamed and not buffered. +Complete is safe to retry on those transient errors; a retry after a successful +first attempt surfaces as HTTP 404 `NoSuchUpload`, which is not treated as +retryable. + ## Streaming Concurrency The [`streaming`](streaming) module provides [`StreamExecutor`](streaming::StreamExecutor) diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 924efae5..7e121091 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -890,6 +890,13 @@ impl From for crate::multipart::CompleteMultipartError { /// XXX: Any change that affects this implementation should be manually tested against real GCS. /// That's because the fork of [storage-testbench](https://github.com/googleapis/storage-testbench) /// that we test against has an incomplete implementation of the XML multipart API that likely doesn't match GCS's behavior in many cases. +/// +/// Request-level retries use the same [`GcsBackend::with_retry`] helper as JSON API calls for +/// initiate/list/abort/complete. Part upload is intentionally not retried: the body is streamed and +/// not buffered. Complete is safe to retry on transport/`5xx` errors; if the first attempt already +/// assembled the object, a later retry gets HTTP 404 `NoSuchUpload`, which our retry policy does +/// not treat as retryable and propagates to the caller (and the tiered layer can recover via +/// changelog / presence checks when needed). #[async_trait::async_trait] impl MultipartUploadBackend for GcsBackend { #[tracing::instrument(level = "debug", fields(?id), skip_all)] @@ -902,35 +909,45 @@ 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 content_type = metadata.content_type.clone(); let meta_headers = metadata_to_gcs_headers(metadata)?; - for (name, value) in &meta_headers { - builder = builder.header(name, value); - } - let resp = builder - .send_traced() - .await - .check_error("GCS: initiate multipart upload") - .await?; + self.with_retry("initiate_multipart", || { + let url = url.clone(); + let content_type = content_type.clone(); + let meta_headers = meta_headers.clone(); + async move { + let mut builder = self + .request(Method::POST, url) + .await? + .header(header::CONTENT_TYPE, content_type.as_ref()) + .header(header::CONTENT_LENGTH, "0"); - let body = resp - .bytes() - .await - .map_err(|e| Error::reqwest("GCS: read initiate multipart body", e))?; + for (name, value) in &meta_headers { + builder = builder.header(name, value); + } - 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 resp = builder + .send_traced() + .await + .check_error("GCS: initiate multipart upload") + .await?; - Ok(xml.try_into()?) + 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)), + })?; + + xml.try_into() + } + }) + .await } #[tracing::instrument(level = "debug", skip(self, content_md5, body))] @@ -943,6 +960,7 @@ impl MultipartUploadBackend for GcsBackend { content_md5: Option<&str>, body: ClientStream, ) -> Result { + // Not retried: the part body is a one-shot stream and is not buffered. objectstore_log::debug!("Uploading part to GCS backend"); let mut url = self.xml_object_url(id)?; url.query_pairs_mut() @@ -998,26 +1016,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))] @@ -1030,16 +1054,22 @@ 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 { + self.request(Method::DELETE, url) + .await? + .send_traced() + .await + .check_error("GCS: abort multipart upload") + .await? + .drain_body() + .await; - Ok(()) + Ok(()) + } + }) + .await } #[tracing::instrument(level = "debug", skip(self, parts))] @@ -1053,32 +1083,43 @@ impl MultipartUploadBackend for GcsBackend { let mut url = self.xml_object_url(id)?; url.query_pairs_mut().append_pair("uploadId", upload_id); + // Serialize once — the payload is small and identical across retries. let body = XmlCompleteMultipartUpload::from(parts); let xml = quick_xml::se::to_string(&body).map_err(|e| Error::Generic { context: "GCS: failed to serialize complete multipart request".into(), 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))?; + let body = resp + .bytes() + .await + .map_err(|e| Error::reqwest("GCS: read complete multipart body", e))?; - let error = quick_xml::de::from_reader::<_, XmlError>(body.as_ref()) - .ok() - .map(Into::into); + // Semantic complete errors (InvalidPart, etc.) come back as XML bodies under a + // success-ish status in our mapping, and are returned as Ok(Some(error)) — not + // retried, since they are permanent request errors. + let error = quick_xml::de::from_reader::<_, XmlError>(body.as_ref()) + .ok() + .map(Into::into); - Ok(error) + Ok(error) + } + }) + .await } } From 9ab1f09f087ce01e008d55e55d895f193f9b9f04 Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Tue, 21 Jul 2026 14:49:35 +0200 Subject: [PATCH 2/3] fix(service): Drop private rustdoc link in GCS multipart docs GcsBackend::with_retry is private, so the intra-doc link failed rustdoc with -Dwarnings in Lint and docs CI jobs. --- objectstore-service/src/backend/gcs.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 7e121091..977e57fe 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -891,7 +891,7 @@ impl From for crate::multipart::CompleteMultipartError { /// That's because the fork of [storage-testbench](https://github.com/googleapis/storage-testbench) /// that we test against has an incomplete implementation of the XML multipart API that likely doesn't match GCS's behavior in many cases. /// -/// Request-level retries use the same [`GcsBackend::with_retry`] helper as JSON API calls for +/// Request-level retries use the same internal retry helper as JSON API calls for /// initiate/list/abort/complete. Part upload is intentionally not retried: the body is streamed and /// not buffered. Complete is safe to retry on transport/`5xx` errors; if the first attempt already /// assembled the object, a later retry gets HTTP 404 `NoSuchUpload`, which our retry policy does From 033741cd411b02ed83acb0aaf352ca9b46bc7a09 Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Tue, 21 Jul 2026 14:55:12 +0200 Subject: [PATCH 3/3] fix(service): Make GCS multipart abort idempotent on 404 After a successful abort whose response is lost, retry DELETE can get 404 NoSuchUpload. Treat that as success, matching delete_object. --- objectstore-service/docs/architecture.md | 7 ++++--- objectstore-service/src/backend/gcs.rs | 26 ++++++++++++++++++------ 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 4b8adf63..963b6190 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -201,9 +201,10 @@ Multipart operations share the same concurrency limiter as regular operations. On the GCS backend, initiate/list/abort/complete reuse the same internal request retry helper as the JSON API (`408`/`429`/`5xx` and transient transport errors). Part upload is not retried because the part body is streamed and not buffered. -Complete is safe to retry on those transient errors; a retry after a successful -first attempt surfaces as HTTP 404 `NoSuchUpload`, which is not treated as -retryable. +Abort treats HTTP 404 as success so a retry after a lost successful abort still +succeeds. Complete is safe to retry on those transient errors; a retry after a +successful first attempt surfaces as HTTP 404 `NoSuchUpload`, which is not +treated as retryable. ## Streaming Concurrency diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 977e57fe..410ab231 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -893,10 +893,11 @@ impl From for crate::multipart::CompleteMultipartError { /// /// Request-level retries use the same internal retry helper as JSON API calls for /// initiate/list/abort/complete. Part upload is intentionally not retried: the body is streamed and -/// not buffered. Complete is safe to retry on transport/`5xx` errors; if the first attempt already -/// assembled the object, a later retry gets HTTP 404 `NoSuchUpload`, which our retry policy does -/// not treat as retryable and propagates to the caller (and the tiered layer can recover via -/// changelog / presence checks when needed). +/// not buffered. Abort treats HTTP 404 as success (idempotent if the session is already gone, e.g. +/// after a successful abort whose response was lost). Complete is safe to retry on transport/`5xx` +/// errors; if the first attempt already assembled the object, a later retry gets HTTP 404 +/// `NoSuchUpload`, which our retry policy does not treat as retryable and propagates to the caller +/// (and the tiered layer can recover via changelog / presence checks when needed). #[async_trait::async_trait] impl MultipartUploadBackend for GcsBackend { #[tracing::instrument(level = "debug", fields(?id), skip_all)] @@ -1057,11 +1058,21 @@ impl MultipartUploadBackend for GcsBackend { self.with_retry("abort_multipart", || { let url = url.clone(); async move { - self.request(Method::DELETE, url) + let resp = self + .request(Method::DELETE, url) .await? .send_traced() .await - .check_error("GCS: abort multipart upload") + .map_err(|e| Error::reqwest("GCS: abort multipart upload", e))?; + + // Idempotent: absent upload means the postcondition is already satisfied + // (including retry after a successful abort whose response was lost). + if resp.status() == StatusCode::NOT_FOUND { + resp.drain_body().await; + return Ok(()); + } + + resp.check_error("GCS: abort multipart upload") .await? .drain_body() .await; @@ -1771,6 +1782,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(()) }