Conditional writes are not atomic; If-Match * is broken

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,

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<WriteCondition>,
+    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")]`.

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");
+}