feat(rs-client): Add automatic resumable uploads

clients/rust/Cargo.toml

@@ -21,31 +21,35 @@ infer = { workspace = true }
 jsonwebtoken = { workspace = true }
 multer = { workspace = true }
 objectstore-types = { workspace = true }
+rand = { workspace = true }
 # Pinned below workspace version for MSRV compatibility
 reqwest = { version = "0.13.1", default-features = false, features = ["charset", "http2", "system-proxy", "json", "stream", "multipart", "query"] }
 sentry-core = { version = ">=0.41", default-features = false, features = ["client"] }
 serde = { workspace = true }
+tempfile = { workspace = true }
 thiserror = { workspace = true }
-tokio = { version = "1.45.0", default-features = false, features = ["fs"] }
+tokio = { version = "1.45.0", default-features = false, features = ["fs", "io-util", "rt", "time"] }
 tokio-util = { workspace = true, features = ["io"] }
+tracing = { workspace = true, optional = true }
 url = { workspace = true }
 zstd-safe = { workspace = true }
 
 [dev-dependencies]
 anyhow = { workspace = true }
+axum = { workspace = true }
 flate2 = { workspace = true }
 futures-util = { workspace = true }
 http-body-util = { workspace = true }
 jsonwebtoken = { workspace = true }
 objectstore-test = { workspace = true }
 serde = { workspace = true }
-tempfile = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt"] }
 zstd = { workspace = true }
 
 [features]
 default = ["native-tls", "hickory-dns"]
 resumable-upload-api = []
+tracing = ["dep:tracing"]
 
 rustls = ["reqwest/rustls"]
 native-tls = ["reqwest/native-tls"]

clients/rust/README.md

@@ -151,20 +151,29 @@ session.put("payload")
     .send().await?;
 ```
 
-### Resumable Upload API
+### Resumable Uploads
+
+The Objectstore server supports an additional API to perform PUT operations in a way
+that allows resuming. After a failed request, the upload can resume from the
+server's persisted offset.
+
+Large PUTs automatically use resumable uploads.
+See [`PutBuilder`] for eligibility, configuration, and recovery behavior.
+
+### Manual Resumable Uploads
 
 > **Feature flag required:** Enable `resumable-upload-api` to use this API.
 >
 > ```toml
 > objectstore-client = { version = "...", features = ["resumable-upload-api"] }
 > ```
 
-The resumable upload API allows you to upload an object across multiple requests.
-It is particularly suitable for large objects, where restarting an upload from
-scratch would be expensive.
+Use this API only if you need granular control over resumable upload operations.
+Most users should not need this, and instead rely on automatic behavior implemented by [`PutBuilder`].
+If you need more controls to customize that behavior, please open an issue.
+
 It's recommended to always try to upload the whole object in a single request if
 possible, as that's always the more efficient approach.
-If the request fails midway, it will be possible to resume it from the persisted offset.
 
 **Important:** resumable uploads do not automatically compress chunk contents. The `compression`
 setting only records how the object is encoded; the caller must compress the payload accordingly.
@@ -176,7 +185,8 @@ If a non-final chunk is larger than the granularity but its size is not a multip
 server may persist only its aligned prefix.
 
 ```rust,no_run
-#![cfg(feature = "resumable-upload-api")]
+# #[cfg(feature = "resumable-upload-api")]
+# mod example {
 
 use bytes::Bytes;
 use anyhow::{Context as _, Result};
@@ -189,7 +199,7 @@ async fn upload_large_object(session: &Session, object: Bytes) -> Result<()> {
         .content_type("application/octet-stream")
         .send()
         .await?
-        // Fall back to a regular [`Session:;put`].
+        // Fall back to a regular Session::put.
         .context("resumable upload declined for this object")?;
 
     let mut offset = 0;
@@ -208,6 +218,8 @@ async fn upload_large_object(session: &Session, object: Bytes) -> Result<()> {
         };
     }
 }
