feat(rs-client): Add automatic resumable uploads

clients/rust/Cargo.toml

@@ -21,25 +21,27 @@ 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"] }
 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 }
 

clients/rust/README.md

@@ -151,6 +151,13 @@ session.put("payload")
     .send().await?;
 ```
 
+### Automatic Resumable Uploads
+
+Large PUTs automatically use resumable uploads, resending only what the server hasn't
+stored after a failed request.
+See [`PutBuilder`] for eligibility, configuration, and recovery behavior, and
+[`Session::put_path`] for the path stability requirement.
+
 ### Resumable Upload API
 
 > **Feature flag required:** Enable `resumable-upload-api` to use this API.
@@ -176,7 +183,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 +197,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 +216,8 @@ async fn upload_large_object(session: &Session, object: Bytes) -> Result<()> {
         };
     }
 }
+# }
+# fn main() {}
 ```
 
 ### Many API

clients/rust/src/client.rs

@@ -1,4 +1,5 @@
 use std::io;
+use std::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: Arc<UsecaseName>,
     compression: Option<Compression>,
     expiration_policy: ExpirationPolicy,
+    resumable_threshold_bytes: Option<u64>,
 }
 
 impl Usecase {
@@ -177,6 +193,7 @@ impl Usecase {
             name: Arc::new(UsecaseName::new(name)?),
             compression: Some(Compression::Zstd),
             expiration_policy: ExpirationPolicy::default(),
+            resumable_threshold_bytes: Some(256 * 1024 * 1024),
         })
     }
 
@@ -234,6 +251,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`].
@@ -325,6 +357,7 @@ impl Scope {
 
 #[derive(Debug)]
 pub(crate) struct ClientInner {
+    pub(crate) spool_dir: Option<PathBuf>,
     reqwest: reqwest::Client,
     service_url: Url,
     propagate_traces: bool,

clients/rust/src/error.rs

@@ -56,19 +56,21 @@ pub enum Error {
         /// The error message.
         message: String,
     },
+    /// An automatic resumable upload failed after session creation.
+    ///
+    /// The source contains the original failure.
+    #[error("upload failed: {0}")]
+    UploadFailed(#[source] Box<dyn std::error::Error + Send + Sync>),
     /// 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;
 

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,

clients/rust/src/put.rs

@@ -18,6 +18,9 @@ pub use objectstore_types::metadata::{Compression, ExpirationPolicy};
 use crate::response::ResponseExt as _;
 use crate::{ClientStream, ObjectKey, Session};
 
+mod automatic;
+mod payload;
+
 /// The response returned from the service after uploading an object.
 #[derive(Debug, Deserialize)]
 pub struct PutResponse {
@@ -75,6 +78,7 @@ impl Session {
                 .compression()
                 .map(CompressionMode::Compress),
             key: None,
+            resumable_threshold_bytes: self.scope.usecase().resumable_threshold_bytes(),
             body,
         }
     }
@@ -85,11 +89,15 @@ impl Session {
     }
 
     /// Creates or replaces an object using a streaming payload.
+    ///
+    /// Streams use direct uploads, regardless of the resumable threshold.
     pub fn put_stream(&self, body: ClientStream) -> PutBuilder {
         self.put_body(PutBody::Stream(body))
     }
 
     /// Creates or replaces an object using an [`AsyncRead`] payload.
+    ///
+    /// Readers use direct uploads. Use [`Self::put_path`] for automatic resumable file uploads.
     pub fn put_read<R>(&self, body: R) -> PutBuilder
     where
         R: AsyncRead + Send + Sync + 'static,
@@ -100,6 +108,8 @@ impl Session {
 
     /// Creates or replaces an object using the contents of an opened file.
     ///
+    /// Uploads from the current position using a direct request, regardless of the resumable threshold.
+    ///
     /// The file descriptor is held open from the moment this method is called until the
     /// upload completes. When enqueueing many files via [`Session::many`], prefer
     /// [`put_path`](Session::put_path) instead: it defers opening the file until just before
@@ -111,6 +121,10 @@ impl Session {
 
     /// Creates or replaces an object using the contents of the file at `path`.
     ///
+    /// For resumable uploads, the path must stay readable and must not be modified, renamed,
+    /// removed, or replaced until the upload completes. Each attempt opens a new handle.
+    /// Handles may remain open until pending HTTP requests release their readers.
+    ///
     /// Unlike [`put_file`](Session::put_file), this method defers opening the file until the
     /// request is actually sent. When enqueueing many file uploads via [`Session::many`], this
     /// ensures that file descriptors are opened only within the active concurrency window,
@@ -125,16 +139,43 @@ impl Session {
 }
 
 /// A [`put`](Session::put) request builder.
+///
+/// # Automatic resumable uploads
+///
+/// Large buffers and paths to regular files use resumable uploads automatically,
+/// recovering from interrupted requests. Open files, streams, readers, and uploads queued through
+/// [`Session::many`] do not use resumables. No feature flag is required.
+///
+/// Configure this with [`Self::resumable_threshold_bytes`] and [`crate::ClientBuilder::spool_dir`].
 #[derive(Debug)]
 pub struct PutBuilder {
     pub(crate) session: Session,
     pub(crate) metadata: Metadata,
     pub(crate) compression: Option<CompressionMode>,
     pub(crate) key: Option<ObjectKey>,
     pub(crate) body: PutBody,
+    pub(crate) resumable_threshold_bytes: Option<u64>,
 }
 
 impl PutBuilder {
+    /// Sets the minimum remaining input size for automatic resumable uploads.
+    ///
+    /// Measures remaining bytes before client compression. Inherits the
+    /// [`Usecase` threshold](crate::Usecase::with_resumable_threshold_bytes) unless overridden.
+    /// Pass [`None`] to disable or zero to include every supported payload.
+    /// See [`PutBuilder`] for eligibility.
+    ///
+    /// ```no_run
+    /// # async fn example(session: objectstore_client::Session) -> objectstore_client::Result<()> {
+    /// session.put("payload").resumable_threshold_bytes(None).send().await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn resumable_threshold_bytes(mut self, threshold: impl Into<Option<u64>>) -> Self {
+        self.resumable_threshold_bytes = threshold.into();
+        self
+    }
+
     /// Sets an explicit object key.
     ///
     /// If a key is specified, the object will be stored under that key. Otherwise, the Objectstore
