feat(rust-client): Add Multipart Upload API

clients/rust/Cargo.toml

@@ -12,6 +12,7 @@ publish = true
 
 [dependencies]
 async-compression = { version = "0.4.27", features = ["tokio", "zstd"] }
+base64 = { version = "0.22.1", optional = true }
 percent-encoding = { workspace = true }
 bytes = { workspace = true }
 futures-util = { workspace = true }
@@ -39,6 +40,13 @@ zstd = "0.13.3"
 
 [features]
 default = ["native-tls", "hickory-dns"]
+multipart = ["dep:base64"]
+
 rustls = ["reqwest/rustls"]
 native-tls = ["reqwest/native-tls"]
 hickory-dns = ["reqwest/hickory-dns"]
+
+[[test]]
+name = "multipart"
+path = "tests/multipart.rs"
+required-features = ["multipart"]

clients/rust/README.md

@@ -128,6 +128,89 @@ session.put("payload")
     .send().await?;
 ```
 
+### Multipart Upload API
+
+For large objects, use multipart uploads to upload parts concurrently with bounded
+parallelism.
+
+**Important:** unlike single-object uploads, multipart uploads do **not** auto-compress.
+The caller must pre-compress each part according to the compression set as part of the metadata
+when initiating the upload.
+
+```rust,ignore
+use futures_util::StreamExt as _;
+use futures_util::stream;
+use objectstore_client::Compression;
+
+let upload = session
+    .initiate_multipart_upload()
+    .key("my-large-object")
+    .compression(Compression::Zstd)
+    .send()
+    .await?;
+
+let parts: Vec<(Vec<u8>, u32)> = vec![
+    (zstd::encode_all(&part1_data[..], 0)?, 1),
+    (zstd::encode_all(&part2_data[..], 0)?, 2),
+];
+
+let results: Vec<_> = stream::iter(
+    parts
+        .into_iter()
+        .map(|(data, part_number)| upload.put(data, part_number, None)),
+)
+.buffer_unordered(8)
+.collect()
+.await;
+
+let mut done = Vec::new();
+let mut errors = Vec::new();
+for result in results {
+    match result {
+        Ok(part) => done.push(part),
+        Err(e) => errors.push(e),
+    }
+}
+
+if !errors.is_empty() {
+    // reupload failed parts...
+}
+
+let key = upload.complete(done).await?;
+// or
+upload.abort().await?;
+```
+
+You can also resume an in-progress multipart upload, e.g. after a process restart.
+
+```rust,ignore
+use futures_util::{StreamExt as _, TryStreamExt as _};
+use futures_util::stream;
+use objectstore_client::CompletePart;
+
+let upload = session.resume_multipart_upload("my-large-object", saved_upload_id)?;
+
+let existing = upload.list_parts().await?;
+let total_parts = 10;
+let uploaded: Vec<u32> = existing.iter().map(|p| p.part_number.get()).collect();
+let missing: Vec<u32> = (1..=total_parts)
+    .filter(|n| !uploaded.contains(n))
+    .collect();
+
+let mut done: Vec<_> = stream::iter(
+    missing
+        .into_iter()
+        .map(|part_number| upload.put(get_part_data(part_number), part_number, None)),
+)
+.buffer_unordered(8)
+.try_collect()
+.await?;
+
+done.extend(existing.into_iter().map(CompletePart::from));
+
+let key = upload.complete(done).await?;
+```
+
 ### Many API
 
 The Many API allows you to enqueue multiple requests that the client can execute using Objectstore's batch endpoint, minimizing network overhead.

clients/rust/src/client.rs

@@ -151,7 +151,7 @@ impl ClientBuilder {
 #[derive(Debug, Clone)]
 pub struct Usecase {
     name: Arc<str>,
-    compression: Compression,
+    compression: Option<Compression>,
     expiration_policy: ExpirationPolicy,
 }
 
@@ -160,7 +160,7 @@ impl Usecase {
     pub fn new(name: &str) -> Self {
         Self {
             name: name.into(),
-            compression: Compression::Zstd,
+            compression: Some(Compression::Zstd),
             expiration_policy: Default::default(),
         }
     }
@@ -173,18 +173,18 @@ impl Usecase {
 
     /// Returns the compression algorithm to use for operations within this usecase.
     #[inline]
-    pub fn compression(&self) -> Compression {
+    pub fn compression(&self) -> Option<Compression> {
         self.compression
     }
 
     /// Sets the compression algorithm to use for operations within this usecase.
     ///
     /// It's still possible to override this default on each operation's builder.
     ///
-    /// By default, [`Compression::Zstd`] is used.
-    pub fn with_compression(self, compression: Compression) -> Self {
+    /// By default, [`Compression::Zstd`] is used. Pass [`None`] to disable compression.
+    pub fn with_compression(self, compression: impl Into<Option<Compression>>) -> Self {
         Self {
-            compression,
+            compression: compression.into(),
             ..self
         }
     }
@@ -464,6 +464,41 @@ impl Session {
         url
     }
 
+    #[cfg(feature = "multipart")]
+    fn multipart_url(
+        &self,
+        suffix: Option<&'static str>,
+        object_key: Option<&str>,
+        query_pairs: Option<Vec<(&str, String)>>,
+    ) -> Url {
+        let mut url = self.client.service_url.clone();
+
+        // `path_segments_mut` can only error if the url is cannot-be-a-base,
+        // and we check that in `ClientBuilder::new`, therefore this will never panic.
+        let mut segments = url.path_segments_mut().unwrap();
+        segments
+            .push("v1")
+            .push(match suffix {
+                Some("parts") => "objects:multipart:parts",
+                Some("complete") => "objects:multipart:complete",
+                _ => "objects:multipart",
+            })
+            .push(&self.scope.usecase.name)
+            .push(&self.scope.scopes.as_api_path().to_string());
+        if let Some(object_key) = object_key.filter(|key| !key.is_empty()) {
+            segments.extend(object_key.split("/"));
+        }
+        drop(segments);
+        if let Some(query_pairs) = query_pairs {
+            let mut pairs = url.query_pairs_mut();
+            for (key, value) in query_pairs {
+                pairs.append_pair(key, &value);
+            }
+        }
+
+        url
+    }
+
     fn prepare_builder(&self, mut builder: RequestBuilder) -> crate::Result<RequestBuilder> {
         if let Some(token) = self.mint_token()? {
             builder = builder.header("x-os-auth", format!("Bearer {token}"));
@@ -493,6 +528,19 @@ impl Session {
         let builder = self.client.reqwest.post(url);
         self.prepare_builder(builder)
     }
+
+    #[cfg(feature = "multipart")]
+    pub(crate) fn multipart_request(
+        &self,
+        method: reqwest::Method,
+        action: Option<&'static str>,
+        object_key: Option<&str>,
+        query_pairs: Option<Vec<(&str, String)>>,
+    ) -> crate::Result<RequestBuilder> {
+        let url = self.multipart_url(action, object_key, query_pairs);
+        let builder = self.client.reqwest.request(method, url);
+        self.prepare_builder(builder)
+    }
 }
 
 #[cfg(test)]

clients/rust/src/error.rs

@@ -44,6 +44,20 @@ pub enum Error {
         /// The error message.
         message: String,
     },
+    /// Error when part number validation fails (must be >= 1).
+    #[error("invalid part number: {0}")]
+    InvalidPartNumber(u32),
+    /// Error when upload ID validation fails.
+    #[error(transparent)]
+    InvalidUploadId(#[from] objectstore_types::multipart::InvalidUploadId),
+    /// Error returned when attempting to complete a multipart upload.
+    #[error("multipart complete failed ({code}): {message}")]
+    MultipartComplete {
+        /// The error code or kind.
+        code: String,
+        /// The error message.
+        message: String,
+    },
 }
 
 /// A convenience alias that defaults our [`Error`] type.

clients/rust/src/get.rs

@@ -128,7 +128,12 @@ pub(crate) fn maybe_decompress(
     match (metadata.compression, decompress && !encoding_accepted) {
         (Some(Compression::Zstd), true) => {
             metadata.compression = None;
-            ReaderStream::new(ZstdDecoder::new(StreamReader::new(stream))).boxed()
+            let mut decoder = ZstdDecoder::new(StreamReader::new(stream));
+            // Multipart uploads with compression, when each part is compressed individually,
+            // will consist of multiple concatenated zstd frames.
+            // This allows the client to handle automatic decompression for these objects transparently.
+            decoder.multiple_members(true);
+            ReaderStream::new(decoder).boxed()
         }
         _ => stream,
     }
@@ -230,4 +235,22 @@ mod tests {
         assert_eq!(collect(out).await, payload);
         assert_eq!(metadata.compression, None);
     }
+
+    #[tokio::test]
+    async fn zstd_concatenated_frames_decompress() {
+        let payload1 = b"hello ";
+        let payload2 = b"world";
+        let compressed1 = collect(compressed_zstd_stream(payload1)).await;
+        let compressed2 = collect(compressed_zstd_stream(payload2)).await;
+        let stream = futures_util::stream::iter([
+            Ok::<_, std::io::Error>(bytes::Bytes::from(compressed1)),
+            Ok::<_, std::io::Error>(bytes::Bytes::from(compressed2)),
+        ])
+        .boxed();
+
+        let mut metadata = zstd_metadata();
+        let out = maybe_decompress(stream, &mut metadata, true, &[]);
+        assert_eq!(collect(out).await, b"hello world");
+        assert_eq!(metadata.compression, None);
+    }
 }

clients/rust/src/lib.rs

@@ -10,6 +10,8 @@ mod get;
 mod head;
 mod key;
 mod many;
+#[cfg(feature = "multipart")]
+mod multipart;
 mod put;
 pub mod utils;
 
@@ -23,4 +25,6 @@ pub use get::*;
 pub use head::*;
 pub use key::*;
 pub use many::*;
+#[cfg(feature = "multipart")]
+pub use multipart::*;
 pub use put::*;