+# }
+# fn main() {}
 ```
 
 ### Many API
@@ -311,6 +323,11 @@ let client = Client::builder("http://localhost:8888/")
 To create a token with narrower permissions or a different expiry, use the
 methods on the token request returned from `create_token`.
 
+## Feature flags
+
+- `tracing`: enables diagnostic tracing events and spans
+- `resumable-upload-api`: enables the manual Resumable Uploads API (see above)
+
 ## Configuration
 
 In production, store the [`Client`] and [`Usecase`] in a `static` and reuse them.

clients/rust/src/client.rs

@@ -1,4 +1,5 @@
 use std::io;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -21,6 +22,7 @@ struct ClientBuilderInner {
     propagate_traces: bool,
     reqwest_builder: reqwest::ClientBuilder,
     token: Option<TokenProvider>,
+    spool_dir: Option<PathBuf>,
 }
 
 impl ClientBuilderInner {
@@ -71,6 +73,7 @@ impl ClientBuilder {
             propagate_traces: false,
             reqwest_builder,
             token: None,
+            spool_dir: None,
         }))
     }
 
@@ -124,6 +127,17 @@ impl ClientBuilder {
         Self(Ok(inner))
     }
 
+    /// Sets the directory for temporary files used to prepare compressed resumable uploads.
+    ///
+    /// By default, the system temporary directory is used. The directory must already exist.
+    /// If a temporary file cannot be created, uploads fall back to direct streaming.
+    /// Temporary storage may outlive the upload while HTTP readers remain open.
+    pub fn spool_dir(self, path: impl Into<PathBuf>) -> Self {
+        let Ok(mut inner) = self.0 else { return self };
+        inner.spool_dir = Some(path.into());
+        Self(Ok(inner))
+    }
+
     /// Returns a [`Client`] that uses this [`ClientBuilder`] configuration.
     ///
     /// # Errors
@@ -141,6 +155,7 @@ impl ClientBuilder {
                 service_url: inner.service_url,
                 propagate_traces: inner.propagate_traces,
                 token: inner.token,
+                spool_dir: inner.spool_dir,
             }),
         })
     }
@@ -161,6 +176,7 @@ pub struct Usecase {
     name: UsecaseName,
     compression: Option<Compression>,
     expiration_policy: ExpirationPolicy,
+    resumable_threshold_bytes: Option<u64>,
 }
 
 impl Usecase {
@@ -203,6 +219,7 @@ impl Usecase {
             compression: Some(Compression::Zstd),
             // `ExpirationPolicy::default()` isn't callable in a const fn.
             expiration_policy: ExpirationPolicy::Manual,
+            resumable_threshold_bytes: Some(256 * 1024 * 1024),
         }
     }
 
@@ -249,6 +266,21 @@ impl Usecase {
         }
     }
 
+    /// Returns the automatic resumable upload threshold, or [`None`] if disabled.
+    pub fn resumable_threshold_bytes(&self) -> Option<u64> {
+        self.resumable_threshold_bytes
+    }
+
+    /// Sets the minimum remaining input size for automatic resumable uploads.
+    ///
+    /// Defaults to 256 MiB, measured before client compression. Pass [`None`] to disable or
+    /// zero to include every supported payload. Individual uploads can override this using
+    /// [`crate::PutBuilder::resumable_threshold_bytes`]. See [`crate::PutBuilder`] for eligibility.
+    pub fn with_resumable_threshold_bytes(mut self, threshold: impl Into<Option<u64>>) -> Self {
+        self.resumable_threshold_bytes = threshold.into();
+        self
+    }
+
     /// Creates a new custom [`Scope`].
     ///
     /// Add parts to it using [`Scope::push`].
@@ -340,12 +372,19 @@ impl Scope {
 
 #[derive(Debug)]
 pub(crate) struct ClientInner {
+    spool_dir: Option<PathBuf>,
     reqwest: reqwest::Client,
     service_url: Url,
     propagate_traces: bool,
     token: Option<TokenProvider>,
 }
 
+impl ClientInner {
+    pub fn spool_dir(&self) -> Option<&Path> {
+        self.spool_dir.as_deref()
+    }
+}
+
 /// A client for Objectstore. Use [`Client::builder`] to configure and construct a Client.
 ///
 /// To perform CRUD operations, one has to create a Client, and then scope it to a [`Usecase`]
@@ -446,6 +485,13 @@ impl Client {
         ClientBuilder::new(service_url)
     }
 
+    /// Returns the configured spool directory.
+    ///
+    /// `None` means the system temporary directory is used.
+    pub fn spool_dir(&self) -> Option<&Path> {
+        self.inner.spool_dir()
+    }
+
     /// Creates a session for the given scope using this client.
     ///
     /// # Errors

clients/rust/src/error.rs

@@ -57,18 +57,15 @@ pub enum Error {
         message: String,
     },
     /// A resumable upload session expired, was canceled, or could otherwise not be found.
-    #[cfg(feature = "resumable-upload-api")]
     #[error("resumable upload session is not available")]
     ResumableUploadUnavailable,
     /// A resumable chunk starts at a different offset than the server currently holds.
-    #[cfg(feature = "resumable-upload-api")]
     #[error("upload offset mismatch (server holds {offset} bytes)")]
     UploadOffsetMismatch {
         /// The server's authoritative upload offset in bytes.
         offset: u64,
     },
     /// A non-final chunk is shorter than the upload granularity.
-    #[cfg(feature = "resumable-upload-api")]
     #[error(
         "non-final chunk {chunk_length} is smaller than upload granularity {upload_granularity}"
     )]

clients/rust/src/lib.rs

@@ -13,7 +13,6 @@ mod key;
 mod many;
 mod put;
 mod response;
-#[cfg(feature = "resumable-upload-api")]
 mod resumable;
 pub mod utils;
 
@@ -30,5 +29,9 @@ pub use head::*;
 pub use key::*;
 pub use many::*;
 pub use put::*;
+
 #[cfg(feature = "resumable-upload-api")]
-pub use resumable::*;
+pub use resumable::{
+    CancelUploadBuilder, CreateResumableUploadBuilder, PutChunkBuilder, ResumableUpload,
+    SessionToken, UploadProgress, UploadProgressBuilder,
+};

clients/rust/src/many.rs

@@ -117,6 +117,7 @@ impl From<PutBuilder> for BatchOperation {
             compression,
             body,
             session: _session,
+            resumable_threshold_bytes: _,
         } = value;
         BatchOperation::Insert {
             key,
@@ -660,6 +661,8 @@ async fn execute_individual(op: BatchOperation, session: &Session) -> OperationR
         } => {
             let error_key = key.clone().unwrap_or_else(|| "<unknown>".to_owned());
             let put = PutBuilder {
+                // Automatic resumables are currently limited to standalone uploads.
+                resumable_threshold_bytes: None,
                 session: session.clone(),
                 metadata,
                 compression,