diff --git a/Cargo.lock b/Cargo.lock index 4ad71265..770bcaab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2840,6 +2840,7 @@ version = "0.3.0" dependencies = [ "anyhow", "async-compression", + "axum", "bytes", "flate2", "futures-util", @@ -2849,6 +2850,7 @@ dependencies = [ "multer", "objectstore-test", "objectstore-types", + "rand 0.10.3", "reqwest 0.13.5", "sentry-core", "serde", @@ -2856,6 +2858,7 @@ dependencies = [ "thiserror 2.0.21", "tokio", "tokio-util", + "tracing", "url", "zstd", "zstd-safe", diff --git a/clients/rust/Cargo.toml b/clients/rust/Cargo.toml index 386f0247..4d7cf08e 100644 --- a/clients/rust/Cargo.toml +++ b/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"] diff --git a/clients/rust/README.md b/clients/rust/README.md index bd6c855e..09acc075 100644 --- a/clients/rust/README.md +++ b/clients/rust/README.md @@ -151,7 +151,16 @@ 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. > @@ -159,12 +168,12 @@ session.put("payload") > 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. diff --git a/clients/rust/src/client.rs b/clients/rust/src/client.rs index 832ca474..99e7de09 100644 --- a/clients/rust/src/client.rs +++ b/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, + spool_dir: Option, } 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) -> 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, expiration_policy: ExpirationPolicy, + resumable_threshold_bytes: Option, } 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 { + 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>) -> 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, reqwest: reqwest::Client, service_url: Url, propagate_traces: bool, token: Option, } +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 diff --git a/clients/rust/src/error.rs b/clients/rust/src/error.rs index a99ead26..2142cceb 100644 --- a/clients/rust/src/error.rs +++ b/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}" )] diff --git a/clients/rust/src/lib.rs b/clients/rust/src/lib.rs index 2924b257..8f2c0f89 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; @@ -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, +}; diff --git a/clients/rust/src/many.rs b/clients/rust/src/many.rs index 751d02ba..7086969f 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, @@ -660,6 +661,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 2528ecc1..eeceb9f6 100644 --- a/clients/rust/src/put.rs +++ b/clients/rust/src/put.rs @@ -12,12 +12,18 @@ use serde::Deserialize; use tokio::fs::File; use tokio::io::{AsyncBufRead, AsyncRead, BufReader}; use tokio_util::io::{ReaderStream, StreamReader}; +#[cfg(feature = "tracing")] +use tracing::debug; pub use objectstore_types::metadata::{Compression, ExpirationPolicy}; use crate::response::ResponseExt as _; +use crate::resumable::CreateResumableUploadBuilder; 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 +81,7 @@ impl Session { .compression() .map(CompressionMode::Compress), key: None, + resumable_threshold_bytes: self.scope.usecase().resumable_threshold_bytes(), body, } } @@ -85,11 +92,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(&self, body: R) -> PutBuilder where R: AsyncRead + Send + Sync + 'static, @@ -100,6 +111,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 +124,12 @@ impl Session { /// Creates or replaces an object using the contents of the file at `path`. /// + /// The path must stay readable and must not be modified, renamed, removed, or replaced + /// until the upload completes. + /// If the request is executed using a resumable upload, then multiple attempts may be + /// performed, and 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,6 +144,14 @@ 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, @@ -132,9 +159,28 @@ 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. + /// + /// 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>) -> 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 +192,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. @@ -307,15 +353,37 @@ impl PutBuilder { None => reqwest::Method::POST, }; + self.metadata.compression = self.compression.map(CompressionMode::compression); + let headers = self.metadata.to_headers("")?; + let prepared = payload::prepare( + self.body, + self.compression, + self.resumable_threshold_bytes, + self.session.client.spool_dir().map(PathBuf::from), + ) + .await?; + if let payload::PreparedBody::Replayable(body) = &prepared { + let create = CreateResumableUploadBuilder::new( + self.session.clone(), + body.len(), + self.key.clone(), + self.metadata, + ); + match create.send().await { + Ok(Some(upload)) => return automatic::upload(body, upload).await, + #[cfg(feature = "tracing")] + Err(error) => { + debug!(%error, "Resumable upload creation failed; falling back to direct upload"); + } + _ => {} + } + } + + let body = prepared.into_reqwest_body(self.compression).await?; 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("")?); - + builder = builder.headers(headers); let response = builder.body(body).send().await?; Ok(response.error_for_status_and_drain().await?.json().await?) } diff --git a/clients/rust/src/put/automatic.rs b/clients/rust/src/put/automatic.rs new file mode 100644 index 00000000..a272c035 --- /dev/null +++ b/clients/rust/src/put/automatic.rs @@ -0,0 +1,105 @@ +//! Contains the implementation for the automatic resumable upload loop. + +use std::io; +use std::time::Duration; + +use crate::Error; +use crate::put::PutResponse; +use crate::put::payload::{ReadError, ReplayableBody}; +use crate::resumable::{ResumableUpload, UploadProgress}; + +const MAX_RETRIES: u32 = 2; +const BASE_RETRY_DELAY: Duration = Duration::from_secs(2); + +pub(crate) async fn upload( + body: &ReplayableBody, + handle: ResumableUpload, +) -> crate::Result { + match upload_inner(body, &handle).await { + Ok(()) => Ok(PutResponse { + key: handle.key().to_owned(), + }), + Err(err) => { + let _ = handle.cancel().send().await; + Err(err) + } + } +} + +async fn upload_inner(body: &ReplayableBody, handle: &ResumableUpload) -> crate::Result<()> { + let mut offset = 0; + let mut retries = 0; + let mut query_progress = false; + loop { + let response = if query_progress { + handle.progress().send().await + } else { + handle + .put_stream(offset, body.len() - offset, body.stream(offset).await?) + .send() + .await + }; + let failure = match get_offset(response) { + Ok(None) => return Ok(()), + Ok(Some(next)) => { + if next < offset || next > body.len() { + return Err(Error::MalformedResponse("invalid upload offset".into())); + } + if next > offset || query_progress { + if next > offset { + retries = 0; + } + offset = next; + query_progress = false; + continue; + } + io::Error::other("upload made no progress").into() + } + Err(error) if is_transient(&error) => error, + Err(error) => return Err(error), + }; + + // Writes and progress queries share a budget, reset only by an advancing offset. + if retries >= MAX_RETRIES { + return Err(failure); + } + let delay = + BASE_RETRY_DELAY * (1 << retries) + Duration::from_secs_f64(rand::random::()); + tokio::time::sleep(delay).await; + retries += 1; + query_progress = true; + } +} + +fn get_offset(result: crate::Result) -> crate::Result> { + match result { + Ok(UploadProgress::Complete) => Ok(None), + Ok(UploadProgress::Incomplete { offset }) | Err(Error::UploadOffsetMismatch { offset }) => { + Ok(Some(offset)) + } + Err(error) => Err(error), + } +} + +fn is_transient(error: &Error) -> bool { + let mut source: Option<&(dyn std::error::Error + 'static)> = Some(error); + while let Some(error) = source { + if error + .downcast_ref::() + .and_then(io::Error::get_ref) + .is_some_and(|inner| inner.is::()) + { + return false; + } + source = error.source(); + } + 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..435e1208 --- /dev/null +++ b/clients/rust/src/put/payload.rs @@ -0,0 +1,280 @@ +//! Helpers to buffer resumable upload bodies. + +use std::io::{self, SeekFrom}; +use std::path::PathBuf; +use std::sync::Arc; + +use async_compression::tokio::bufread::{GzipEncoder, ZstdEncoder}; +use bytes::{Bytes, BytesMut}; +use futures_util::{StreamExt, TryStreamExt, stream}; +use tokio::fs::File; +use tokio::io::{AsyncBufRead, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader}; +use tokio_util::io::ReaderStream; + +use crate::put::{CompressionMode, PutBody, encode_body}; +use crate::{ClientStream, Compression}; + +const READ_BUFFER_SIZE: usize = 4 * 1024; + +pub(crate) enum PreparedBody { + OneShot(PutBody), + Replayable(ReplayableBody), +} + +impl PreparedBody { + pub async fn into_reqwest_body( + self, + compression: Option, + ) -> io::Result { + match self { + Self::OneShot(body) => encode_body(body, compression).await, + Self::Replayable(body) => Ok(reqwest::Body::wrap_stream(body.stream(0).await?)), + } + } +} + +pub(crate) enum ReplayableBody { + Buffer(Bytes), + Path { + path: PathBuf, + length: u64, + }, + Temporary { + file: Arc, + length: u64, + }, +} + +impl ReplayableBody { + pub fn len(&self) -> u64 { + match self { + Self::Buffer(bytes) => bytes.len() as u64, + Self::Path { length, .. } | Self::Temporary { length, .. } => *length, + } + } + + /// Streams the remaining suffix with an independent read position. + pub async fn stream(&self, offset: u64) -> io::Result { + let mut file = match self { + Self::Buffer(bytes) => { + let bytes = bytes.slice(offset as usize..); + // Streaming bodies prevent independent replays by HTTP middleware. + return Ok(stream::once(async move { Ok(bytes) }).boxed()); + } + Self::Path { path, .. } => File::open(path).await?, + Self::Temporary { file, .. } => { + let file = Arc::clone(file); + let length = self.len(); + // Positional reads let old requests finish without changing a retry's position. + return Ok(stream::iter((offset..length).step_by(READ_BUFFER_SIZE)) + .then(move |position| { + let file = Arc::clone(&file); + async move { + tokio::task::spawn_blocking(move || { + let mut buffer = BytesMut::zeroed( + (length - position).min(READ_BUFFER_SIZE as u64) as usize, + ); + read_exact_at(&file, &mut buffer, position)?; + Ok::<_, io::Error>(buffer.freeze()) + }) + .await + .map_err(io::Error::other)? + } + }) + .map_err(|error| io::Error::other(ReadError(error))) + .boxed()); + } + }; + if file.metadata().await?.len() < self.len() { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "upload file became shorter", + )); + } + file.seek(SeekFrom::Start(offset)).await?; + Ok(ReaderStream::new(file.take(self.len() - offset)) + .map_err(|error| io::Error::other(ReadError(error))) + .boxed()) + } +} + +pub async fn prepare( + body: PutBody, + compression: Option, + threshold: Option, + spool_dir: Option, +) -> crate::Result { + let Some(threshold) = threshold else { + return Ok(PreparedBody::OneShot(body)); + }; + let length = match &body { + PutBody::Buffer(bytes) => Some(bytes.len() as u64), + PutBody::Path(path) => tokio::fs::metadata(path) + .await + .ok() + .filter(|metadata| metadata.is_file()) + .map(|metadata| metadata.len()), + _ => None, + }; + let Some(length) = length.filter(|length| *length >= threshold) else { + return Ok(PreparedBody::OneShot(body)); + }; + + if let Some(CompressionMode::Compress(compression)) = 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 { + return Ok(PreparedBody::OneShot(body)); + }; + let mut output = File::from_std(temporary); + let length = match body { + PutBody::Buffer(bytes) => { + compress_into(io::Cursor::new(bytes), compression, &mut output).await? + } + PutBody::Path(path) => { + compress_into( + BufReader::new(File::open(path).await?), + compression, + &mut output, + ) + .await? + } + // Unreachable: the eligibility check above only accepts buffers and paths. + body => return Ok(PreparedBody::OneShot(body)), + }; + output.flush().await?; + return Ok(PreparedBody::Replayable(ReplayableBody::Temporary { + file: Arc::new(output.into_std().await), + length, + })); + } + + Ok(PreparedBody::Replayable(match body { + PutBody::Buffer(bytes) => ReplayableBody::Buffer(bytes), + PutBody::Path(path) => ReplayableBody::Path { path, length }, + // Unreachable: the eligibility check above only accepts buffers and paths. + body => return Ok(PreparedBody::OneShot(body)), + })) +} + +fn read_exact_at(file: &std::fs::File, buffer: &mut [u8], position: u64) -> io::Result<()> { + #[cfg(unix)] + { + use std::os::unix::fs::FileExt; + file.read_exact_at(buffer, position) + } + #[cfg(windows)] + { + use std::os::windows::fs::FileExt; + let mut buffer = buffer; + let mut position = position; + while !buffer.is_empty() { + match file.seek_read(buffer, position) { + Ok(0) => return Err(io::ErrorKind::UnexpectedEof.into()), + Ok(count) => { + position += count as u64; + buffer = &mut buffer[count..]; + } + Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(error) => return Err(error), + } + } + Ok(()) + } + #[cfg(not(any(unix, windows)))] + { + let _ = (file, buffer, position); + Err(io::ErrorKind::Unsupported.into()) + } +} + +async fn compress_into( + reader: impl AsyncBufRead + Unpin, + compression: Compression, + output: &mut File, +) -> io::Result { + match compression { + Compression::Zstd => tokio::io::copy(&mut ZstdEncoder::new(reader), output).await, + Compression::Gzip => tokio::io::copy(&mut GzipEncoder::new(reader), output).await, + } +} + +/// Distinguishes local read failures from transport failures wrapped by reqwest. +#[derive(Debug, thiserror::Error)] +#[error("could not read upload payload: {0}")] +pub(crate) struct ReadError(#[source] io::Error); + +#[cfg(test)] +mod tests { + use futures_util::TryStreamExt; + use tempfile::NamedTempFile; + + use super::*; + + #[tokio::test] + async fn attempts_have_independent_readers() { + let data: Vec = (0..200_000).map(|n| (n % 251) as u8).collect(); + let mut file = NamedTempFile::new().unwrap(); + std::io::Write::write_all(&mut file, &data).unwrap(); + let mut temporary = tempfile::tempfile().unwrap(); + std::io::Write::write_all(&mut temporary, &data).unwrap(); + for body in [ + ReplayableBody::Path { + path: file.path().to_owned(), + length: data.len() as u64, + }, + ReplayableBody::Temporary { + file: Arc::new(temporary), + length: data.len() as u64, + }, + ] { + let mut first = body.stream(0).await.unwrap(); + let mut received = first.try_next().await.unwrap().unwrap().to_vec(); + let second = body.stream(100).await.unwrap(); + drop(body); + let (suffix, tail): (Vec<_>, Vec<_>) = + futures_util::try_join!(second.try_collect(), first.try_collect()).unwrap(); + received.extend(tail.concat()); + assert_eq!(received, data); + assert_eq!(suffix.concat(), data[100..]); + } + } + + #[test] + fn cancelled_read_does_not_interfere_with_retry() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .max_blocking_threads(1) + .build() + .unwrap(); + runtime.block_on(async { + let data: Vec = (0..20_000).map(|n| (n % 251) as u8).collect(); + let mut file = tempfile::tempfile().unwrap(); + std::io::Write::write_all(&mut file, &data).unwrap(); + let body = ReplayableBody::Temporary { + file: Arc::new(file), + length: data.len() as u64, + }; + + // Keep the first read pending while its request is dropped and a retry starts. + let (release, blocked) = std::sync::mpsc::channel(); + let worker = tokio::task::spawn_blocking(move || blocked.recv().unwrap()); + let mut first = body.stream(0).await.unwrap(); + assert!(futures_util::poll!(first.next()).is_pending()); + drop(first); + let mut retry = body.stream(107).await.unwrap(); + assert!(futures_util::poll!(retry.next()).is_pending()); + drop(body); + release.send(()).unwrap(); + worker.await.unwrap(); + + let suffix: Vec<_> = retry.try_collect().await.unwrap(); + assert!(suffix.iter().all(|chunk| chunk.len() <= READ_BUFFER_SIZE)); + assert_eq!(suffix.concat(), data[107..]); + }); + } +} diff --git a/clients/rust/src/resumable.rs b/clients/rust/src/resumable.rs index b4dc58bb..83be80c7 100644 --- a/clients/rust/src/resumable.rs +++ b/clients/rust/src/resumable.rs @@ -3,9 +3,10 @@ //! 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. +//! The client also uses this API internally for eligible `Session::put` calls. + +// Automatic uploads use only part of this module when the public API is disabled. +#![cfg_attr(not(feature = "resumable-upload-api"), allow(dead_code))] use std::borrow::Cow; use std::collections::BTreeMap; @@ -56,6 +57,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. /// @@ -72,12 +74,7 @@ impl Session { ..Default::default() }; - CreateResumableUploadBuilder { - session: self.clone(), - total_length: object_length, - key: None, - metadata, - } + CreateResumableUploadBuilder::new(self.clone(), object_length, None, metadata) } /// Reconstructs a resumable upload handle without contacting the server. @@ -85,17 +82,21 @@ impl Session { /// The `key` and `token` must come from the same previously created upload and scope. The /// server rejects a mismatched or unknown token when an operation is sent. pub fn resume_upload(&self, key: impl Into, token: SessionToken) -> ResumableUpload { - ResumableUpload { - session: self.clone(), - key: key.into(), + ResumableUpload::new(self.clone(), key.into(), token) + } +} + +impl ResumableUpload { + fn new(session: Session, key: ObjectKey, token: SessionToken) -> Self { + Self { + session, + key, token, total_length: None, granularity: None, } } -} -impl ResumableUpload { /// Returns the object key bound to this upload. pub fn key(&self) -> &str { &self.key @@ -201,7 +202,7 @@ impl ResumableUpload { } } -/// A builder for [`Session::create_upload`]. +/// A builder for creating a resumable upload. #[derive(Debug)] pub struct CreateResumableUploadBuilder { session: Session, @@ -211,6 +212,20 @@ pub struct CreateResumableUploadBuilder { } impl CreateResumableUploadBuilder { + pub(crate) fn new( + session: Session, + total_length: u64, + key: Option, + metadata: Metadata, + ) -> Self { + Self { + session, + total_length, + key, + metadata, + } + } + /// Sets an explicit object key instead of asking the server to generate one. pub fn key(mut self, key: impl Into) -> Self { self.key = Some(key.into()).filter(|key| !key.is_empty()); diff --git a/clients/rust/tests/automatic.rs b/clients/rust/tests/automatic.rs new file mode 100644 index 00000000..04ede97a --- /dev/null +++ b/clients/rust/tests/automatic.rs @@ -0,0 +1,634 @@ +//! 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; + +fn decode(compression: Compression, encoded: &[u8]) -> Vec { + match compression { + Compression::Zstd => zstd::decode_all(encoded).unwrap(), + Compression::Gzip => { + let mut decoded = Vec::new(); + io::Read::read_to_end(&mut flate2::read::GzDecoder::new(encoded), &mut decoded) + .unwrap(); + decoded + } + } +} + +#[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_remain_direct_and_paths_open_lazily() { + let server = Server::new([ + Reply::Status(200), + Reply::Status(200), + Reply::Create, + Reply::Offset(204, 2), + Reply::Complete, + ]) + .await; + let session = server.session(Some(0)); + let mut temporary = tempfile::NamedTempFile::new().unwrap(); + temporary.write_all(b"prefix-data").unwrap(); + for position in [7, 8] { + let mut file = tokio::fs::File::open(temporary.path()).await.unwrap(); + file.seek(SeekFrom::Start(position)).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].body, "data"); + assert_eq!(requests[1].body, "ata"); + assert!( + requests[..2] + .iter() + .all(|r| !r.headers.contains_key("upload-length")) + ); + assert_eq!(requests[2].headers["upload-length"], "4"); + assert_eq!(requests[3].body, "path"); + assert_eq!(requests[4].body, "th"); +} + +#[tokio::test] +async fn path_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::NamedTempFile::new().unwrap(); + file.write_all(&data).unwrap(); + let put = server.session(Some(0)).put_path(file.path()); + tokio::time::timeout(Duration::from_secs(10), put.send()) + .await + .expect("retry must use an independent reader") + .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 compression in [Compression::Zstd, Compression::Gzip] { + 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_compression(compression) + .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!(decode(compression, encoded), vec![b'x'; 1024]); + assert!(encoded.len() < 1024); // Eligibility was decided before compression. + assert_eq!( + requests[0].headers["upload-length"], + encoded.len().to_string() + ); + if !fallback { + 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)); + 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"], + compression.as_str() + ); + } + 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::Io(_))); + 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 (case, 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)], + ] + .into_iter() + .enumerate() + { + let server = Server::new(replies).await; + let error = server + .session(Some(0)) + .put("data") + .send() + .await + .unwrap_err(); + match (case, &error) { + (0, Error::Reqwest(error)) => { + assert_eq!(error.status(), Some(reqwest::StatusCode::FORBIDDEN)); + } + (1 | 2, Error::MalformedResponse(message)) => { + assert_eq!(message, "invalid upload offset"); + } + (3, Error::ResumableUploadUnavailable) => {} + _ => panic!("unexpected error for case {case}: {error}"), + } + 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::Io(error) if error.kind() == io::ErrorKind::UnexpectedEof)); + assert_eq!(error.to_string(), "upload file became shorter"); + let requests = server.finish(); + assert_eq!(requests.len(), 2); + assert_eq!(requests[1].method, Method::DELETE); +} + +#[tokio::test] +async fn compressed_path_replays_the_prepared_bytes() { + for compression in [Compression::Zstd, Compression::Gzip] { + let server = Server::new([Reply::Create, Reply::Offset(204, 4), Reply::Complete]).await; + let mut file = tempfile::NamedTempFile::new().unwrap(); + file.write_all(b"payload").unwrap(); + server + .session(Some(0)) + .put_path(file.path()) + .compress(compression) + .send() + .await + .unwrap(); + let requests = server.finish(); + let encoded = &requests[1].body; + assert_eq!(requests[2].body, encoded.slice(4..)); + assert_eq!(decode(compression, encoded), b"payload"); + } +} + +#[cfg(unix)] +#[tokio::test] +async fn non_regular_paths_use_direct_uploads() { + let server = Server::new([Reply::Status(200)]).await; + server + .session(Some(0)) + .put_path("/dev/null") + .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 bfa1922f..f78ed84e 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] @@ -157,29 +174,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] @@ -852,30 +872,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]