fix(service): Retry GCS multipart initiate and complete

objectstore-service/docs/architecture.md

@@ -198,6 +198,14 @@ 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.
+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
 
 The [`streaming`](streaming) module provides [`StreamExecutor`](streaming::StreamExecutor)

objectstore-service/src/backend/gcs.rs

@@ -890,6 +890,14 @@ impl From<XmlError> 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 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. 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)]
@@ -902,35 +910,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?;
+
+                let body = resp
+                    .bytes()
+                    .await
+                    .map_err(|e| Error::reqwest("GCS: read initiate multipart body", e))?;
 
-        Ok(xml.try_into()?)
+                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 +961,7 @@ impl MultipartUploadBackend for GcsBackend {
         content_md5: Option<&str>,
         body: ClientStream,
     ) -> Result<UploadPartResponse> {
+        // 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 +1017,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 +1055,32 @@ 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(())
+                // 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;
+
+                Ok(())
+            }
+        })
+        .await
     }
 
     #[tracing::instrument(level = "debug", skip(self, parts))]
@@ -1053,32 +1094,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
     }
 }
 
@@ -1730,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(())
     }