diff --git a/Cargo.lock b/Cargo.lock index 6bcc5d59..04078717 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2822,6 +2822,7 @@ version = "0.2.10" dependencies = [ "anyhow", "async-compression", + "axum", "bytes", "futures-util", "http-body-util", @@ -2830,6 +2831,7 @@ dependencies = [ "multer", "objectstore-test", "objectstore-types", + "rand 0.10.2", "reqwest 0.13.4", "sentry-core", "serde", diff --git a/clients/rust/Cargo.toml b/clients/rust/Cargo.toml index 85c3bc6b..00db68fd 100644 --- a/clients/rust/Cargo.toml +++ b/clients/rust/Cargo.toml @@ -21,24 +21,26 @@ 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", "sync", "time"] } tokio-util = { workspace = true, features = ["io"] } url = { workspace = true } zstd-safe = { workspace = true } [dev-dependencies] anyhow = { workspace = true } +axum = { 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 } diff --git a/clients/rust/README.md b/clients/rust/README.md index 99b4987e..b3f2c57d 100644 --- a/clients/rust/README.md +++ b/clients/rust/README.md @@ -151,6 +151,13 @@ session.put("payload") .send().await?; ``` +### Automatic Resumable Uploads + +Large uploads automatically use resumable uploads and resend only what the server hasn't +stored after a failed write. No feature flag is required. + +See [`PutBuilder`] for eligibility, configuration, and recovery behavior. + ### 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 diff --git a/clients/rust/src/client.rs b/clients/rust/src/client.rs index b66cdaf0..6f0fbb26 100644 --- a/clients/rust/src/client.rs +++ b/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, + spool_dir: Option, } impl ClientBuilderInner { @@ -71,6 +73,7 @@ impl ClientBuilder { propagate_traces: false, reqwest_builder, token: None, + spool_dir: None, })) } @@ -124,6 +127,16 @@ 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. + pub fn spool_dir(self, path: impl Into) -> 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 +154,7 @@ impl ClientBuilder { service_url: inner.service_url, propagate_traces: inner.propagate_traces, token: inner.token, + spool_dir: inner.spool_dir, }), }) } @@ -161,6 +175,7 @@ pub struct Usecase { name: Arc, compression: Option, expiration_policy: ExpirationPolicy, + resumable_threshold_bytes: Option, } impl Usecase { @@ -177,6 +192,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 +250,21 @@ impl Usecase { } } + /// Returns the minimum remaining input size for automatic resumable uploads. + pub fn resumable_threshold_bytes(&self) -> Option { + self.resumable_threshold_bytes + } + + /// Sets the minimum remaining input size for automatic resumable uploads. + /// + /// Defaults to 256 MiB. Pass [`None`] to disable automatic resumables 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>) -> Self { + self.resumable_threshold_bytes = threshold.into(); + self + } + /// Creates a new custom [`Scope`]. /// /// Add parts to it using [`Scope::push`]. @@ -325,6 +356,7 @@ impl Scope { #[derive(Debug)] pub(crate) struct ClientInner { + pub(crate) spool_dir: Option, reqwest: reqwest::Client, service_url: Url, propagate_traces: bool, diff --git a/clients/rust/src/error.rs b/clients/rust/src/error.rs index e1985144..477bc6d5 100644 --- a/clients/rust/src/error.rs +++ b/clients/rust/src/error.rs @@ -56,19 +56,22 @@ pub enum Error { /// The error message. message: String, }, + /// An automatic resumable upload failed after session creation. + /// + /// The source preserves the failure that caused the upload to stop. Cancellation errors + /// are ignored so they cannot replace that failure. + #[error("upload failed: {0}")] + UploadFailed(#[source] Box), /// 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}" )] diff --git a/clients/rust/src/lib.rs b/clients/rust/src/lib.rs index 2924b257..42710e7e 100644 --- a/clients/rust/src/lib.rs +++ b/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; diff --git a/clients/rust/src/many.rs b/clients/rust/src/many.rs index 0f9b12a7..bdc2a590 100644 --- a/clients/rust/src/many.rs +++ b/clients/rust/src/many.rs @@ -117,6 +117,7 @@ impl From for BatchOperation { compression, body, session: _session, + resumable_threshold_bytes: _, } = value; BatchOperation::Insert { key, @@ -640,6 +641,8 @@ async fn execute_individual(op: BatchOperation, session: &Session) -> OperationR } => { let error_key = key.clone().unwrap_or_else(|| "".to_owned()); let put = PutBuilder { + // Automatic resumables are currently limited to standalone uploads. + resumable_threshold_bytes: None, session: session.clone(), metadata, compression, diff --git a/clients/rust/src/put.rs b/clients/rust/src/put.rs index 6e4e5ba8..24953663 100644 --- a/clients/rust/src/put.rs +++ b/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_file`] for automatic resumable file uploads. pub fn put_read(&self, body: R) -> PutBuilder where R: AsyncRead + Send + Sync + 'static, @@ -125,6 +133,20 @@ impl Session { } /// A [`put`](Session::put) request builder. +/// +/// # Automatic resumable uploads +/// +/// Buffers and regular files (including paths) use resumable uploads at 256 MiB by default, +/// measured from the current position before client compression. Override or disable this with +/// [`Self::resumable_threshold_bytes`]. Streams, generic readers, and files of unknown size use +/// direct uploads. Files must remain unchanged during upload. +/// +/// Input is reused unless compression is needed, in which case the compressed payload is spooled +/// to a temporary file. Use [`crate::ClientBuilder::spool_dir`] to choose its directory. +/// Transient failures and writes making no progress are retried up to twice. Acknowledged +/// progress resets the retry count. +/// +/// No feature flag is required. Uploads queued through [`Session::many`] do not use resumables. #[derive(Debug)] pub struct PutBuilder { pub(crate) session: Session, @@ -132,9 +154,26 @@ pub struct PutBuilder { pub(crate) compression: Option, pub(crate) key: Option, pub(crate) body: PutBody, + pub(crate) resumable_threshold_bytes: Option, } impl PutBuilder { + /// Sets the minimum remaining input size for automatic resumable uploads. + /// + /// Inherits [`crate::Usecase::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>) -> 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 +185,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. @@ -323,17 +362,42 @@ impl PutBuilder { None => reqwest::Method::POST, }; - let mut builder = self - .session - .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 mut prepared = payload::prepare( + self.body, + self.compression, + self.resumable_threshold_bytes, + self.session.client.spool_dir.clone(), + ) + .await?; + if let payload::PreparedBody::Replay(body) = &mut prepared { + let create = crate::resumable::CreateResumableUploadBuilder { + session: self.session.clone(), + total_length: body.len(), + key: self.key.clone(), + metadata: 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 { + // Mint authentication after preparation, which may outlast a token's validity. + let mut builder = self + .session + .request(method, self.key.as_deref().unwrap_or_default())? + .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 } } diff --git a/clients/rust/src/put/automatic.rs b/clients/rust/src/put/automatic.rs new file mode 100644 index 00000000..55d71f6c --- /dev/null +++ b/clients/rust/src/put/automatic.rs @@ -0,0 +1,95 @@ +//! Write prepared payloads, recovering from authoritative server progress. + +use std::io; +use std::time::Duration; + +use super::PutResponse; +use super::payload::ReplayBody; +use crate::Error; +use crate::resumable::{ResumableUpload, UploadProgress}; + +pub(super) async fn upload( + body: &mut ReplayBody, + handle: ResumableUpload, +) -> crate::Result { + match upload_inner(body, &handle).await { + Ok(()) => Ok(PutResponse { + key: handle.key().to_owned(), + }), + Err(error) => { + let _ = handle.cancel().send().await; + Err(Error::UploadFailed(Box::new(error))) + } + } +} + +async fn upload_inner(body: &mut ReplayBody, handle: &ResumableUpload) -> crate::Result<()> { + let mut offset = 0; + let mut retries = 0; + loop { + let remaining = body.len() - offset; + let write = body + .with_stream(offset, |stream| async { + handle.put_stream(offset, remaining, stream).send().await + }) + .await; + let progress = match authoritative_progress(write) { + Ok(UploadProgress::Incomplete { offset: next }) if next == offset => { + let error = io::Error::other("upload made no progress").into(); + recover(handle, &mut retries, error).await? + } + Err(error) if is_transient(&error) => recover(handle, &mut retries, error).await?, + other => other?, + }; + match progress { + UploadProgress::Complete => return Ok(()), + UploadProgress::Incomplete { offset: next } => { + if next < offset || next > body.len() { + return Err(Error::MalformedResponse("invalid upload offset".into())); + } + if next > offset { + retries = 0; + } + offset = next; + } + } + } +} + +/// Writes and failed probes share this budget; only acknowledged progress resets it. +async fn recover( + handle: &ResumableUpload, + retries: &mut u32, + mut failure: Error, +) -> crate::Result { + while *retries < 2 { + let delay = + Duration::from_secs(2 << *retries) + Duration::from_secs_f64(rand::random::()); + tokio::time::sleep(delay).await; + *retries += 1; + match authoritative_progress(handle.progress().send().await) { + Err(error) if is_transient(&error) => failure = error, + result => return result, + } + } + Err(failure) +} + +fn authoritative_progress(result: crate::Result) -> crate::Result { + match result { + Err(Error::UploadOffsetMismatch { offset }) => Ok(UploadProgress::Incomplete { offset }), + other => other, + } +} + +fn is_transient(error: &Error) -> bool { + match error { + Error::Reqwest(error) => match error.status() { + Some(status) => matches!(status.as_u16(), 408 | 429 | 502 | 503 | 504), + None => { + error.is_timeout() || error.is_connect() || error.is_request() || error.is_body() + } + }, + _ => false, + } +} diff --git a/clients/rust/src/put/payload.rs b/clients/rust/src/put/payload.rs new file mode 100644 index 00000000..d997324e --- /dev/null +++ b/clients/rust/src/put/payload.rs @@ -0,0 +1,274 @@ +//! Prepare replayable wire bytes and give each file request exclusive reader ownership. + +use std::io::{self, SeekFrom}; +use std::path::PathBuf; + +use bytes::Bytes; +use futures_util::{StreamExt, stream}; +use tokio::fs::File; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; +use tokio::sync::{mpsc, oneshot}; + +use super::{CompressionMode, PutBody}; +use crate::ClientStream; + +pub(super) enum PreparedBody { + Direct(PutBody), + Replay(ReplayBody), +} + +impl PreparedBody { + /// Supplies a direct request with either the original input or prepared wire bytes. + pub(super) async fn with_body( + self, + compression: Option, + request: impl FnOnce(reqwest::Body, Option) -> F, + ) -> crate::Result + where + F: std::future::Future>, + { + match self { + Self::Direct(body) => request(super::encode_body(body, compression).await?, None).await, + Self::Replay(mut body) => { + let length = body.len(); + body.with_stream(0, |stream| { + request(reqwest::Body::wrap_stream(stream), Some(length)) + }) + .await + } + } + } +} + +pub(super) enum ReplayBody { + Buffer(Bytes), + File { file: File, start: u64, length: u64 }, +} + +pub(super) async fn prepare( + body: PutBody, + compression: Option, + threshold: Option, + spool_dir: Option, +) -> crate::Result { + let Some(threshold) = threshold else { + return Ok(PreparedBody::Direct(body)); + }; + let body = match body { + PutBody::Path(path) => PutBody::File(File::open(path).await?), + body => body, + }; + let mut replay = match body { + PutBody::Buffer(bytes) => ReplayBody::Buffer(bytes), + PutBody::File(mut file) => match regular_file_span(&mut file).await { + Some((start, length)) => ReplayBody::File { + file, + start, + length, + }, + None => return Ok(PreparedBody::Direct(PutBody::File(file))), + }, + body => return Ok(PreparedBody::Direct(body)), + }; + if replay.len() < threshold { + return Ok(PreparedBody::Direct(replay.into_put_body())); + } + + if let Some(CompressionMode::Compress(crate::Compression::Zstd)) = compression { + let temporary = tokio::task::spawn_blocking(move || match spool_dir { + Some(dir) => tempfile::tempfile_in(dir), + None => tempfile::tempfile(), + }) + .await + .map_err(io::Error::other)?; + let Ok(temporary) = temporary else { + // Creation has not consumed input, so direct streaming is still possible. + return Ok(PreparedBody::Direct(replay.into_put_body())); + }; + let mut temporary = File::from_std(temporary); + let reader: Box = match replay { + ReplayBody::Buffer(bytes) => Box::new(io::Cursor::new(bytes)), + ReplayBody::File { file, .. } => Box::new(file), + }; + let mut encoder = super::ZstdEncoder::new(super::BufReader::new(reader)); + let length = tokio::io::copy(&mut encoder, &mut temporary).await?; + temporary.flush().await?; + temporary.rewind().await?; + replay = ReplayBody::File { + file: temporary, + start: 0, + length, + }; + } + + Ok(PreparedBody::Replay(replay)) +} + +/// Returns the current position and remaining length of a regular file. +async fn regular_file_span(file: &mut File) -> Option<(u64, u64)> { + file.flush().await.ok()?; + let metadata = file.metadata().await.ok()?; + if !metadata.is_file() { + return None; + } + let start = file.stream_position().await.ok()?; + Some((start, metadata.len().saturating_sub(start))) +} + +impl ReplayBody { + /// Returns the unprepared input for a direct upload, positioned where it started. + fn into_put_body(self) -> PutBody { + match self { + Self::Buffer(bytes) => PutBody::Buffer(bytes), + Self::File { file, .. } => PutBody::File(file), + } + } + + pub(super) fn len(&self) -> u64 { + match self { + Self::Buffer(bytes) => bytes.len() as u64, + Self::File { length, .. } => *length, + } + } + + /// Runs a request with a suffix, then stops its reader before returning the result. + /// + /// The request body may outlive the response. Keeping reader ownership here makes local + /// read errors explicit and prevents an old request from reading after a retry seeks. + pub(super) async fn with_stream( + &mut self, + offset: u64, + request: impl FnOnce(ClientStream) -> F, + ) -> crate::Result + where + F: std::future::Future>, + { + match self { + Self::Buffer(bytes) => { + let bytes = bytes.slice(offset as usize..); + // Streaming bodies prevent independent replays by HTTP middleware. + request(stream::once(async move { Ok(bytes) }).boxed()).await + } + Self::File { + file, + start, + length, + } => { + let (stream, reader) = + FileReader::start(file, *start + offset, *length - offset).await?; + let response = request(stream).await; + // Read failures are returned directly, independently of reqwest's error wrapping. + reader.finish().await?; + response + } + } + } +} + +/// Explicitly stoppable because HTTP may retain the request body after returning a response. +/// Dropping this handle signals cancellation; finishing also waits for pending disk IO. +struct FileReader { + stop: oneshot::Sender<()>, + task: tokio::task::JoinHandle>, +} + +impl FileReader { + async fn start(file: &File, offset: u64, length: u64) -> io::Result<(ClientStream, Self)> { + // Duplicate the open handle, never reopen a path. Clones share the OS cursor, so + // with_stream must finish this reader before starting another request. + let mut file = file.try_clone().await?; + file.seek(SeekFrom::Start(offset)).await?; + let (send, receive) = mpsc::channel(1); + let (stop, stopped) = oneshot::channel(); + let task = tokio::spawn(async move { + let result = { + let read = forward_file(&mut file, length, send); + futures_util::pin_mut!(read, stopped); + match futures_util::future::select(stopped, read).await { + futures_util::future::Either::Left(_) => Ok(()), + futures_util::future::Either::Right((result, _)) => result, + } + }; + // An async read may leave blocking IO running after cancellation. + drop(file.into_std().await); + result + }); + let stream = stream::unfold(receive, |mut receive| async { + receive.recv().await.map(|bytes| (bytes, receive)) + }) + .boxed(); + Ok((stream, Self { stop, task })) + } + + async fn finish(self) -> io::Result<()> { + drop(self.stop); + self.task.await.map_err(io::Error::other)? + } +} + +async fn forward_file( + file: &mut File, + length: u64, + send: mpsc::Sender>, +) -> io::Result<()> { + let mut remaining = length; + while remaining > 0 { + let Ok(permit) = send.reserve().await else { + break; + }; + let mut bytes = vec![0; remaining.min(64 * 1024) as usize]; + if let Err(error) = file.read_exact(&mut bytes).await { + // Fail the HTTP body too, but keep the original error for the caller. + permit.send(Err(io::Error::other(error.to_string()))); + return Err(error); + } + remaining -= bytes.len() as u64; + permit.send(Ok(Bytes::from(bytes))); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use futures_util::TryStreamExt; + + use super::*; + + #[test] + fn stopping_reader_waits_for_pending_read_even_if_http_keeps_body() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .max_blocking_threads(1) + .build() + .unwrap(); + runtime.block_on(async { + let data: Vec = (0..200_000).map(|n| (n % 251) as u8).collect(); + let mut file = tempfile::tempfile().unwrap(); + std::io::Write::write_all(&mut file, &data).unwrap(); + let file = File::from_std(file); + let (mut old, reader) = FileReader::start(&file, 7, 150_000).await.unwrap(); + + // Occupy the sole blocking worker so the old read remains pending. + let (release, blocked) = std::sync::mpsc::channel(); + let (started, running) = oneshot::channel(); + let worker = tokio::task::spawn_blocking(move || { + started.send(()).unwrap(); + blocked.recv().unwrap(); + }); + running.await.unwrap(); + tokio::task::yield_now().await; + assert!(futures_util::poll!(old.next()).is_pending()); + let finish = reader.finish(); + tokio::pin!(finish); + assert!(futures_util::poll!(&mut finish).is_pending()); + release.send(()).unwrap(); + worker.await.unwrap(); + finish.await.unwrap(); + let (stream, reader) = FileReader::start(&file, 107, 149_900).await.unwrap(); + let remaining: Vec<_> = stream.try_collect().await.unwrap(); + reader.finish().await.unwrap(); + assert_eq!(remaining.concat(), data[107..150_007]); + drop(old); + }); + } +} diff --git a/clients/rust/src/resumable.rs b/clients/rust/src/resumable.rs index b4dc58bb..7cd9c315 100644 --- a/clients/rust/src/resumable.rs +++ b/clients/rust/src/resumable.rs @@ -3,14 +3,15 @@ //! Enable the `resumable-upload-api` feature to use this API. //! It maps directly to the server operations and is intended for advanced use cases that need //! low-level protocol calls. -//! A future release of `objectstore-client` will use this API under the hood for eligible -//! `Session::put` calls, so users can benefit from this feature without the need to interact with -//! this API directly. +//! Eligible uploads use this protocol automatically, without requiring the feature flag. +#[cfg(feature = "resumable-upload-api")] use std::borrow::Cow; +#[cfg(feature = "resumable-upload-api")] use std::collections::BTreeMap; use std::fmt; +#[cfg(feature = "resumable-upload-api")] use bytes::Bytes; use objectstore_types::metadata::Metadata; use objectstore_types::resumable::{ @@ -19,13 +20,17 @@ use objectstore_types::resumable::{ }; use reqwest::{Body, Method, Response, StatusCode}; use serde::Serialize; +#[cfg(feature = "resumable-upload-api")] use tokio::io::AsyncRead; +#[cfg(feature = "resumable-upload-api")] use tokio_util::io::ReaderStream; pub use objectstore_types::resumable::{SessionToken, UploadProgress}; use crate::response::ResponseExt as _; -use crate::{ClientStream, Compression, Error, ExpirationPolicy, ObjectKey, Session}; +use crate::{ClientStream, Error, ObjectKey, Session}; +#[cfg(feature = "resumable-upload-api")] +use crate::{Compression, ExpirationPolicy}; #[derive(Serialize)] #[serde(rename_all = "snake_case")] @@ -56,6 +61,7 @@ pub struct ResumableUpload { granularity: Option, } +#[cfg(feature = "resumable-upload-api")] impl Session { /// Starts building a resumable upload for an object with `object_length` bytes. /// @@ -102,6 +108,7 @@ impl ResumableUpload { } /// Returns the session token bound to this upload. + #[cfg(feature = "resumable-upload-api")] pub fn token(&self) -> &SessionToken { &self.token } @@ -110,6 +117,7 @@ impl ResumableUpload { /// /// The granularity is the persistence unit for non-final chunks: chunks shorter than one /// unit are rejected, and larger chunks may persist only an aligned prefix. + #[cfg(feature = "resumable-upload-api")] pub fn granularity(&self) -> Option { self.granularity } @@ -129,6 +137,7 @@ impl ResumableUpload { /// /// The returned progress contains the authoritative server offset, which should be used for /// subsequent requests. + #[cfg(feature = "resumable-upload-api")] pub fn put(&self, offset: u64, chunk: impl Into) -> PutChunkBuilder { let chunk = chunk.into(); self.put_body(offset, chunk.len() as u64, chunk.into()) @@ -146,6 +155,7 @@ impl ResumableUpload { /// /// The reader must produce exactly `length` bytes. If this upload records compression, /// those bytes must come from the object after it has been compressed in full. + #[cfg(feature = "resumable-upload-api")] pub fn put_read(&self, offset: u64, length: u64, reader: R) -> PutChunkBuilder where R: AsyncRead + Send + Sync + 'static, @@ -201,15 +211,16 @@ impl ResumableUpload { } } -/// A builder for [`Session::create_upload`]. +/// A builder for creating a resumable upload. #[derive(Debug)] pub struct CreateResumableUploadBuilder { - session: Session, - total_length: u64, - key: Option, - metadata: Metadata, + pub(crate) session: Session, + pub(crate) total_length: u64, + pub(crate) key: Option, + pub(crate) metadata: Metadata, } +#[cfg(feature = "resumable-upload-api")] impl CreateResumableUploadBuilder { /// Sets an explicit object key instead of asking the server to generate one. pub fn key(mut self, key: impl Into) -> Self { @@ -264,7 +275,9 @@ impl CreateResumableUploadBuilder { self.metadata.custom.insert(key.into(), value.into()); self } +} +impl CreateResumableUploadBuilder { /// Creates the resumable upload and returns a handle bound to it. /// /// Returns `None` when the server declines resumable uploads for this object. @@ -336,8 +349,7 @@ impl UploadProgressBuilder { } } -/// A builder for [`ResumableUpload::put`], [`ResumableUpload::put_stream`], or -/// [`ResumableUpload::put_read`]. +/// A builder for writing a resumable upload chunk. pub struct PutChunkBuilder { upload: ResumableUpload, offset: u64, diff --git a/clients/rust/tests/automatic.rs b/clients/rust/tests/automatic.rs new file mode 100644 index 00000000..2ca528e6 --- /dev/null +++ b/clients/rust/tests/automatic.rs @@ -0,0 +1,621 @@ +//! Automatic upload tests do not require the `resumable-upload-api` feature. + +use std::collections::VecDeque; +use std::io::{self, SeekFrom, Write}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use axum::body::{Body, Bytes, to_bytes}; +use axum::extract::{Request, State}; +use axum::http::{HeaderMap, Method, Response, Uri}; +use futures_util::{StreamExt, stream}; +use objectstore_client::{Client, Compression, Error, ExpirationPolicy, Session, Usecase}; +use tokio::io::AsyncSeekExt; + +#[derive(Clone, Debug)] +struct Received { + method: Method, + uri: Uri, + headers: HeaderMap, + body: Bytes, +} + +enum Reply { + Create, + Complete, + Offset(u16, u64), + EarlyOffset(u64), + Status(u16), + LostCompletion, + Response(Response), +} + +impl Reply { + fn response(self) -> Response { + let mut response = Response::builder(); + let body = match self { + Self::Create => Body::from(r#"{"key":"key","session":"dG9rZW4","granularity":4}"#), + Self::Complete => { + response = response.status(201); + Body::from(r#"{"key":"key"}"#) + } + Self::Offset(status, offset) => { + response = response.status(status).header("upload-offset", offset); + Body::empty() + } + Self::EarlyOffset(offset) => return Self::Offset(409, offset).response(), + Self::Status(status) => { + response = response.status(status); + Body::from(r#"{"key":"key"}"#) + } + Self::LostCompletion => { + response = response.status(201).header("content-length", "100"); + Body::from_stream(stream::once(async { + Err::(io::Error::other("connection lost after commit")) + })) + } + Self::Response(response) => return response, + }; + response.body(body).unwrap() + } +} + +#[derive(Default)] +struct Script { + replies: VecDeque, + received: Vec, +} + +struct Server { + url: String, + script: Arc>, + task: tokio::task::JoinHandle<()>, +} + +impl Server { + async fn new(replies: impl IntoIterator) -> Self { + async fn handle( + State(script): State>>, + request: Request, + ) -> Response { + let (parts, body) = request.into_parts(); + // Model a server that responds before consuming the request body. + { + let mut script = script.lock().unwrap(); + if matches!(script.replies.front(), Some(Reply::EarlyOffset(_))) { + script.received.push(Received { + method: parts.method, + uri: parts.uri, + headers: parts.headers, + body: Bytes::new(), + }); + // Drain in the background so HTTP/1 can deliver the early response + // without resetting the connection due to unread request data. + tokio::spawn(async move { + let _ = to_bytes(body, 16 * 1024 * 1024).await; + }); + return script.replies.pop_front().unwrap().response(); + } + } + // A truncated request body is expected in the local-file-failure test. + let Ok(body) = to_bytes(body, 16 * 1024 * 1024).await else { + return Reply::Status(400).response(); + }; + let reply = { + let mut script = script.lock().unwrap(); + script.received.push(Received { + method: parts.method, + uri: parts.uri, + headers: parts.headers, + body, + }); + script.replies.pop_front().expect("unexpected HTTP request") + }; + reply.response() + } + let script = Arc::new(Mutex::new(Script { + replies: replies.into_iter().collect(), + ..Default::default() + })); + let app = axum::Router::new() + .fallback(handle) + .with_state(script.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + Self { url, script, task } + } + + fn client(&self) -> objectstore_client::ClientBuilder { + Client::builder(&self.url) + .configure_reqwest(|builder| builder.no_proxy().connect_timeout(Duration::from_secs(5))) + } + + fn session(&self, threshold: Option) -> Session { + self.client() + .build() + .unwrap() + .session( + Usecase::from_static("testing") + .with_compression(None) + .with_resumable_threshold_bytes(threshold) + .for_organization(123), + ) + .unwrap() + } + + fn finish(&self) -> Vec { + let script = self.script.lock().unwrap(); + assert!( + script.replies.is_empty(), + "{} unused replies", + script.replies.len() + ); + script.received.clone() + } +} + +impl Drop for Server { + fn drop(&mut self) { + self.task.abort(); + } +} + +#[tokio::test] +async fn threshold_inheritance_overrides_and_empty_payloads() { + assert_eq!( + Usecase::from_static("testing").resumable_threshold_bytes(), + Some(256 * 1024 * 1024) + ); + for (threshold, override_threshold, data, resumable) in [ + (Some(4), None, "abc", false), + (Some(4), None, "abcd", true), + (Some(4), Some(None), "abcd", false), + (None, None, "abcd", false), + (None, Some(Some(4)), "abcd", true), + (Some(0), None, "", true), + ] { + let replies = if resumable { + vec![Reply::Create, Reply::Complete] + } else { + vec![Reply::Status(200)] + }; + let server = Server::new(replies).await; + let mut put = server.session(threshold).put(data); + if let Some(threshold) = override_threshold { + put = put.resumable_threshold_bytes(threshold); + } + assert_eq!(put.send().await.unwrap().key, "key"); + let requests = server.finish(); + assert_eq!(requests.last().unwrap().body.as_ref(), data.as_bytes()); + assert_eq!(requests[0].headers.contains_key("upload-length"), resumable); + } +} + +#[tokio::test] +async fn streams_and_readers_remain_direct() { + let server = Server::new([Reply::Status(200), Reply::Status(200)]).await; + let session = server.session(Some(0)); + session + .put_stream( + stream::iter([Ok(Bytes::from_static(b"a")), Ok(Bytes::from_static(b"bc"))]).boxed(), + ) + .send() + .await + .unwrap(); + session + .put_read(io::Cursor::new(b"def")) + .send() + .await + .unwrap(); + let requests = server.finish(); + assert_eq!(requests[0].body, "abc"); + assert_eq!(requests[1].body, "def"); + assert!( + requests + .iter() + .all(|r| !r.headers.contains_key("upload-length")) + ); +} + +#[tokio::test] +async fn files_use_remaining_length_and_paths_open_lazily() { + let server = Server::new([ + Reply::Create, + Reply::Offset(204, 2), + Reply::Complete, + Reply::Status(200), + Reply::Create, + Reply::Complete, + ]) + .await; + let session = server.session(Some(4)); + let mut temporary = tempfile::NamedTempFile::new().unwrap(); + temporary.write_all(b"prefix-data").unwrap(); + let mut file = tokio::fs::File::open(temporary.path()).await.unwrap(); + file.seek(SeekFrom::Start(7)).await.unwrap(); + session.put_file(file).send().await.unwrap(); + let mut file = tokio::fs::File::open(temporary.path()).await.unwrap(); + file.seek(SeekFrom::Start(8)).await.unwrap(); + session.put_file(file).send().await.unwrap(); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("created-after-builder"); + let put = session.put_path(&path); + tokio::fs::write(path, b"path").await.unwrap(); + put.send().await.unwrap(); + let requests = server.finish(); + assert_eq!(requests[0].headers["upload-length"], "4"); + assert_eq!(requests[1].body, "data"); + assert_eq!(requests[2].body, "ta"); + assert_eq!(requests[3].body, "ata"); + assert_eq!(requests[5].body, "path"); +} + +#[tokio::test] +async fn file_retry_after_response_before_body_is_consumed() { + let server = Server::new([Reply::Create, Reply::EarlyOffset(4), Reply::Complete]).await; + let data: Vec = (0..8 * 1024 * 1024).map(|n| (n % 251) as u8).collect(); + let mut file = tempfile::tempfile().unwrap(); + file.write_all(&data).unwrap(); + std::io::Seek::rewind(&mut file).unwrap(); + let put = server + .session(Some(0)) + .put_file(tokio::fs::File::from_std(file)); + tokio::time::timeout(Duration::from_secs(10), put.send()) + .await + .expect("retry must recover ownership of the previous request's file") + .unwrap(); + let requests = server.finish(); + assert_eq!(requests[2].method, Method::PUT); + assert_eq!(requests[2].body.len(), data.len() - 4); + assert!(requests[2].body.as_ref() == &data[4..]); +} + +#[tokio::test] +async fn compression_spooling_metadata_and_creation_fallback() { + for creation in [Reply::Create, Reply::Status(501), Reply::Status(503)] { + let fallback = !matches!(creation, Reply::Create); + let server = Server::new([ + creation, + if fallback { + Reply::Status(200) + } else { + Reply::Complete + }, + ]) + .await; + let dir = tempfile::tempdir().unwrap(); + let client = server.client().spool_dir(dir.path()).build().unwrap(); + let session = client + .session( + Usecase::from_static("testing") + .with_resumable_threshold_bytes(1024) + .for_organization(123), + ) + .unwrap(); + session + .put(vec![b'x'; 1024]) + .key("key") + .content_type("text/plain") + .filename("file.txt") + .origin("127.0.0.1") + .append_metadata("test", "value") + .expiration_policy(ExpirationPolicy::TimeToLive(Duration::from_secs(60))) + .send() + .await + .unwrap(); + let requests = server.finish(); + let encoded = &requests[1].body; + assert_eq!( + zstd::decode_all(encoded.as_ref()).unwrap(), + vec![b'x'; 1024] + ); + assert!(encoded.len() < 1024); // Eligibility was decided before compression. + assert_eq!( + requests[0].headers["upload-length"], + encoded.len().to_string() + ); + assert_eq!( + requests[1].headers["content-length"], + encoded.len().to_string() + ); + assert_eq!(requests[0].method, Method::PUT); + let metadata = + objectstore_types::metadata::Metadata::from_headers(&requests[0].headers, "").unwrap(); + assert_eq!(metadata.compression, Some(Compression::Zstd)); + assert_eq!(metadata.content_type, "text/plain"); + assert_eq!(metadata.filename.as_deref(), Some("file.txt")); + assert_eq!(metadata.origin.as_deref(), Some("127.0.0.1")); + assert_eq!(metadata.custom["test"], "value"); + assert_eq!( + metadata.expiration_policy, + ExpirationPolicy::TimeToLive(Duration::from_secs(60)) + ); + if fallback { + assert_eq!(requests[1].headers["content-encoding"], "zstd"); + } + assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); + } +} + +#[tokio::test] +async fn missing_spool_directory_falls_back_and_precompressed_input_needs_no_spool() { + let server = Server::new([Reply::Status(200), Reply::Create, Reply::Complete]).await; + let dir = tempfile::tempdir().unwrap(); + let client = server + .client() + .spool_dir(dir.path().join("missing")) + .build() + .unwrap(); + let session = client + .session( + Usecase::from_static("testing") + .with_resumable_threshold_bytes(0) + .for_organization(123), + ) + .unwrap(); + session.put("original").send().await.unwrap(); + let compressed = zstd::encode_all(&b"original"[..], 0).unwrap(); + session + .put(compressed.clone()) + .precompressed(Compression::Zstd) + .send() + .await + .unwrap(); + let requests = server.finish(); + assert!(!requests[0].headers.contains_key("upload-length")); + assert_eq!( + zstd::decode_all(requests[0].body.as_ref()).unwrap(), + b"original" + ); + assert_eq!(requests[2].body.as_ref(), compressed); +} + +#[tokio::test] +async fn many_keeps_oversized_uploads_direct() { + let server = Server::new([Reply::Status(200)]).await; + let session = server.session(Some(0)); + let data = Bytes::from(vec![b'x'; 1024 * 1024 + 1]); + session + .many() + .push(session.put(data.clone()).resumable_threshold_bytes(0)) + .send() + .await + .error_for_failures() + .await + .map_err(|errors| errors.collect::>()) + .unwrap(); + let requests = server.finish(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].uri.query(), None); + assert_eq!(requests[0].body, data); +} + +#[tokio::test] +async fn partial_progress_and_conflicts_replay_only_the_suffix() { + let server = Server::new([ + Reply::Create, + Reply::Offset(204, 3), + Reply::Offset(409, 5), + Reply::Complete, + ]) + .await; + server + .session(Some(0)) + .put("abcdefgh") + .send() + .await + .unwrap(); + let requests = server.finish(); + for (request, offset, body) in [ + (&requests[1], "0", "abcdefgh"), + (&requests[2], "3", "defgh"), + (&requests[3], "5", "fgh"), + ] { + assert_eq!(request.headers["upload-offset"], offset); + assert_eq!(request.body, body); + } +} + +#[tokio::test] +async fn recovery_probes_and_progress_resets_the_budget() { + let server = Server::new([ + Reply::Create, + Reply::Status(503), + Reply::Status(502), + Reply::Offset(204, 2), + Reply::Status(429), + Reply::Status(504), + Reply::Offset(204, 4), + Reply::Complete, + ]) + .await; + // Even with middleware retries enabled, a failed write must be followed by a probe. + let client = server + .client() + .configure_reqwest(|b| { + b.retry( + reqwest::retry::for_host("127.0.0.1") + .no_budget() + .classify_fn(|r| { + if r.status() == Some(reqwest::StatusCode::SERVICE_UNAVAILABLE) { + r.retryable() + } else { + r.success() + } + }), + ) + }) + .build() + .unwrap(); + let session = client + .session( + Usecase::from_static("testing") + .with_compression(None) + .with_resumable_threshold_bytes(0) + .for_organization(123), + ) + .unwrap(); + session.put("abcdef").send().await.unwrap(); + let requests = server.finish(); + let offsets: Vec<_> = requests[1..] + .iter() + .map(|r| r.headers["upload-offset"].to_str().unwrap()) + .collect(); + assert_eq!(offsets, ["0", "*", "*", "2", "*", "*", "4"]); + assert_eq!(requests[4].body, "cdef"); + assert_eq!(requests[7].body, "ef"); +} + +#[tokio::test] +async fn no_progress_and_probe_failures_share_the_retry_budget() { + let server = Server::new([ + Reply::Create, + Reply::Offset(204, 0), + Reply::Status(503), + Reply::Offset(204, 0), + Reply::Offset(409, 0), + Reply::Status(500), + ]) + .await; + let error = server + .session(Some(0)) + .put("data") + .send() + .await + .unwrap_err(); + assert!(matches!(&error, Error::UploadFailed(_))); + assert!(error.to_string().contains("no progress")); + let requests = server.finish(); + assert_eq!(requests[2].headers["upload-offset"], "*"); + assert_eq!(requests[3].headers["upload-offset"], "*"); + assert_eq!(requests.last().unwrap().method, Method::DELETE); +} + +#[tokio::test] +async fn permanent_errors_and_invalid_offsets_cancel_without_restarting() { + for replies in [ + vec![Reply::Create, Reply::Status(403), Reply::Status(500)], + vec![Reply::Create, Reply::Offset(204, 10), Reply::Status(204)], + vec![ + Reply::Create, + Reply::Offset(204, 2), + Reply::Offset(204, 1), + Reply::Status(204), + ], + vec![Reply::Create, Reply::Status(410), Reply::Status(204)], + ] { + let server = Server::new(replies).await; + let error = server + .session(Some(0)) + .put("data") + .send() + .await + .unwrap_err(); + assert!(matches!(error, Error::UploadFailed(_))); + assert!(std::error::Error::source(&error).is_some()); + assert_eq!(server.finish().last().unwrap().method, Method::DELETE); + } +} + +#[tokio::test] +async fn lost_completion_is_recovered_without_reuploading() { + let server = Server::new([Reply::Create, Reply::LostCompletion, Reply::Complete]).await; + server.session(Some(0)).put("data").send().await.unwrap(); + let requests = server.finish(); + assert_eq!(requests[2].headers["upload-offset"], "*"); + assert!(requests[2].body.is_empty()); +} + +#[tokio::test] +async fn local_file_errors_are_not_retried() { + let mut file = tempfile::NamedTempFile::new().unwrap(); + file.write_all(b"data").unwrap(); + let path = file.path().to_owned(); + let (parts, body) = Reply::Create.response().into_parts(); + let body = Body::from_stream(stream::once(async move { + // The client has measured the file, but cannot upload until it receives this body. + tokio::fs::write(path, []).await?; + to_bytes(body, 1024).await.map_err(io::Error::other) + })); + let server = Server::new([ + Reply::Response(Response::from_parts(parts, body)), + Reply::Status(204), + ]) + .await; + let error = server + .session(Some(0)) + .put_path(file.path()) + .send() + .await + .unwrap_err(); + assert!(matches!(error, Error::UploadFailed(_))); + assert!(error.to_string().contains("upload failed")); + let requests = server.finish(); + assert_eq!(requests.len(), 2); + assert_eq!(requests[1].method, Method::DELETE); +} + +#[tokio::test] +async fn compressed_file_replays_the_prepared_bytes() { + let server = Server::new([Reply::Create, Reply::Offset(204, 4), Reply::Complete]).await; + let mut file = tempfile::NamedTempFile::new().unwrap(); + file.write_all(b"prefix-payload").unwrap(); + let mut input = tokio::fs::File::open(file.path()).await.unwrap(); + input.seek(SeekFrom::Start(7)).await.unwrap(); + server + .session(Some(0)) + .put_file(input) + .compress(Compression::Zstd) + .send() + .await + .unwrap(); + let requests = server.finish(); + let encoded = &requests[1].body; + assert_eq!(requests[2].body, encoded.slice(4..)); + assert_eq!(zstd::decode_all(encoded.as_ref()).unwrap(), b"payload"); +} + +#[tokio::test] +async fn preparation_read_failure_does_not_fall_back() { + let server = Server::new([]).await; + let mut file = tempfile::NamedTempFile::new().unwrap(); + file.write_all(b"payload").unwrap(); + let write_only = std::fs::OpenOptions::new() + .write(true) + .open(file.path()) + .unwrap(); + let error = server + .session(Some(0)) + .put_file(tokio::fs::File::from_std(write_only)) + .compress(Compression::Zstd) + .send() + .await + .unwrap_err(); + assert!(matches!(error, Error::Io(_))); + assert!(server.finish().is_empty()); +} + +#[cfg(unix)] +#[tokio::test] +async fn non_regular_files_use_direct_uploads() { + let server = Server::new([Reply::Status(200)]).await; + let file = tokio::fs::File::open("/dev/null").await.unwrap(); + server.session(Some(0)).put_file(file).send().await.unwrap(); + let requests = server.finish(); + assert!(!requests[0].headers.contains_key("upload-length")); + assert!(requests[0].body.is_empty()); +} + +#[test] +fn upload_futures_are_send() { + fn assert_send(_: impl Send) {} + let session = Client::builder("http://localhost") + .build() + .unwrap() + .session(Usecase::from_static("testing").for_organization(123)) + .unwrap(); + assert_send(session.put("data").send()); + assert_send(session.put_path("data").send()); + assert_send(session.many().push(session.put("data")).send()); +} diff --git a/clients/rust/tests/e2e.rs b/clients/rust/tests/e2e.rs index 96d8e763..c37e78af 100644 --- a/clients/rust/tests/e2e.rs +++ b/clients/rust/tests/e2e.rs @@ -71,14 +71,23 @@ async fn stores_uncompressed() { let body = "oh hai!"; - let stored_id = session.put(body).compress(None).send().await.unwrap().key; + for threshold in [None, Some(0)] { + let stored_id = session + .put(body) + .resumable_threshold_bytes(threshold) + .compress(None) + .send() + .await + .unwrap() + .key; - let response = session.get(&stored_id).send().await.unwrap().unwrap(); - assert_eq!(response.metadata.compression, None); - assert!(response.metadata.time_created.is_some()); + let response = session.get(&stored_id).send().await.unwrap().unwrap(); + assert_eq!(response.metadata.compression, None); + assert!(response.metadata.time_created.is_some()); - let received = response.payload().await.unwrap(); - assert_eq!(received, "oh hai!"); + let received = response.payload().await.unwrap(); + assert_eq!(received, "oh hai!"); + } } #[tokio::test] @@ -93,23 +102,31 @@ async fn uses_zstd_by_default() { let session = client.session(usecase.for_organization(12345)).unwrap(); let body = "oh hai!"; - let stored_id = session.put(body).send().await.unwrap().key; + for threshold in [None, Some(0)] { + let stored_id = session + .put(body) + .resumable_threshold_bytes(threshold) + .send() + .await + .unwrap() + .key; - // when the user indicates that it can deal with zstd, it gets zstd - let request = session.get(&stored_id).decompress(false); - let response = request.send().await.unwrap().unwrap(); - assert_eq!(response.metadata.compression, Some(Compression::Zstd)); + // when the user indicates that it can deal with zstd, it gets zstd + let request = session.get(&stored_id).decompress(false); + let response = request.send().await.unwrap().unwrap(); + assert_eq!(response.metadata.compression, Some(Compression::Zstd)); - let received_compressed = response.payload().await.unwrap(); - let decompressed = zstd::bulk::decompress(&received_compressed, 1024).unwrap(); - assert_eq!(decompressed, b"oh hai!"); + let received_compressed = response.payload().await.unwrap(); + let decompressed = zstd::bulk::decompress(&received_compressed, 1024).unwrap(); + assert_eq!(decompressed, b"oh hai!"); - // otherwise, the client does the decompression - let response = session.get(&stored_id).send().await.unwrap().unwrap(); - assert_eq!(response.metadata.compression, None); + // otherwise, the client does the decompression + let response = session.get(&stored_id).send().await.unwrap().unwrap(); + assert_eq!(response.metadata.compression, None); - let received = response.payload().await.unwrap(); - assert_eq!(received, "oh hai!"); + let received = response.payload().await.unwrap(); + assert_eq!(received, "oh hai!"); + } } #[tokio::test] @@ -124,29 +141,32 @@ async fn stores_precompressed_payload_verbatim() { let session = client.session(usecase.for_organization(12345)).unwrap(); let compressed = zstd::bulk::compress(b"oh hai!", 0).unwrap(); - let stored_id = session - .put(compressed.clone()) - .precompressed(Compression::Zstd) - .send() - .await - .unwrap() - .key; - - // the payload is stored as-is, without another compression pass - let response = session - .get(&stored_id) - .accept_encoding([Compression::Zstd]) - .send() - .await - .unwrap() - .unwrap(); - assert_eq!(response.metadata.compression, Some(Compression::Zstd)); - assert_eq!(response.payload().await.unwrap(), compressed); + for threshold in [None, Some(0)] { + let stored_id = session + .put(compressed.clone()) + .resumable_threshold_bytes(threshold) + .precompressed(Compression::Zstd) + .send() + .await + .unwrap() + .key; - // and the recorded encoding makes downloads decompress transparently - let response = session.get(&stored_id).send().await.unwrap().unwrap(); - assert_eq!(response.metadata.compression, None); - assert_eq!(response.payload().await.unwrap(), "oh hai!"); + // the payload is stored as-is, without another compression pass + let response = session + .get(&stored_id) + .accept_encoding([Compression::Zstd]) + .send() + .await + .unwrap() + .unwrap(); + assert_eq!(response.metadata.compression, Some(Compression::Zstd)); + assert_eq!(response.payload().await.unwrap(), compressed); + + // and the recorded encoding makes downloads decompress transparently + let response = session.get(&stored_id).send().await.unwrap().unwrap(); + assert_eq!(response.metadata.compression, None); + assert_eq!(response.payload().await.unwrap(), "oh hai!"); + } } #[tokio::test] @@ -819,30 +839,39 @@ async fn put_file_with_compression() { let body = "hello compressed file!"; std::fs::write(&path, body).unwrap(); - // Default compression is zstd - let file = tokio::fs::File::open(&path).await.unwrap(); - let stored_id = session.put_file(file).send().await.unwrap().key; - - // When requesting raw, the stored object should be zstd-compressed - let response = session - .get(&stored_id) - .decompress(false) - .send() - .await - .unwrap() - .unwrap(); - assert_eq!(response.metadata.compression, Some(Compression::Zstd)); - - let compressed = response.payload().await.unwrap(); - let decompressed = zstd::bulk::decompress(&compressed, 1024).unwrap(); - assert_eq!(decompressed, body.as_bytes()); - - // Default get decompresses transparently - let response = session.get(&stored_id).send().await.unwrap().unwrap(); - assert_eq!(response.metadata.compression, None); - - let received = response.payload().await.unwrap(); - assert_eq!(received, body); + for threshold in [None, Some(0)] { + let file = tokio::fs::File::open(&path).await.unwrap(); + for put in [session.put_file(file), session.put_path(&path)] { + // Default compression is zstd. + let stored_id = put + .resumable_threshold_bytes(threshold) + .send() + .await + .unwrap() + .key; + + // When requesting raw, the stored object should be zstd-compressed + let response = session + .get(&stored_id) + .decompress(false) + .send() + .await + .unwrap() + .unwrap(); + assert_eq!(response.metadata.compression, Some(Compression::Zstd)); + + let compressed = response.payload().await.unwrap(); + let decompressed = zstd::bulk::decompress(&compressed, 1024).unwrap(); + assert_eq!(decompressed, body.as_bytes()); + + // Default get decompresses transparently + let response = session.get(&stored_id).send().await.unwrap().unwrap(); + assert_eq!(response.metadata.compression, None); + + let received = response.payload().await.unwrap(); + assert_eq!(received, body); + } + } } #[tokio::test]