@@ -146,7 +187,7 @@ impl PutBuilder {
 
     /// Sets an explicit compression algorithm to be used for this payload.
     ///
-    /// The client compresses the payload while uploading it and records the algorithm in the
+    /// The client compresses the payload and records the algorithm in the
     /// object's metadata. [`None`] should be used if no compression should be performed by the
     /// client, either because the payload is uncompressible (such as a media format), or if the
     /// compression should not be recorded for this object.
@@ -312,12 +353,37 @@ impl PutBuilder {
             .request(method, self.key.as_deref().unwrap_or_default())?;
 
         self.metadata.compression = self.compression.map(CompressionMode::compression);
-        let body = encode_body(self.body, self.compression).await?;
-
-        builder = builder.headers(self.metadata.to_headers("")?);
+        let headers = self.metadata.to_headers("")?;
+        let prepared = payload::prepare(
+            self.body,
+            self.compression,
+            self.resumable_threshold_bytes,
+            self.session.client.spool_dir.clone(),
+        )
+        .await?;
+        if let payload::PreparedBody::Replayable(body) = &prepared {
+            let create = crate::resumable::implementation::CreateResumableUploadBuilder::new(
+                self.session.clone(),
+                body.len(),
+                self.key.clone(),
+                self.metadata,
+            );
+            if let Ok(Some(upload)) = create.send().await {
+                return automatic::upload(body, upload).await;
+            }
+        }
 
-        let response = builder.body(body).send().await?;
-        Ok(response.error_for_status_and_drain().await?.json().await?)
+        // All direct uploads, including preparation and session-creation fallbacks, converge here.
+        prepared
+            .with_body(self.compression, |body, content_length| async move {
+                builder = builder.headers(headers);
+                if let Some(length) = content_length {
+                    builder = builder.header(reqwest::header::CONTENT_LENGTH, length);
+                }
+                let response = builder.body(body).send().await?;
+                Ok(response.error_for_status_and_drain().await?.json().await?)
+            })
+            .await
     }
 }