diff --git a/crates/storage/src/read.rs b/crates/storage/src/read.rs index 05a384e..4b4ba69 100644 --- a/crates/storage/src/read.rs +++ b/crates/storage/src/read.rs @@ -137,6 +137,12 @@ impl Storage { Storage::cleanup_tmp(&tmp_path).await; return Err(e.into()); } + if let Ok(f) = fs::File::open(&tmp_path).await + && let Err(e) = f.sync_data().await + { + Storage::cleanup_tmp(&tmp_path).await; + return Err(e.into()); + } let attrs = xattr::read_object(&src)?; if let Err(e) = xattr::set_object( &tmp_path, diff --git a/crates/storage/src/write.rs b/crates/storage/src/write.rs index f003145..c2df72e 100644 --- a/crates/storage/src/write.rs +++ b/crates/storage/src/write.rs @@ -74,6 +74,7 @@ impl Storage { if let Some(WriteCondition::IfMatch(expected)) = &condition { let dest = final_path.clone(); let expected = expected.clone(); + let any = expected == "*" || expected == "\"*\""; let check = tokio::task::spawn_blocking(move || xattr::get(&dest, xattr::ETAG)) .await .map_err(|e| StorageError::Io(std::io::Error::other(e)))?; @@ -85,6 +86,7 @@ impl Storage { key: key.into(), }); } + Some(_) if any => {} Some(actual) if actual.as_slice() != expected.as_bytes() => { Storage::cleanup_tmp(&tmp_path).await; return Err(StorageError::EtagMismatch); @@ -145,13 +147,16 @@ impl Storage { } // Common commit path: xattr write + rename in a single spawn_blocking. - // Keeps both blocking syscalls off the async thread and halves the number - // of thread-pool round-trips vs calling them separately. + // Holds an exclusive flock on `{dest}.lock` so If-Match / If-None-Match + // re-check the dest under the lock (check-then-rename was racy). let tmp = tmp_path.clone(); let dest = final_path.clone(); let etag_c = etag.clone(); let content_type = meta.content_type.clone(); let user_metadata = meta.user_metadata.clone(); + let cond = condition; + let bucket_owned = bucket.to_string(); + let key_owned = key.to_string(); tokio::task::spawn_blocking(move || { xattr::set_object( &tmp, @@ -160,7 +165,11 @@ impl Storage { meta.access, &user_metadata, )?; - std::fs::rename(&tmp, &dest).map_err(StorageError::Io) + if cond.is_some() { + commit_locked(&tmp, &dest, cond, bucket_owned, key_owned) + } else { + std::fs::rename(&tmp, &dest).map_err(StorageError::Io) + } }) .await .map_err(|e| StorageError::Io(std::io::Error::other(e)))??; @@ -204,6 +213,58 @@ pub(crate) async fn stream_to_tmp( Ok((etag, total, file)) } +fn commit_locked( + tmp: &std::path::Path, + dest: &std::path::Path, + cond: Option, + bucket: String, + key: String, +) -> Result<()> { + use std::fs::OpenOptions; + use std::os::unix::io::AsRawFd; + + // Append ".lock" — do not use with_extension, which replaces ".json" etc. + let mut lock_path = dest.as_os_str().to_os_string(); + lock_path.push(".lock"); + let lock_path = std::path::PathBuf::from(lock_path); + let lock = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(&lock_path) + .map_err(StorageError::Io)?; + let rc = unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX) }; + if rc != 0 { + return Err(StorageError::Io(std::io::Error::last_os_error())); + } + let _hold = lock; + + match cond { + Some(WriteCondition::IfNoneMatch) if dest.exists() => { + return Err(StorageError::ObjectExists { bucket, key }); + } + Some(WriteCondition::IfMatch(expected)) => { + let any = expected == "*" || expected == "\"*\""; + match xattr::get(dest, xattr::ETAG)? { + None => { + return Err(StorageError::NotFound { bucket, key }); + } + Some(_) if any => {} + Some(actual) if actual.as_slice() != expected.as_bytes() => { + return Err(StorageError::EtagMismatch); + } + _ => {} + } + } + _ => {} + } + + std::fs::rename(tmp, dest).map_err(StorageError::Io)?; + let _ = std::fs::remove_file(&lock_path); + Ok(()) +} + /// Atomic create-or-fail rename using `renameat2(RENAME_NOREPLACE)`. /// Returns `Err` with `EEXIST` if the destination already exists. /// Only available on Linux; callers guard with `#[cfg(target_os = "linux")]`. diff --git a/crates/storage/tests/conditional_writes.rs b/crates/storage/tests/conditional_writes.rs new file mode 100644 index 0000000..7f83882 --- /dev/null +++ b/crates/storage/tests/conditional_writes.rs @@ -0,0 +1,121 @@ +//! Conditional-write contracts. Fail on current main (B2/B12/B18). + +use std::io::Cursor; +use std::sync::Arc; + +use beyond_objects_storage::{AccessLevel, ObjectMeta, Storage, StorageError, WriteCondition}; +use tempfile::TempDir; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn if_none_match_exactly_one_winner() { + for _ in 0..24 { + let dir = TempDir::new().unwrap(); + let store = Arc::new(Storage::new(dir.path())); + store + .create_bucket("bkt", AccessLevel::Private) + .await + .unwrap(); + let mut joins = Vec::new(); + for i in 0..32 { + let s = Arc::clone(&store); + joins.push(tokio::spawn(async move { + s.write_object( + "bkt", + "only-once", + Cursor::new(format!("w{i}").into_bytes()), + ObjectMeta::default(), + Some(WriteCondition::IfNoneMatch), + ) + .await + })); + } + let mut oks = 0u32; + for j in joins { + match j.await.unwrap() { + Ok(_) => oks += 1, + Err(StorageError::ObjectExists { .. }) => {} + Err(e) => panic!("unexpected {e}"), + } + } + assert!(oks >= 1, "at least one create must succeed"); + if oks != 1 { + panic!("If-None-Match is create-only: got {oks} successes"); + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn if_match_is_compare_and_swap() { + let dir = TempDir::new().unwrap(); + let store = Arc::new(Storage::new(dir.path())); + store + .create_bucket("bkt", AccessLevel::Private) + .await + .unwrap(); + let (etag, _) = store + .write_object( + "bkt", + "k", + Cursor::new(b"v0".to_vec()), + ObjectMeta::default(), + None, + ) + .await + .unwrap(); + + let mut joins = Vec::new(); + for i in 0..12 { + let s = Arc::clone(&store); + let et = etag.clone(); + joins.push(tokio::spawn(async move { + s.write_object( + "bkt", + "k", + Cursor::new(format!("v{i}").into_bytes()), + ObjectMeta::default(), + Some(WriteCondition::IfMatch(et)), + ) + .await + })); + } + let mut oks = 0u32; + for j in joins { + if j.await.unwrap().is_ok() { + oks += 1; + } + } + assert_eq!( + oks, 1, + "If-Match is CAS: exactly one concurrent update may succeed, got {oks}" + ); +} + +#[tokio::test] +async fn if_match_star_updates_existing() { + let dir = TempDir::new().unwrap(); + let store = Storage::new(dir.path()); + store + .create_bucket("bkt", AccessLevel::Private) + .await + .unwrap(); + store + .write_object( + "bkt", + "k", + Cursor::new(b"v0".to_vec()), + ObjectMeta::default(), + None, + ) + .await + .unwrap(); + store + .write_object( + "bkt", + "k", + Cursor::new(b"v1".to_vec()), + ObjectMeta::default(), + Some(WriteCondition::IfMatch("\"*\"".into())), + ) + .await + .expect("S3 If-Match: * must succeed when the object exists"); +}