diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 96e51980..963b6190 100644 --- a/objectstore-service/docs/architecture.md +++ b/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) diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 924efae5..410ab231 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -890,6 +890,14 @@ 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 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 { + // 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(()) }