From 45bc7c2ed3f47a2f05626626d5460b22d1382af9 Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Fri, 30 Oct 2026 13:47:48 +0100 Subject: [PATCH 1/3] ref(rs-client): Read temporary upload files positionally --- clients/rust/src/put/payload.rs | 139 +++++++++++++++++++++++++++----- 1 file changed, 117 insertions(+), 22 deletions(-) diff --git a/clients/rust/src/put/payload.rs b/clients/rust/src/put/payload.rs index 5df16e6e..10a51bbd 100644 --- a/clients/rust/src/put/payload.rs +++ b/clients/rust/src/put/payload.rs @@ -5,9 +5,8 @@ use std::path::PathBuf; use std::sync::Arc; use async_compression::tokio::bufread::{GzipEncoder, ZstdEncoder}; -use bytes::Bytes; +use bytes::{Bytes, BytesMut}; use futures_util::{StreamExt, TryStreamExt, stream}; -use tempfile::NamedTempFile; use tokio::fs::File; use tokio::io::{AsyncBufRead, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader}; use tokio_util::io::ReaderStream; @@ -15,6 +14,8 @@ 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), @@ -39,7 +40,7 @@ pub(crate) enum ReplayableBody { length: u64, }, Temporary { - file: Arc, + file: Arc, length: u64, }, } @@ -52,7 +53,7 @@ impl ReplayableBody { } } - /// Opens an independent reader for the remaining suffix. + /// 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) => { @@ -63,11 +64,25 @@ impl ReplayableBody { Self::Path { path, .. } => File::open(path).await?, Self::Temporary { file, .. } => { let file = Arc::clone(file); - File::from_std( - tokio::task::spawn_blocking(move || file.reopen()) - .await - .map_err(io::Error::other)??, - ) + 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() { @@ -106,20 +121,16 @@ pub async fn prepare( }; if let Some(CompressionMode::Compress(compression)) = compression { - let temporary = tokio::task::spawn_blocking(move || { - let temporary = match spool_dir { - Some(dir) => NamedTempFile::new_in(dir), - None => NamedTempFile::new(), - }?; - let output = temporary.as_file().try_clone()?; - Ok::<_, io::Error>((temporary, output)) + 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, output)) = temporary else { + let Ok(temporary) = temporary else { return Ok(PreparedBody::OneShot(body)); }; - let mut output = File::from_std(output); + let mut output = File::from_std(temporary); let length = match body { PutBody::Buffer(bytes) => { compress_into(io::Cursor::new(bytes), compression, &mut output).await? @@ -137,7 +148,7 @@ pub async fn prepare( }; output.flush().await?; return Ok(PreparedBody::Replayable(ReplayableBody::Temporary { - file: Arc::new(temporary), + file: Arc::new(output.into_std().await), length, })); } @@ -150,6 +161,37 @@ pub async fn prepare( })) } +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, @@ -169,6 +211,7 @@ pub(crate) struct ReadError(#[source] io::Error); #[cfg(test)] mod tests { use futures_util::TryStreamExt; + use tempfile::NamedTempFile; use super::*; @@ -177,24 +220,76 @@ mod tests { 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(file), + 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(); - let suffix: Vec<_> = second.try_collect().await.unwrap(); - let tail: Vec<_> = first.try_collect().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..]); + }); + } + + #[tokio::test] + async fn short_temporary_file_returns_a_local_read_error() { + let body = ReplayableBody::Temporary { + file: Arc::new(tempfile::tempfile().unwrap()), + length: 1, + }; + let error = body.stream(0).await.unwrap().try_next().await.unwrap_err(); + let error = error + .get_ref() + .unwrap() + .downcast_ref::() + .unwrap(); + assert_eq!(error.0.kind(), io::ErrorKind::UnexpectedEof); + } } From c7e149430dca50e1b44d8c33e6a1f65c91b1b4d5 Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Fri, 30 Oct 2026 13:59:20 +0100 Subject: [PATCH 2/3] test(rs-client): Remove redundant temporary-file EOF test --- clients/rust/src/put/payload.rs | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/clients/rust/src/put/payload.rs b/clients/rust/src/put/payload.rs index 10a51bbd..435e1208 100644 --- a/clients/rust/src/put/payload.rs +++ b/clients/rust/src/put/payload.rs @@ -277,19 +277,4 @@ mod tests { assert_eq!(suffix.concat(), data[107..]); }); } - - #[tokio::test] - async fn short_temporary_file_returns_a_local_read_error() { - let body = ReplayableBody::Temporary { - file: Arc::new(tempfile::tempfile().unwrap()), - length: 1, - }; - let error = body.stream(0).await.unwrap().try_next().await.unwrap_err(); - let error = error - .get_ref() - .unwrap() - .downcast_ref::() - .unwrap(); - assert_eq!(error.0.kind(), io::ErrorKind::UnexpectedEof); - } } From 1b31ee76e0c340ef476694508d806f203b134e0f Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Fri, 30 Oct 2026 14:37:07 +0100 Subject: [PATCH 3/3] docs(rs-client): Clarify resumable uploads and feature flags --- clients/rust/README.md | 27 +++++++++++++++++---------- 1 file changed, 17 insertions(+), 10 deletions(-) diff --git a/clients/rust/README.md b/clients/rust/README.md index 90f567fc..026a2d4d 100644 --- a/clients/rust/README.md +++ b/clients/rust/README.md @@ -151,14 +151,16 @@ session.put("payload") .send().await?; ``` -### Automatic Resumable Uploads +### Resumable Uploads -Large PUTs automatically use resumable uploads, resending only what the server hasn't -stored after a failed request. -See [`PutBuilder`] for eligibility, configuration, and recovery behavior, and -[`Session::put_path`] for the path stability requirement. +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. -### Resumable Upload API +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. > @@ -166,12 +168,12 @@ See [`PutBuilder`] for eligibility, configuration, and recovery behavior, and > 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. @@ -321,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.