fix(gcs+s3): Drain response bodies (#538)

Author: lcian

objectstore-service/src/backend/gcs.rs

@@ -537,6 +537,7 @@ impl GcsBackend {
                     .map_err(|e| Error::reqwest("GCS: get metadata request", e))?;
 
                 if resp.status() == StatusCode::NOT_FOUND {
+                    resp.drain_body().await;
                     return Ok(None);
                 }
 
@@ -595,7 +596,9 @@ impl GcsBackend {
                 .send()
                 .await
                 .check_error("GCS: update custom time")
-                .await?;
+                .await?
+                .drain_body()
+                .await;
             Ok(())
         })
         .await
@@ -667,7 +670,9 @@ impl Backend for GcsBackend {
             .send()
             .await
             .check_error("GCS: upload object")
-            .await?;
+            .await?
+            .drain_body()
+            .await;
 
         Ok(())
     }
@@ -701,17 +706,14 @@ impl Backend for GcsBackend {
                         .get(header::CONTENT_RANGE)
                         .and_then(|v| v.to_str().ok());
                     let total = raw.and_then(ContentRange::parse_unsatisfiable_total);
-                    match total {
-                        Some(total) => return Err(Error::RangeNotSatisfiable { total }),
-                        None => {
-                            return Err(Error::Generic {
-                                context: format!(
-                                    "GCS: 416 response with invalid Content-Range: {raw:?}"
-                                ),
-                                cause: None,
-                            });
-                        }
-                    }
+                    let err = match total {
+                        Some(total) => Error::RangeNotSatisfiable { total },
+                        None => Error::generic(format!(
+                            "GCS: 416 response with invalid Content-Range: {raw:?}"
+                        )),
+                    };
+                    resp.drain_body().await;
+                    return Err(err);
                 }
 
                 resp.check_error("GCS: get payload").await
@@ -764,10 +766,14 @@ impl Backend for GcsBackend {
 
             // Do not error for objects that do not exist
             if resp.status() == StatusCode::NOT_FOUND {
+                resp.drain_body().await;
                 return Ok(());
             }
 
-            resp.check_error("GCS: delete object").await?;
+            resp.check_error("GCS: delete object")
+                .await?
+                .drain_body()
+                .await;
 
             Ok(())
         })
@@ -960,6 +966,8 @@ impl MultipartUploadBackend for GcsBackend {
             .map(|s| s.to_owned())
             .ok_or_else(|| Error::generic("GCS: upload part response missing ETag header"))?;
 
+        resp.drain_body().await;
+
         Ok(etag)
     }
 
@@ -1021,7 +1029,9 @@ impl MultipartUploadBackend for GcsBackend {
             .send()
             .await
             .check_error("GCS: abort multipart upload")
-            .await?;
+            .await?
+            .drain_body()
+            .await;
 
         Ok(())
     }

objectstore-service/src/backend/response.rs

@@ -66,6 +66,13 @@ pub trait ResponseExt {
     /// When called on `Result<Response, reqwest::Error>`, transport errors are
     /// wrapped as [`Error::Reqwest`] with the same context string.
     async fn check_error(self, context: &'static str) -> Result<Response>;
+
+    /// Drains the response body of a response we are otherwise done with.
+    ///
+    /// reqwest only returns a connection to its pool once the response body has been fully read, so
+    /// we need to explicitly drain it. Errors are swallowed, since the caller has already obtained
+    /// everything it needs from the response.
+    async fn drain_body(self);
 }
 
 impl ResponseExt for Response {
@@ -86,9 +93,11 @@ impl ResponseExt for Response {
         } else if ct.starts_with("application/xml") || ct.starts_with("text/xml") {
             parse_xml_error(self).await
         } else {
-            return self
-                .error_for_status()
-                .map_err(|e| Error::reqwest(context, e));
+            let Err(e) = self.error_for_status_ref() else {
+                return Ok(self);
+            };
+            self.drain_body().await;
+            return Err(Error::reqwest(context, e));
         };
 
         Err(Error::BackendResponse {
@@ -97,6 +106,10 @@ impl ResponseExt for Response {
             detail,
         })
     }
+
+    async fn drain_body(mut self) {
+        while let Ok(Some(_)) = self.chunk().await {}
+    }
 }
 
 impl ResponseExt for Result<Response, reqwest::Error> {
@@ -109,6 +122,12 @@ impl ResponseExt for Result<Response, reqwest::Error> {
             }),
         }
     }
+
+    async fn drain_body(self) {
+        if let Ok(resp) = self {
+            resp.drain_body().await;
+        }
+    }
 }
 
 async fn parse_json_error(resp: Response) -> BackendDetail {

objectstore-service/src/backend/s3_compatible.rs

@@ -179,6 +179,7 @@ where
 
         if response.status() == StatusCode::NOT_FOUND {
             objectstore_log::debug!("Object not found");
+            response.drain_body().await;
             return Ok(None);
         }
 
@@ -188,15 +189,14 @@ where
                 .get(reqwest::header::CONTENT_RANGE)
                 .and_then(|v| v.to_str().ok());
             let total = raw.and_then(ContentRange::parse_unsatisfiable_total);
-            match total {
-                Some(total) => return Err(Error::RangeNotSatisfiable { total }),
-                None => {
-                    return Err(Error::Generic {
-                        context: format!("S3: 416 response with invalid Content-Range: {raw:?}"),
-                        cause: None,
-                    });
-                }
-            }
+            let err = match total {
+                Some(total) => Error::RangeNotSatisfiable { total },
+                None => Error::generic(format!(
+                    "S3: 416 response with invalid Content-Range: {raw:?}"
+                )),
+            };
+            response.drain_body().await;
+            return Err(err);
         }
 
         let response = response.check_error("S3: failed to get object").await?;
@@ -263,7 +263,9 @@ where
             .send()
             .await
             .check_error("S3: update expiration time")
-            .await?;
+            .await?
+            .drain_body()
+            .await;
 
         Ok(())
     }
@@ -312,7 +314,9 @@ impl<T: TokenProvider> Backend for S3CompatibleBackend<T> {
             .send()
             .await
             .check_error("S3: failed to put object")
-            .await?;
+            .await?
+            .drain_body()
+            .await;
 
         Ok(())
     }
@@ -352,10 +356,17 @@ impl<T: TokenProvider> Backend for S3CompatibleBackend<T> {
             })?;
 
         // Do not error for objects that do not exist.
-        if response.status() != StatusCode::NOT_FOUND {
-            response.check_error("S3: failed to delete object").await?;
+        if response.status() == StatusCode::NOT_FOUND {
+            response.drain_body().await;
+            return Ok(());
         }
 
+        response
+            .check_error("S3: failed to delete object")
+            .await?
+            .drain_body()
+            .await;
+
         Ok(())
     }
 }