From 30729bdf08225a2e645b6d48354b3c980c50ac66 Mon Sep 17 00:00:00 2001 From: Panagiotis Moustafellos Date: Mon, 2 Nov 2026 14:25:49 +0200 Subject: [PATCH 1/5] test(s3s-fs): race conditional puts and reads on one key Three tests for writes and reads that race on one object: - 16 create-if-absent writers: exactly one may succeed, and the stored checksum and content type must belong to the stored body. - 16 If-Match writers holding the same ETag: exactly one may succeed. - 8 unconditional writers per round over 40 rounds, with two readers running throughout: every read must pair a body with its own checksum and its own content type. A gated body holds each writer's last chunk until every writer has streamed its first, so all of them pass the early condition check before any of them stores the object. That makes the first two tests fail every time on the current code, which checks the conditions only once, before the body streams. Each test body is also written as its content type, so a read that mixes the body of one write with the attributes or the checksum of another fails the pairing check. Each writer is bounded by a 60 s timeout, since a writer that fails before the gate would leave the others waiting at it. Refs: NOFireAI/ravel#3652 Signed-off-by: Panagiotis Moustafellos --- crates/s3s-fs/src/s3.rs | 250 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 249 insertions(+), 1 deletion(-) diff --git a/crates/s3s-fs/src/s3.rs b/crates/s3s-fs/src/s3.rs index 976d1db6c..ec4396936 100644 --- a/crates/s3s-fs/src/s3.rs +++ b/crates/s3s-fs/src/s3.rs @@ -2186,7 +2186,14 @@ mod tests { use super::*; use std::env; - + use std::pin::Pin; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering as AtomicOrdering}; + use std::task::{Context, Poll}; + + use bytes::Bytes; + use futures::{Stream, StreamExt}; + use http::{Extensions, HeaderMap, Method, Uri}; use s3s::S3ErrorCode; use uuid::Uuid; @@ -2702,4 +2709,245 @@ mod tests { listing_stats::visited() ); } + + fn request(input: T) -> S3Request { + S3Request { + input, + method: Method::PUT, + uri: Uri::from_static("/"), + headers: HeaderMap::new(), + extensions: Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + } + } + + fn new_fs(name: &str) -> (TestRoot, Arc) { + let root = env::temp_dir().join(format!("s3s-fs-{name}-{}", Uuid::new_v4())); + std::fs::create_dir_all(root.join("bucket")).unwrap(); + let fs = Arc::new(FileSystem::new(&root).unwrap()); + (TestRoot(root), fs) + } + + fn crc64nvme(body: &[u8]) -> Option { + let mut hasher = s3s::checksum::ChecksumHasher { + crc64nvme: Some(default()), + ..Default::default() + }; + hasher.update(body); + hasher.finalize().checksum_crc64nvme + } + + /// A body that sends its first chunk, then holds back its last one until every writer + /// counted in `arrived` has sent a first chunk. A writer streams its body only after its early + /// condition check, so once the gate opens every writer has passed that check and they all + /// reach the publish step together. + struct GatedBody { + chunks: VecDeque, + arrived: Arc, + writers: usize, + counted: bool, + } + + impl Stream for GatedBody { + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.get_mut(); + if this.chunks.len() == 1 { + if !this.counted { + this.arrived.fetch_add(1, AtomicOrdering::SeqCst); + this.counted = true; + } + if this.arrived.load(AtomicOrdering::SeqCst) < this.writers { + cx.waker().wake_by_ref(); + return Poll::Pending; + } + } + Poll::Ready(this.chunks.pop_front().map(Ok)) + } + } + + fn gated_put(key: &str, body: &[u8], arrived: &Arc, writers: usize) -> PutObjectInput { + let (head, tail) = body.split_at(1); + let stream = GatedBody { + chunks: VecDeque::from([Bytes::copy_from_slice(head), Bytes::copy_from_slice(tail)]), + arrived: Arc::clone(arrived), + writers, + counted: false, + }; + PutObjectInput::builder() + .bucket("bucket".to_owned()) + .key(key.to_owned()) + .body(Some(StreamingBlob::wrap(stream))) + .checksum_crc64nvme(crc64nvme(body)) + .content_type(Some(String::from_utf8(body.to_vec()).unwrap())) + .build() + .unwrap() + } + + /// Runs every put at once. A writer that errors before the gate leaves the others waiting at + /// it for good, so each one is bounded. + async fn race_puts(fs: &Arc, inputs: Vec) -> Vec>> { + let tasks: Vec<_> = inputs + .into_iter() + .map(|input| { + let fs = Arc::clone(fs); + tokio::spawn(async move { fs.put_object(request(input)).await }) + }) + .collect(); + let mut results = Vec::with_capacity(tasks.len()); + for task in tasks { + let joined = tokio::time::timeout(Duration::from_secs(60), task) + .await + .expect("a writer did not finish within 60 s"); + results.push(joined.unwrap()); + } + results + } + + /// An object as one read returned it. Every test body is also written as its content type, + /// so a read pairs correctly only when its body, checksum and attributes all come from one + /// write. + struct Stored { + body: Vec, + checksum: Option, + content_type: Option, + } + + impl Stored { + fn is_paired(&self) -> bool { + self.checksum == crc64nvme(&self.body) && self.content_type.as_deref().map(str::as_bytes) == Some(&self.body[..]) + } + } + + async fn get(fs: &FileSystem, key: &str) -> Stored { + let input = GetObjectInput::builder() + .bucket("bucket".to_owned()) + .key(key.to_owned()) + .build() + .unwrap(); + let output = fs.get_object(request(input)).await.unwrap().output; + let mut body = output.body.unwrap(); + let mut bytes = Vec::new(); + while let Some(chunk) = body.next().await { + bytes.extend_from_slice(&chunk.unwrap()); + } + Stored { + body: bytes, + checksum: output.checksum_crc64nvme, + content_type: output.content_type, + } + } + + /// The single successful writer among `results`, after checking that every other one was + /// refused with `PreconditionFailed`. + fn sole_winner(results: &[S3Result>]) -> usize { + let winners: Vec = (0..results.len()).filter(|&i| results[i].is_ok()).collect(); + assert_eq!(winners.len(), 1, "exactly one conditional writer may succeed, got {}", winners.len()); + for result in results { + if let Err(err) = result { + assert_eq!(err.code(), &S3ErrorCode::PreconditionFailed); + } + } + winners[0] + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn concurrent_create_if_absent_has_one_winner() { + const WRITERS: usize = 16; + let (_root, fs) = new_fs("create-race"); + let arrived = Arc::new(AtomicUsize::new(0)); + let bodies: Vec> = (0..WRITERS).map(|i| format!("writer {i:02} record").into_bytes()).collect(); + let inputs = bodies + .iter() + .map(|body| { + let mut input = gated_put("prov", body, &arrived, WRITERS); + input.if_none_match = Some(ETagCondition::Any); + input + }) + .collect(); + + let winner = sole_winner(&race_puts(&fs, inputs).await); + + let stored = get(&fs, "prov").await; + assert_eq!(stored.body, bodies[winner], "the stored body is the winner's"); + assert!(stored.is_paired(), "the stored checksum and attributes belong to the stored body"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn concurrent_if_match_has_one_winner() { + const WRITERS: usize = 16; + let (_root, fs) = new_fs("cas-race"); + let seed = gated_put("record", b"version 0", &Arc::new(AtomicUsize::new(0)), 1); + let etag = fs.put_object(request(seed)).await.unwrap().output.e_tag.unwrap(); + let arrived = Arc::new(AtomicUsize::new(0)); + let bodies: Vec> = (0..WRITERS) + .map(|i| format!("version 1 from writer {i:02}").into_bytes()) + .collect(); + let inputs = bodies + .iter() + .map(|body| { + let mut input = gated_put("record", body, &arrived, WRITERS); + input.if_match = Some(ETagCondition::ETag(etag.clone())); + input + }) + .collect(); + + let winner = sole_winner(&race_puts(&fs, inputs).await); + + let stored = get(&fs, "record").await; + assert_eq!(stored.body, bodies[winner], "the stored body is the winner's"); + assert!(stored.is_paired(), "the stored checksum and attributes belong to the stored body"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn overwrites_and_reads_pair_each_body_with_its_checksum() { + const WRITERS: usize = 8; + const ROUNDS: usize = 40; + let (_root, fs) = new_fs("overwrite-race"); + let seed = gated_put("object", b"seed", &Arc::new(AtomicUsize::new(0)), 1); + fs.put_object(request(seed)).await.unwrap(); + + let stop = Arc::new(AtomicBool::new(false)); + let readers: Vec<_> = (0..2) + .map(|_| { + let fs = Arc::clone(&fs); + let stop = Arc::clone(&stop); + tokio::spawn(async move { + let (mut total, mut torn) = (0_usize, 0_usize); + while !stop.load(AtomicOrdering::SeqCst) { + total += 1; + if !get(&fs, "object").await.is_paired() { + torn += 1; + } + } + (total, torn) + }) + }) + .collect(); + + for round in 0..ROUNDS { + let arrived = Arc::new(AtomicUsize::new(0)); + let inputs = (0..WRITERS) + .map(|i| gated_put("object", format!("round {round:02} writer {i:02}").as_bytes(), &arrived, WRITERS)) + .collect(); + for result in race_puts(&fs, inputs).await { + result.unwrap(); + } + assert!( + get(&fs, "object").await.is_paired(), + "round {round}: the last write's body, checksum and attributes differ" + ); + } + + stop.store(true, AtomicOrdering::SeqCst); + for reader in readers { + let (total, torn) = reader.await.unwrap(); + assert!(total > 0, "a reader never ran"); + assert_eq!(torn, 0, "{torn} of {total} reads paired a body with another write's files"); + } + } } From 0b0ae884853cf18eaf50f43e58446b49f58998ad Mon Sep 17 00:00:00 2001 From: Panagiotis Moustafellos Date: Mon, 2 Nov 2026 14:40:40 +0200 Subject: [PATCH 2/5] fix(s3s-fs): publish an object and its checksums under one lock An object is three files: the body, its attributes and its internal info, which holds the stored checksums and ETag. put_object and complete_multipart_upload checked If-None-Match and If-Match once, before streaming the body, then renamed the body into place and wrote the two sidecars afterwards, each with its own rename. Writers that arrived together all passed the check, so several create-if-absent or If-Match writers could each get 200, and the last body rename and the last internal-info rename could come from different writers. A read could also open a new body and load the old sidecars. Ravel's benchmark hit this: concurrent create-if-absent writes of one key left a body whose stored CRC64-NVME belonged to another write, and every later GET of it failed the client's checksum check (NOFireAI/ravel#3652). Every object now maps by hash onto one of 1024 read-write locks. - put_object, copy_object and complete_multipart_upload write all three files as flushed temporary files first. They then take the object's write lock, check their conditions again, and rename the files in one blocking task that owns the lock. A full disk fails the write before anything is published, and a dropped request cannot stop the renames half way. The early condition check stays as a fast refusal. POST Object goes through put_object, and a directory key takes the same lock for its check and its directory. - copy_object takes the destination's write lock and the source's read lock in index order, so two copies in opposite directions cannot deadlock. It no longer copies over the destination in place, which truncated a body a reader could still be streaming. When the source has no stored ETag, the destination's ETag is now the digest of the source instead of a re-read of the destination. - get_object, head_object and upload_part_copy open the body and load the sidecars under the read lock. The body then streams from the open file, so it stays the version the checksums describe. - delete_object and delete_objects hold the write lock while they remove an object. - An ETag computed for an object with none stored is written back to its internal info. That write used to run without any ordering against a writer, so it could put the old body's ETag and checksums beside a new body. Every caller now holds the object's lock, including the early condition check, which takes the read lock. The lock is per process, which matches s3s-fs's one-process model. Listings do not read the sidecars, so they stay unlocked. Refs: NOFireAI/ravel#3652 Signed-off-by: Panagiotis Moustafellos --- crates/s3s-fs/Cargo.toml | 2 +- crates/s3s-fs/src/fs.rs | 167 ++++++++++++++++++++++++++++++++-- crates/s3s-fs/src/s3.rs | 191 ++++++++++++++++++++++++--------------- 3 files changed, 278 insertions(+), 82 deletions(-) diff --git a/crates/s3s-fs/Cargo.toml b/crates/s3s-fs/Cargo.toml index 6babf7055..504ee5b23 100644 --- a/crates/s3s-fs/Cargo.toml +++ b/crates/s3s-fs/Cargo.toml @@ -85,7 +85,7 @@ serde.workspace = true serde_json.workspace = true std-next.workspace = true thiserror.workspace = true -tokio = { workspace = true, features = ["fs", "io-util", "time"] } +tokio = { workspace = true, features = ["fs", "io-util", "rt", "sync", "time"] } tokio-util = { workspace = true, features = ["io"] } tracing.workspace = true tracing-error.workspace = true diff --git a/crates/s3s-fs/src/fs.rs b/crates/s3s-fs/src/fs.rs index 543ffcef2..42f15f06a 100644 --- a/crates/s3s-fs/src/fs.rs +++ b/crates/s3s-fs/src/fs.rs @@ -11,23 +11,44 @@ use s3s::dto; use s3s::dto::PartNumber; use std::env; +use std::fmt; +use std::hash::{DefaultHasher, Hash, Hasher}; use std::ops::Not; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::SystemTime; use tokio::fs; use tokio::fs::File; use tokio::io::{AsyncReadExt, AsyncWriteExt, BufWriter}; +use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock}; +use numeric_cast::NumericCast; use path_absolutize::Absolutize; use tracing::warn; use uuid::Uuid; +/// How many locks the objects share; see [`FileSystem::lock_object_write`]. +const OBJECT_LOCK_STRIPES: u64 = 1024; + #[derive(Debug)] pub struct FileSystem { pub(crate) root: PathBuf, tmp_file_counter: AtomicU64, + object_locks: ObjectLocks, +} + +/// The locks that objects share by hash. Its `Debug` output stays short because every request +/// span records the file system. +struct ObjectLocks(Box<[Arc>]>); + +impl fmt::Debug for ObjectLocks { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ObjectLocks") + .field("len", &self.0.len()) + .finish_non_exhaustive() + } } pub(crate) type InternalInfo = serde_json::Map; @@ -123,7 +144,67 @@ impl FileSystem { let root = env::current_dir()?.join(root).canonicalize()?; clean_old_tmp_files(&root)?; let tmp_file_counter = AtomicU64::new(0); - Ok(Self { root, tmp_file_counter }) + let object_locks = ObjectLocks((0..OBJECT_LOCK_STRIPES).map(|_| Arc::new(RwLock::new(()))).collect()); + Ok(Self { + root, + tmp_file_counter, + object_locks, + }) + } + + fn object_lock(&self, bucket: &str, key: &str) -> (usize, Arc>) { + let mut hasher = DefaultHasher::new(); + bucket.hash(&mut hasher); + key.hash(&mut hasher); + let index = (hasher.finish() % OBJECT_LOCK_STRIPES).numeric_cast::(); + (index, Arc::clone(&self.object_locks.0[index])) + } + + /// Locks an object for writing. + /// + /// An object is three files: the body, its attributes and its internal info, which holds the + /// stored checksums and `ETag`. A write stages all three as temporary files, then checks its + /// conditions under this lock and renames them into place with [`publish`], and a read opens + /// the body and loads both sidecars under [`Self::lock_object_read`]. So two conditional + /// writers cannot both pass their check, and a reader never pairs one write's body with + /// another write's checksums. Objects share a fixed set of locks by hash: unrelated objects + /// can contend, but the set never grows. + /// + /// The lock is not reentrant: a task that holds it for an object must not lock that object + /// again. + pub(crate) async fn lock_object_write(&self, bucket: &str, key: &str) -> OwnedRwLockWriteGuard<()> { + self.object_lock(bucket, key).1.write_owned().await + } + + /// Locks an object for reading; see [`Self::lock_object_write`]. + pub(crate) async fn lock_object_read(&self, bucket: &str, key: &str) -> OwnedRwLockReadGuard<()> { + self.object_lock(bucket, key).1.read_owned().await + } + + /// Locks a copy's destination for writing and its source for reading. The two are taken in + /// index order, so two copies in opposite directions cannot deadlock. A source that shares the + /// destination's lock is covered by the write lock alone. + pub(crate) async fn lock_object_copy( + &self, + src_bucket: &str, + src_key: &str, + dst_bucket: &str, + dst_key: &str, + ) -> (Option>, OwnedRwLockWriteGuard<()>) { + let (src_index, src_lock) = self.object_lock(src_bucket, src_key); + let (dst_index, dst_lock) = self.object_lock(dst_bucket, dst_key); + if src_index == dst_index { + return (None, dst_lock.write_owned().await); + } + if src_index < dst_index { + let read = src_lock.read_owned().await; + let write = dst_lock.write_owned().await; + (Some(read), write) + } else { + let write = dst_lock.write_owned().await; + let read = src_lock.read_owned().await; + (Some(read), write) + } } pub(crate) fn resolve_abs_path(&self, path: impl AsRef) -> Result { @@ -284,6 +365,8 @@ impl FileSystem { /// when it was opened. Without that check a request racing with a replacement would pin an `ETag` /// that never matches the new body, while an unremembered value only stays stale until the next /// request, as it did before. + /// + /// The caller holds the object's read or write lock, which [`Self::remember_e_tag`] needs. pub(crate) async fn get_md5_sum(&self, bucket: &str, key: &str) -> Result { let object_path = self.get_object_path(bucket, key)?; let mut file = File::open(&object_path).await?; @@ -318,6 +401,7 @@ impl FileSystem { /// /// The stored tag is preferred; a stored one is absent for objects written /// before it was recorded or by another writer, so the body is hashed then. + /// The caller holds the object's read or write lock, as for [`Self::get_md5_sum`]. pub(crate) async fn current_etag(&self, bucket: &str, key: &str) -> Result { let info = self.load_internal_info(bucket, key).await?; let value = match info.as_ref().and_then(crate::checksum::load_e_tag) { @@ -329,11 +413,14 @@ impl FileSystem { /// Write a computed `e_tag` into the object's `internal.json`, best effort. /// - /// The sidecar is re-read here and merged, so a checksum that a concurrent write stored in the - /// meantime is kept; only a missing `e_tag` is filled in. The caller already holds the value it - /// needed, so any failure is logged and ignored instead of changing the response. A concurrent - /// writer can still replace the sidecar between this read and the write, in which case the - /// later write wins, exactly as it does for two writers of the same object today. + /// The sidecar is re-read here and merged, so the checksums it holds are kept; only a missing + /// `e_tag` is filled in. The caller already holds the value it needed, so any failure is logged + /// and ignored instead of changing the response. + /// + /// The caller holds the object's read or write lock, so no write can publish the object between + /// this read and the rename below. A read lock is enough: other holders only read the object + /// or remember the same digest of the same body, and the rename replaces the sidecar + /// atomically. async fn remember_e_tag(&self, bucket: &str, key: &str, e_tag: &str) { let result: Result<()> = async { let path = self.get_internal_info_path(bucket, key)?; @@ -441,6 +528,23 @@ impl FileSystem { } } + /// Writes `content` to a temporary file to be renamed over `path` by [`publish`]. + pub(crate) async fn stage_bytes(&self, path: &Path, content: &[u8]) -> Result { + let mut file_writer = self.prepare_file_write(path).await?; + file_writer.writer().write_all(content).await?; + file_writer.stage().await + } + + pub(crate) async fn stage_object_attributes(&self, bucket: &str, key: &str, attrs: &ObjectAttributes) -> Result { + let path = self.get_metadata_path(bucket, key, None)?; + self.stage_bytes(&path, &serde_json::to_vec(attrs)?).await + } + + pub(crate) async fn stage_internal_info(&self, bucket: &str, key: &str, info: &InternalInfo) -> Result { + let path = self.get_internal_info_path(bucket, key)?; + self.stage_bytes(&path, &serde_json::to_vec(info)?).await + } + /// Write to the filesystem atomically. /// This is done by first writing to a temporary location and then moving the file. pub(crate) async fn prepare_file_write<'a>(&self, path: &'a Path) -> Result> { @@ -490,6 +594,57 @@ impl<'a> FileWriter<'a> { self.clean_tmp = false; Ok(()) } + + /// Flushes the temporary file and hands it to [`publish`] instead of renaming it now. + pub(crate) async fn stage(mut self) -> Result { + self.writer.flush().await?; + self.clean_tmp = false; + Ok(Staged { + tmp_path: std::mem::take(&mut self.tmp_path), + dest_path: self.dest_path.to_owned(), + }) + } +} + +/// A flushed temporary file waiting to be renamed over its destination by [`publish`]. Dropped +/// unpublished, it removes the temporary file. +pub(crate) struct Staged { + tmp_path: PathBuf, + dest_path: PathBuf, +} + +impl Staged { + pub(crate) fn tmp_path(&self) -> &Path { + &self.tmp_path + } +} + +impl Drop for Staged { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.tmp_path); + } +} + +/// Renames each staged file over its destination, in order, while `guard` is held. +/// +/// The renames run in one blocking task that owns the guard, so once they start they all run +/// with the lock held, even if the request that started them is dropped: a cancelled write cannot +/// leave a body beside the previous write's sidecars, and no reader sees the files half +/// published. The data was written and flushed when it was staged, so a full disk fails the +/// write before anything is published. +pub(crate) async fn publish(guard: G, files: Vec) -> Result<()> { + tokio::task::spawn_blocking(move || { + let renamed = files.iter().try_for_each(|file| { + if let Some(dir) = file.dest_path.parent() { + std::fs::create_dir_all(dir)?; + } + std::fs::rename(&file.tmp_path, &file.dest_path) + }); + drop(guard); + renamed + }) + .await??; + Ok(()) } impl Drop for FileWriter<'_> { diff --git a/crates/s3s-fs/src/s3.rs b/crates/s3s-fs/src/s3.rs index ec4396936..400141d61 100644 --- a/crates/s3s-fs/src/s3.rs +++ b/crates/s3s-fs/src/s3.rs @@ -47,7 +47,7 @@ use uuid::Uuid; /// size; 64 KiB is a stable point to pair with connection limiting later on. const READ_CHUNK_SIZE: usize = 64 * 1024; -/// Copy chunk size for assembling a multipart object out of its parts. +/// Copy chunk size for assembling a multipart object out of its parts, and for copying an object. /// /// One buffer is reused for every read from a part and every write to the temporary object, so /// copying an object of `n` bytes costs roughly `2n / COPY_CHUNK_SIZE` syscalls and the buffer is @@ -293,6 +293,44 @@ fn completed_part_checksum(part: &CompletedPart) -> s3s::dto::Checksum { } } +impl FileSystem { + /// Checks a write's `If-None-Match` and `If-Match` conditions against the stored object. + /// + /// A write calls this before it streams its body, as an early refusal, and again under the + /// object's write lock, where the answer cannot change before the write publishes. The caller + /// holds the object's read or write lock either way, because an `ETag` computed here is + /// remembered in the object's internal info. + async fn check_write_conditions( + &self, + bucket: &str, + key: &str, + object_path: &Path, + if_none_match: Option<&ETagCondition>, + if_match: Option<&ETagCondition>, + ) -> S3Result<()> { + // If-None-Match: * means "only write if the object does not exist". A tag + // or a list is not implemented, which is what Amazon S3 answers for it. + if let Some(condition) = if_none_match { + if !condition.is_any() { + return Err(s3_error!(NotImplemented)); + } + if object_path.exists() { + return Err(s3_error!(PreconditionFailed, "Object already exists")); + } + } + // If-Match: means "only overwrite if ETag matches" (CAS). + if let Some(condition) = if_match { + if !object_path.exists() { + return Err(s3_error!(PreconditionFailed, "Object does not exist")); + } + if !condition.is_any() && !condition.matches_strong(&self.current_etag(bucket, key).await?) { + return Err(s3_error!(PreconditionFailed, "ETag does not match")); + } + } + Ok(()) + } +} + #[async_trait::async_trait] impl S3 for FileSystem { #[tracing::instrument] @@ -321,6 +359,7 @@ impl S3 for FileSystem { let src_path = self.get_object_path(bucket, key)?; let dst_path = self.get_object_path(&input.bucket, &input.key)?; + let object_guards = self.lock_object_copy(bucket, key, &input.bucket, &input.key).await; if src_path.exists().not() { return Err(s3_error!(NoSuchKey)); } @@ -369,12 +408,18 @@ impl S3 for FileSystem { return Err(s3_error!(PreconditionFailed)); } - if let Some(dir_path) = dst_path.parent() { - try_!(fs::create_dir_all(&dir_path).await); - } + // Derive the destination ETag from the source ETag when available. + // This preserves non-MD5 ETag formats (e.g., multipart `{hash}-{part_count}`) + // and avoids re-hashing the copied bytes when the source has a stored ETag. + let dst_etag_str = match src_etag { + Some(etag) => etag.into_value(), + None => self.get_md5_sum(bucket, key).await?, + }; - // `fs::copy(p, p)` truncates the file before reading it, so self-replace - // must preserve bytes in place while still updating LastModified. + // The destination's files are staged and renamed into place, never rewritten in place, so + // a reader still streaming the old body keeps reading the old bytes. A self-replace leaves + // the body untouched and only updates LastModified. + let mut staged = Vec::new(); let dst_last_modified = if src_path == dst_path { let now = SystemTime::now(); let file = try_!(std::fs::OpenOptions::new().write(true).open(&dst_path)); @@ -382,20 +427,18 @@ impl S3 for FileSystem { debug!(path = %dst_path.display(), "replace file in place"); Timestamp::from(now) } else { - let _ = try_!(fs::copy(&src_path, &dst_path).await); + let src_file = try_!(fs::File::open(&src_path).await); + let mut src_reader = tokio::io::BufReader::with_capacity(COPY_CHUNK_SIZE, src_file); + let mut file_writer = self.prepare_file_write(&dst_path).await?; + let _ = try_!(tokio::io::copy_buf(&mut src_reader, file_writer.writer()).await); + let body_file = file_writer.stage().await?; debug!(from = %src_path.display(), to = %dst_path.display(), "copy file"); - let dst_metadata = try_!(fs::metadata(&dst_path).await); + // A rename keeps the modification time, so the staged file already has the final one. + let dst_metadata = try_!(fs::metadata(body_file.tmp_path()).await); + staged.push(body_file); Timestamp::from(try_!(dst_metadata.modified())) }; - // Derive the destination ETag from the source ETag when available. - // This preserves non-MD5 ETag formats (e.g., multipart `{hash}-{part_count}`) - // and avoids re-hashing the destination file. - let dst_etag_str = match src_etag { - Some(etag) => etag.into_value(), - None => self.get_md5_sum(&input.bucket, &input.key).await?, - }; - // `MetadataDirective` defaults to `COPY` per AWS API: when the // header is absent the destination should inherit the source's // metadata sidecar verbatim. When set to `REPLACE`, the @@ -422,16 +465,15 @@ impl S3 for FileSystem { checksum_type: None, }; dst_attrs.expires = input.expires; - self.save_object_attributes(&input.bucket, &input.key, &dst_attrs, None) - .await?; + staged.push(self.stage_object_attributes(&input.bucket, &input.key, &dst_attrs).await?); } else { let src_metadata_path = self.get_metadata_path(bucket, key, None)?; if src_metadata_path.exists() { let dst_metadata_path = self.get_metadata_path(&input.bucket, &input.key, None)?; - // Same self-replace guard as for the payload above — `fs::copy` - // would zero the metadata sidecar when src == dst. + // A self-replace keeps the sidecar it already has. if src_metadata_path != dst_metadata_path { - let _ = try_!(fs::copy(src_metadata_path, dst_metadata_path).await); + let content = try_!(fs::read(&src_metadata_path).await); + staged.push(self.stage_bytes(&dst_metadata_path, &content).await?); } } } @@ -439,8 +481,9 @@ impl S3 for FileSystem { { let mut info = src_info.unwrap_or_default(); crate::checksum::save_e_tag(&mut info, &dst_etag_str); - self.save_internal_info(&input.bucket, &input.key, &info).await?; + staged.push(self.stage_internal_info(&input.bucket, &input.key, &info).await?); } + crate::fs::publish(object_guards, staged).await?; let copy_object_result = CopyObjectResult { e_tag: Some(ETag::Strong(dst_etag_str)), @@ -471,6 +514,7 @@ impl S3 for FileSystem { async fn delete_object(&self, req: S3Request) -> S3Result> { let input = req.input; let path = self.get_object_path(&input.bucket, &input.key)?; + let object_guard = self.lock_object_write(&input.bucket, &input.key).await; if path.exists().not() { if self.get_bucket_path(&input.bucket)?.exists().not() { return Err(s3_error!(NoSuchBucket)); @@ -487,6 +531,7 @@ impl S3 for FileSystem { } else { try_!(fs::remove_file(&path).await); } + drop(object_guard); let output = DeleteObjectOutput::default(); // TODO: handle other fields Ok(S3Response::new(output)) } @@ -498,6 +543,7 @@ impl S3 for FileSystem { let mut deleted_objects: Vec = Vec::new(); for object in input.delete.objects { let path = self.get_object_path(&input.bucket, &object.key)?; + let object_guard = self.lock_object_write(&input.bucket, &object.key).await; if object.key.ends_with('/') { match fs::read_dir(&path).await { Ok(mut dir) => { @@ -520,6 +566,7 @@ impl S3 for FileSystem { } } } + drop(object_guard); let deleted_object = DeletedObject { key: Some(object.key), @@ -555,6 +602,9 @@ impl S3 for FileSystem { let input = req.input; let object_path = self.get_object_path(&input.bucket, &input.key)?; + // The body streams from the file opened here after the lock is released, so it stays the + // version that the sidecars loaded under the same lock describe. + let object_guard = self.lock_object_read(&input.bucket, &input.key).await; let mut file = fs::File::open(&object_path).await.map_err(|e| s3_error!(e, NoSuchKey))?; let file_metadata = try_!(file.metadata().await); @@ -594,6 +644,7 @@ impl S3 for FileSystem { Some(e_tag) => e_tag, None => self.get_md5_sum(&input.bucket, &input.key).await?, }; + drop(object_guard); let current = ETag::Strong(md5_sum.clone()); match evaluate_read_condition(input.if_match.as_ref(), input.if_none_match.as_ref(), ¤t) { @@ -665,6 +716,7 @@ impl S3 for FileSystem { let input = req.input; let path = self.get_object_path(&input.bucket, &input.key)?; + let object_guard = self.lock_object_read(&input.bucket, &input.key).await; if !path.exists() { if self.get_bucket_path(&input.bucket)?.exists().not() { return Err(s3_error!(NoSuchBucket)); @@ -687,6 +739,7 @@ impl S3 for FileSystem { Some(e_tag) => e_tag, None => self.get_md5_sum(&input.bucket, &input.key).await?, }; + drop(object_guard); let current = ETag::Strong(md5_sum.clone()); match evaluate_read_condition(input.if_match.as_ref(), input.if_none_match.as_ref(), ¤t) { @@ -881,28 +934,14 @@ impl S3 for FileSystem { let Some(body) = body else { return Err(s3_error!(IncompleteBody)) }; - // Check conditional headers before modifying any state. - // If-None-Match: * means "only create if the object doesn't exist". - // If-Match: means "only overwrite if ETag matches" (CAS). + // Check conditional headers before modifying any state. They are checked again under the + // object's write lock before publishing; this early refusal only saves streaming a body + // that cannot be stored. let object_path = self.get_object_path(&bucket, &key)?; - // If-None-Match: * means "only write if the object does not exist". A tag - // or a list is not implemented, which is what Amazon S3 answers for it. - if let Some(ref condition) = if_none_match { - if !condition.is_any() { - return Err(s3_error!(NotImplemented)); - } - if object_path.exists() { - return Err(s3_error!(PreconditionFailed, "Object already exists")); - } - } - if let Some(ref condition) = if_match { - if !object_path.exists() { - return Err(s3_error!(PreconditionFailed, "Object does not exist")); - } - if !condition.is_any() && !condition.matches_strong(&self.current_etag(&bucket, &key).await?) { - return Err(s3_error!(PreconditionFailed, "ETag does not match")); - } - } + let object_guard = self.lock_object_read(&bucket, &key).await; + self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) + .await?; + drop(object_guard); let mut checksum: s3s::checksum::ChecksumHasher = default(); if input.checksum_crc32.is_some() { @@ -957,7 +996,11 @@ impl S3 for FileSystem { { return Err(s3_error!(UnexpectedContent, "Unexpected request body when creating a directory object.")); } + let object_guard = self.lock_object_write(&bucket, &key).await; + self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) + .await?; try_!(fs::create_dir_all(&object_path).await); + drop(object_guard); let output = PutObjectOutput::default(); return Ok(S3Response::new(output)); } @@ -1056,9 +1099,7 @@ impl S3 for FileSystem { return Err(s3_error!(BadDigest, "checksum_xxhash128 mismatch")); } - file_writer.done().await?; - - debug!(path = %object_path.display(), ?size, %md5_sum, ?checksum, "write file"); + let body_file = file_writer.stage().await?; // Save object attributes (including user metadata and standard attributes) let mut obj_attrs = ObjectAttributes { @@ -1074,12 +1115,19 @@ impl S3 for FileSystem { checksum_type: None, }; obj_attrs.expires = expires; - self.save_object_attributes(&bucket, &key, &obj_attrs, None).await?; + let attrs_file = self.stage_object_attributes(&bucket, &key, &obj_attrs).await?; let mut info: InternalInfo = default(); crate::checksum::save_e_tag(&mut info, &md5_sum); crate::checksum::modify_internal_info(&mut info, &checksum); - self.save_internal_info(&bucket, &key, &info).await?; + let info_file = self.stage_internal_info(&bucket, &key, &info).await?; + + let object_guard = self.lock_object_write(&bucket, &key).await; + self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) + .await?; + crate::fs::publish(object_guard, vec![body_file, attrs_file, info_file]).await?; + + debug!(path = %object_path.display(), ?size, %md5_sum, ?checksum, "write file"); let output = PutObjectOutput { e_tag: Some(ETag::Strong(md5_sum)), @@ -1299,7 +1347,9 @@ impl S3 for FileSystem { let src_path = self.get_object_path(src_bucket, src_key)?; let dst_path = self.resolve_upload_part_path(upload_id, part_number)?; + let src_guard = self.lock_object_read(src_bucket, src_key).await; let mut src_file = fs::File::open(&src_path).await.map_err(|e| s3_error!(e, NoSuchKey))?; + drop(src_guard); let file_len = try_!(src_file.metadata().await).len(); let (start, content_length) = if let Some(copy_range) = &input.copy_source_range { @@ -1502,26 +1552,14 @@ impl S3 for FileSystem { return Err(s3_error!(NotImplemented, "Unsupported multipart checksum type")); } - // Check conditional headers before modifying any state + // Check conditional headers before modifying any state. They are checked again under the + // object's write lock before publishing; this early refusal only saves assembling an + // object that cannot be stored. let object_path = self.get_object_path(&bucket, &key)?; - // If-None-Match: * means "only write if the object does not exist". A tag - // or a list is not implemented, which is what Amazon S3 answers for it. - if let Some(ref condition) = if_none_match { - if !condition.is_any() { - return Err(s3_error!(NotImplemented)); - } - if object_path.exists() { - return Err(s3_error!(PreconditionFailed, "Object already exists")); - } - } - if let Some(ref condition) = if_match { - if !object_path.exists() { - return Err(s3_error!(PreconditionFailed, "Object does not exist")); - } - if !condition.is_any() && !condition.matches_strong(&self.current_etag(&bucket, &key).await?) { - return Err(s3_error!(PreconditionFailed, "ETag does not match")); - } - } + let object_guard = self.lock_object_read(&bucket, &key).await; + self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) + .await?; + drop(object_guard); let expected_checksum = s3s::dto::Checksum { checksum_crc32, @@ -1582,7 +1620,7 @@ impl S3 for FileSystem { try_!(file_writer.writer().write_all(&buf[..nread]).await); size += nread as u64; } - // No flush per part: `done` flushes once, before it renames the temporary object. + // No flush per part: `stage` flushes once, before the temporary object is published. part_md5_hashes.push(part_md5.finalize()); if part_number != total_parts_cnt && size < 5 * 1024 * 1024 { @@ -1605,21 +1643,24 @@ impl S3 for FileSystem { return Err(s3_error!(BadDigest, "{} mismatch", field)); } - file_writer.done().await?; - + let mut staged = vec![file_writer.stage().await?]; if let Some(attrs) = &upload_attrs { - self.save_object_attributes(&bucket, &key, attrs, None).await?; + staged.push(self.stage_object_attributes(&bucket, &key, attrs).await?); } - - debug!(?e_tag, path = %object_path.display(), "multipart etag"); - { let mut info: InternalInfo = default(); crate::checksum::save_e_tag(&mut info, &e_tag); crate::checksum::modify_internal_info(&mut info, &checksum); - self.save_internal_info(&bucket, &key, &info).await?; + staged.push(self.stage_internal_info(&bucket, &key, &info).await?); } + let object_guard = self.lock_object_write(&bucket, &key).await; + self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) + .await?; + crate::fs::publish(object_guard, staged).await?; + + debug!(?e_tag, path = %object_path.display(), "multipart etag"); + // The object is committed. Its upload directory holds the parts, their metadata and the // upload record, so removing the directory retires all of them at once; if that fails, the // directory stays behind and a later abort retries the cleanup. Cleanup failure never turns From e1a3529057a0ec67bf90c219ebdd28683d2ec369 Mon Sep 17 00:00:00 2001 From: Panagiotis Moustafellos Date: Mon, 2 Nov 2026 15:41:02 +0200 Subject: [PATCH 3/5] test(s3s-fs): race copies against reads and against each other Two tests for copy_object under concurrency: - One task copies two sources, with different bodies and content types, onto one destination in turn, 40 times, while two readers read the destination. Every read must pair a body with its own checksum and its own content type. - Two tasks copy A to B and B to A, 40 times each, at the same time. Both must finish within 60 s, and both objects must read back paired. The second test catches a copy that takes the locks of its two objects in an order that can deadlock. The reader loop moves into helpers shared with the overwrite test, and every copy and every task is bounded by 60 s. Refs: NOFireAI/ravel#3652 Signed-off-by: Panagiotis Moustafellos --- crates/s3s-fs/src/s3.rs | 138 ++++++++++++++++++++++++++++++++++------ 1 file changed, 117 insertions(+), 21 deletions(-) diff --git a/crates/s3s-fs/src/s3.rs b/crates/s3s-fs/src/s3.rs index 400141d61..bd3454e92 100644 --- a/crates/s3s-fs/src/s3.rs +++ b/crates/s3s-fs/src/s3.rs @@ -2236,6 +2236,7 @@ mod tests { use futures::{Stream, StreamExt}; use http::{Extensions, HeaderMap, Method, Uri}; use s3s::S3ErrorCode; + use tokio::task::JoinHandle; use uuid::Uuid; struct TestRoot(PathBuf); @@ -2883,6 +2884,62 @@ mod tests { } } + async fn put(fs: &FileSystem, key: &str, body: &[u8]) { + let input = gated_put(key, body, &Arc::new(AtomicUsize::new(0)), 1); + fs.put_object(request(input)).await.unwrap(); + } + + async fn copy(fs: &FileSystem, source: &str, destination: &str) { + let input = CopyObjectInput::builder() + .bucket("bucket".to_owned()) + .key(destination.to_owned()) + .copy_source(CopySource::Bucket { + bucket: "bucket".into(), + key: source.into(), + version_id: None, + }) + .build() + .unwrap(); + tokio::time::timeout(Duration::from_secs(60), fs.copy_object(request(input))) + .await + .expect("a copy did not finish within 60 s") + .unwrap(); + } + + /// Starts two readers that read `key` until `stop` is set. Each returns how many reads it made + /// and how many of them were not paired. + fn spawn_readers(fs: &Arc, key: &'static str, stop: &Arc) -> Vec> { + (0..2) + .map(|_| { + let fs = Arc::clone(fs); + let stop = Arc::clone(stop); + tokio::spawn(async move { + let (mut total, mut torn) = (0_usize, 0_usize); + while !stop.load(AtomicOrdering::SeqCst) { + total += 1; + if !get(&fs, key).await.is_paired() { + torn += 1; + } + } + (total, torn) + }) + }) + .collect() + } + + /// Stops the readers and checks that each one ran and that every read it made was paired. + async fn assert_reads_paired(readers: Vec>, stop: &AtomicBool) { + stop.store(true, AtomicOrdering::SeqCst); + for reader in readers { + let (total, torn) = tokio::time::timeout(Duration::from_secs(60), reader) + .await + .expect("a reader did not stop within 60 s") + .unwrap(); + assert!(total > 0, "a reader never ran"); + assert_eq!(torn, 0, "{torn} of {total} reads paired a body with another write's files"); + } + } + /// The single successful writer among `results`, after checking that every other one was /// refused with `PreconditionFailed`. fn sole_winner(results: &[S3Result>]) -> usize { @@ -2953,22 +3010,7 @@ mod tests { fs.put_object(request(seed)).await.unwrap(); let stop = Arc::new(AtomicBool::new(false)); - let readers: Vec<_> = (0..2) - .map(|_| { - let fs = Arc::clone(&fs); - let stop = Arc::clone(&stop); - tokio::spawn(async move { - let (mut total, mut torn) = (0_usize, 0_usize); - while !stop.load(AtomicOrdering::SeqCst) { - total += 1; - if !get(&fs, "object").await.is_paired() { - torn += 1; - } - } - (total, torn) - }) - }) - .collect(); + let readers = spawn_readers(&fs, "object", &stop); for round in 0..ROUNDS { let arrived = Arc::new(AtomicUsize::new(0)); @@ -2984,11 +3026,65 @@ mod tests { ); } - stop.store(true, AtomicOrdering::SeqCst); - for reader in readers { - let (total, torn) = reader.await.unwrap(); - assert!(total > 0, "a reader never ran"); - assert_eq!(torn, 0, "{torn} of {total} reads paired a body with another write's files"); + assert_reads_paired(readers, &stop).await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn copies_and_reads_pair_each_body_with_its_checksum() { + const ROUNDS: usize = 40; + let (_root, fs) = new_fs("copy-race"); + put(&fs, "source-a", b"the body of source a").await; + put(&fs, "source-b", b"a longer body, written to source b").await; + copy(&fs, "source-a", "copy").await; + + let stop = Arc::new(AtomicBool::new(false)); + let readers = spawn_readers(&fs, "copy", &stop); + + let copier = { + let fs = Arc::clone(&fs); + tokio::spawn(async move { + for round in 0..ROUNDS { + let source = if round % 2 == 0 { "source-b" } else { "source-a" }; + copy(&fs, source, "copy").await; + } + }) + }; + tokio::time::timeout(Duration::from_secs(60), copier) + .await + .expect("the copies did not finish within 60 s") + .unwrap(); + + assert_reads_paired(readers, &stop).await; + assert!(get(&fs, "copy").await.is_paired(), "the last copy's body, checksum and attributes differ"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn opposite_copies_finish_and_leave_both_objects_paired() { + const ROUNDS: usize = 40; + let (_root, fs) = new_fs("copy-swap"); + put(&fs, "left", b"the body of left").await; + put(&fs, "right", b"a longer body, written to right").await; + + let copiers: Vec<_> = [("left", "right"), ("right", "left")] + .into_iter() + .map(|(source, destination)| { + let fs = Arc::clone(&fs); + tokio::spawn(async move { + for _ in 0..ROUNDS { + copy(&fs, source, destination).await; + } + }) + }) + .collect(); + for copier in copiers { + tokio::time::timeout(Duration::from_secs(60), copier) + .await + .expect("copies in opposite directions did not finish within 60 s") + .unwrap(); + } + + for key in ["left", "right"] { + assert!(get(&fs, key).await.is_paired(), "{key}: the body, checksum and attributes differ"); } } } From 04b8bc71d8697316e9d903a3ef71f165edce5140 Mon Sep 17 00:00:00 2001 From: Panagiotis Moustafellos Date: Mon, 2 Nov 2026 15:42:14 +0200 Subject: [PATCH 4/5] fix(s3s-fs): hold only the source lock while a copy stages its files copy_object held the destination's write lock and the source's read lock for the whole copy: the condition checks, any ETag computation, the byte copy and the publish. Copying a large body blocked every read and write of the destination, and of every object sharing its lock, for as long as the copy ran. It also copied the bytes in user space instead of with fs::copy, which the kernel can speed up. A copy now takes only the source's read lock. Under it, the copy checks the source conditions, loads the source's internal info and attributes, computes a missing ETag, copies the body with fs::copy into a temporary file and stages both sidecars. It then releases the source lock and takes the destination's write lock only to publish the staged files. A copy never holds two locks at once, so the ordered locking of source and destination is gone. Every staged file is removed if the copy fails before it publishes. A copy onto itself still updates the object in place, so it holds the object's write lock throughout, as before. Refs: NOFireAI/ravel#3652 Signed-off-by: Panagiotis Moustafellos --- crates/s3s-fs/src/fs.rs | 40 +++++++--------------------------------- crates/s3s-fs/src/s3.rs | 31 ++++++++++++++++++++++--------- 2 files changed, 29 insertions(+), 42 deletions(-) diff --git a/crates/s3s-fs/src/fs.rs b/crates/s3s-fs/src/fs.rs index 42f15f06a..d5124bf77 100644 --- a/crates/s3s-fs/src/fs.rs +++ b/crates/s3s-fs/src/fs.rs @@ -152,12 +152,12 @@ impl FileSystem { }) } - fn object_lock(&self, bucket: &str, key: &str) -> (usize, Arc>) { + fn object_lock(&self, bucket: &str, key: &str) -> Arc> { let mut hasher = DefaultHasher::new(); bucket.hash(&mut hasher); key.hash(&mut hasher); let index = (hasher.finish() % OBJECT_LOCK_STRIPES).numeric_cast::(); - (index, Arc::clone(&self.object_locks.0[index])) + Arc::clone(&self.object_locks.0[index]) } /// Locks an object for writing. @@ -170,41 +170,15 @@ impl FileSystem { /// another write's checksums. Objects share a fixed set of locks by hash: unrelated objects /// can contend, but the set never grows. /// - /// The lock is not reentrant: a task that holds it for an object must not lock that object - /// again. + /// The lock is not reentrant, and objects share locks: a task that holds one must not lock + /// any object again until it has released it. pub(crate) async fn lock_object_write(&self, bucket: &str, key: &str) -> OwnedRwLockWriteGuard<()> { - self.object_lock(bucket, key).1.write_owned().await + self.object_lock(bucket, key).write_owned().await } /// Locks an object for reading; see [`Self::lock_object_write`]. pub(crate) async fn lock_object_read(&self, bucket: &str, key: &str) -> OwnedRwLockReadGuard<()> { - self.object_lock(bucket, key).1.read_owned().await - } - - /// Locks a copy's destination for writing and its source for reading. The two are taken in - /// index order, so two copies in opposite directions cannot deadlock. A source that shares the - /// destination's lock is covered by the write lock alone. - pub(crate) async fn lock_object_copy( - &self, - src_bucket: &str, - src_key: &str, - dst_bucket: &str, - dst_key: &str, - ) -> (Option>, OwnedRwLockWriteGuard<()>) { - let (src_index, src_lock) = self.object_lock(src_bucket, src_key); - let (dst_index, dst_lock) = self.object_lock(dst_bucket, dst_key); - if src_index == dst_index { - return (None, dst_lock.write_owned().await); - } - if src_index < dst_index { - let read = src_lock.read_owned().await; - let write = dst_lock.write_owned().await; - (Some(read), write) - } else { - let write = dst_lock.write_owned().await; - let read = src_lock.read_owned().await; - (Some(read), write) - } + self.object_lock(bucket, key).read_owned().await } pub(crate) fn resolve_abs_path(&self, path: impl AsRef) -> Result { @@ -632,7 +606,7 @@ impl Drop for Staged { /// leave a body beside the previous write's sidecars, and no reader sees the files half /// published. The data was written and flushed when it was staged, so a full disk fails the /// write before anything is published. -pub(crate) async fn publish(guard: G, files: Vec) -> Result<()> { +pub(crate) async fn publish(guard: OwnedRwLockWriteGuard<()>, files: Vec) -> Result<()> { tokio::task::spawn_blocking(move || { let renamed = files.iter().try_for_each(|file| { if let Some(dir) = file.dest_path.parent() { diff --git a/crates/s3s-fs/src/s3.rs b/crates/s3s-fs/src/s3.rs index bd3454e92..9c212026d 100644 --- a/crates/s3s-fs/src/s3.rs +++ b/crates/s3s-fs/src/s3.rs @@ -47,7 +47,7 @@ use uuid::Uuid; /// size; 64 KiB is a stable point to pair with connection limiting later on. const READ_CHUNK_SIZE: usize = 64 * 1024; -/// Copy chunk size for assembling a multipart object out of its parts, and for copying an object. +/// Copy chunk size for assembling a multipart object out of its parts. /// /// One buffer is reused for every read from a part and every write to the temporary object, so /// copying an object of `n` bytes costs roughly `2n / COPY_CHUNK_SIZE` syscalls and the buffer is @@ -359,7 +359,16 @@ impl S3 for FileSystem { let src_path = self.get_object_path(bucket, key)?; let dst_path = self.get_object_path(&input.bucket, &input.key)?; - let object_guards = self.lock_object_copy(bucket, key, &input.bucket, &input.key).await; + // A copy onto itself only updates the object in place, so it holds the object's write lock + // throughout. Any other copy stages the destination's files under the source's read lock + // and takes the destination's write lock only to publish them, so copying a large body + // never blocks the destination's readers and writers. No copy holds two locks at once. + let self_replace = src_path == dst_path; + let (src_guard, dst_guard) = if self_replace { + (None, Some(self.lock_object_write(&input.bucket, &input.key).await)) + } else { + (Some(self.lock_object_read(bucket, key).await), None) + }; if src_path.exists().not() { return Err(s3_error!(NoSuchKey)); } @@ -420,18 +429,15 @@ impl S3 for FileSystem { // a reader still streaming the old body keeps reading the old bytes. A self-replace leaves // the body untouched and only updates LastModified. let mut staged = Vec::new(); - let dst_last_modified = if src_path == dst_path { + let dst_last_modified = if self_replace { let now = SystemTime::now(); let file = try_!(std::fs::OpenOptions::new().write(true).open(&dst_path)); try_!(file.set_times(FileTimes::new().set_modified(now))); debug!(path = %dst_path.display(), "replace file in place"); Timestamp::from(now) } else { - let src_file = try_!(fs::File::open(&src_path).await); - let mut src_reader = tokio::io::BufReader::with_capacity(COPY_CHUNK_SIZE, src_file); - let mut file_writer = self.prepare_file_write(&dst_path).await?; - let _ = try_!(tokio::io::copy_buf(&mut src_reader, file_writer.writer()).await); - let body_file = file_writer.stage().await?; + let body_file = self.prepare_file_write(&dst_path).await?.stage().await?; + let _ = try_!(fs::copy(&src_path, body_file.tmp_path()).await); debug!(from = %src_path.display(), to = %dst_path.display(), "copy file"); // A rename keeps the modification time, so the staged file already has the final one. let dst_metadata = try_!(fs::metadata(body_file.tmp_path()).await); @@ -483,7 +489,14 @@ impl S3 for FileSystem { crate::checksum::save_e_tag(&mut info, &dst_etag_str); staged.push(self.stage_internal_info(&input.bucket, &input.key, &info).await?); } - crate::fs::publish(object_guards, staged).await?; + + drop(src_guard); + let dst_guard = if let Some(guard) = dst_guard { + guard + } else { + self.lock_object_write(&input.bucket, &input.key).await + }; + crate::fs::publish(dst_guard, staged).await?; let copy_object_result = CopyObjectResult { e_tag: Some(ETag::Strong(dst_etag_str)), From 2e428fe757b3b0f822ad5fae09e801b3526d6ab7 Mon Sep 17 00:00:00 2001 From: Panagiotis Moustafellos Date: Mon, 2 Nov 2026 15:42:25 +0200 Subject: [PATCH 5/5] perf(s3s-fs): skip the early condition check for unconditional writes put_object and complete_multipart_upload took the object's read lock for their early condition check even when the request carried no If-None-Match or If-Match. With no condition the check does nothing, so an unconditional write now skips it and that lock. Refs: NOFireAI/ravel#3652 Signed-off-by: Panagiotis Moustafellos --- crates/s3s-fs/src/s3.rs | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/crates/s3s-fs/src/s3.rs b/crates/s3s-fs/src/s3.rs index 9c212026d..3d4c36a2c 100644 --- a/crates/s3s-fs/src/s3.rs +++ b/crates/s3s-fs/src/s3.rs @@ -951,10 +951,12 @@ impl S3 for FileSystem { // object's write lock before publishing; this early refusal only saves streaming a body // that cannot be stored. let object_path = self.get_object_path(&bucket, &key)?; - let object_guard = self.lock_object_read(&bucket, &key).await; - self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) - .await?; - drop(object_guard); + if if_none_match.is_some() || if_match.is_some() { + let object_guard = self.lock_object_read(&bucket, &key).await; + self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) + .await?; + drop(object_guard); + } let mut checksum: s3s::checksum::ChecksumHasher = default(); if input.checksum_crc32.is_some() { @@ -1569,10 +1571,12 @@ impl S3 for FileSystem { // object's write lock before publishing; this early refusal only saves assembling an // object that cannot be stored. let object_path = self.get_object_path(&bucket, &key)?; - let object_guard = self.lock_object_read(&bucket, &key).await; - self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) - .await?; - drop(object_guard); + if if_none_match.is_some() || if_match.is_some() { + let object_guard = self.lock_object_read(&bucket, &key).await; + self.check_write_conditions(&bucket, &key, &object_path, if_none_match.as_ref(), if_match.as_ref()) + .await?; + drop(object_guard); + } let expected_checksum = s3s::dto::Checksum { checksum_crc32,