fix(object): make version undo preconditions atomic (#5225)

* fix(object): make version undo preconditions atomic

* fix(object): fence metadata-only undo copies

* fix(object): order versions by commit time

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -21,6 +21,7 @@ use super::storage_api::test::contract::{
     bucket::{BucketOperations, BucketOptions, MakeBucketOptions},
     list::ListOperations as _,
     multipart::MultipartOperations as _,
+    namespace::NamespaceLocking as _,
     object::{ObjectIO as _, ObjectOperations as _},
 };
 use super::storage_api::test::ecfs::FS;
@@ -40,6 +41,7 @@ use http::{Extensions, HeaderMap, HeaderValue, Method, Uri, header::IF_NONE_MATC
 use rustfs_config::{ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT};
 use rustfs_object_capacity::capacity_manager::{HybridStrategyConfig, create_isolated_manager};
 use rustfs_utils::http::{SUFFIX_FORCE_DELETE, insert_header};
+use rustfs_utils::path::encode_dir_object;
 use s3s::{S3Request, dto::*};
 use serial_test::serial;
 use std::{
@@ -63,6 +65,7 @@ const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOU
 const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED";
 const ENV_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED";
 const ENV_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE";
+const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id";
 
 fn init_tracing() {
     INIT.call_once(|| {});
@@ -546,6 +549,227 @@ async fn copy_object_if_none_match_existing_destination_returns_precondition_fai
     );
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial]
+#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"]
+async fn undo_copy_requires_the_observed_current_version() {
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let fs = FS::new();
+    let usecase = DefaultObjectUsecase::from_global();
+    let bucket = format!("test-undo-copy-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let object = "test/object.txt";
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+
+    let put = |payload: &'static [u8]| {
+        PutObjectInput::builder()
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .body(Some(streaming_blob_from_bytes(payload)))
+            .content_length(Some(payload.len() as i64))
+            .build()
+            .unwrap()
+    };
+    let original = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"original"), Method::PUT)))
+        .await
+        .expect("initial version should be written")
+        .output
+        .version_id
+        .expect("versioned PUT should return a version ID");
+    let observed_current = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"replacement"), Method::PUT)))
+        .await
+        .expect("replacement version should be written")
+        .output
+        .version_id
+        .expect("versioned PUT should return a version ID");
+
+    let copy = || {
+        CopyObjectInput::builder()
+            .copy_source(CopySource::Bucket {
+                bucket: bucket.clone().into(),
+                key: object.to_string().into(),
+                version_id: Some(original.clone().into()),
+            })
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .build()
+            .unwrap()
+    };
+    let mut stale_req = build_request(copy(), Method::PUT);
+    stale_req.headers.insert(
+        RUSTFS_EXPECTED_CURRENT_VERSION_ID,
+        HeaderValue::from_str(&Uuid::new_v4().to_string()).unwrap(),
+    );
+    let versions_before = live_object_version_count(&ecstore, bucket.as_str(), object).await;
+    let err = Box::pin(usecase.execute_copy_object(stale_req))
+        .await
+        .expect_err("stale destination must fail");
+    assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed);
+    assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), object).await, versions_before);
+    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"replacement");
+
+    let mut matching_req = build_request(copy(), Method::PUT);
+    matching_req
+        .headers
+        .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&observed_current).unwrap());
+    Box::pin(usecase.execute_copy_object(matching_req))
+        .await
+        .expect("matching destination version should permit undo copy");
+    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"original");
+    assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), object).await, versions_before + 1);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial]
+#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"]
+async fn undo_delete_only_removes_the_current_matching_delete_marker() {
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let fs = FS::new();
+    let usecase = DefaultObjectUsecase::from_global();
+    let bucket = format!("test-undo-delete-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let object = "test/object.txt";
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+
+    let put = PutObjectInput::builder()
+        .bucket(bucket.clone())
+        .key(object.to_string())
+        .body(Some(streaming_blob_from_bytes(b"visible again")))
+        .content_length(Some(13))
+        .build()
+        .unwrap();
+    Box::pin(usecase.execute_put_object(&fs, build_request(put, Method::PUT)))
+        .await
+        .expect("object version should be written");
+    let marker = Box::pin(
+        usecase.execute_delete_object(build_request(
+            DeleteObjectInput::builder()
+                .bucket(bucket.clone())
+                .key(object.to_string())
+                .build()
+                .unwrap(),
+            Method::DELETE,
+        )),
+    )
+    .await
+    .expect("delete marker should be created")
+    .output
+    .version_id
+    .expect("versioned delete should return the marker version");
+
+    let delete_marker = || {
+        DeleteObjectInput::builder()
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .version_id(Some(marker.clone()))
+            .build()
+            .unwrap()
+    };
+    let mut stale_req = build_request(delete_marker(), Method::DELETE);
+    stale_req.headers.insert(
+        RUSTFS_EXPECTED_CURRENT_VERSION_ID,
+        HeaderValue::from_str(&Uuid::new_v4().to_string()).unwrap(),
+    );
+    let err = Box::pin(usecase.execute_delete_object(stale_req))
+        .await
+        .expect_err("mismatched delete marker must fail");
+    assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed);
+
+    let mut matching_req = build_request(delete_marker(), Method::DELETE);
+    matching_req
+        .headers
+        .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&marker).unwrap());
+    Box::pin(usecase.execute_delete_object(matching_req))
+        .await
+        .expect("current matching delete marker should be removed");
+    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"visible again");
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[serial]
+#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"]
+async fn undo_copy_serializes_a_concurrent_put_until_after_commit() {
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let bucket = format!("test-undo-race-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let object = "test/object.txt";
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+
+    let put = |payload: &'static [u8]| {
+        PutObjectInput::builder()
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .body(Some(streaming_blob_from_bytes(payload)))
+            .content_length(Some(payload.len() as i64))
+            .build()
+            .unwrap()
+    };
+    let fs = FS::new();
+    let usecase = DefaultObjectUsecase::from_global();
+    let original = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"original"), Method::PUT)))
+        .await
+        .expect("original should be written")
+        .output
+        .version_id
+        .expect("original should be versioned");
+    let observed = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"observed"), Method::PUT)))
+        .await
+        .expect("observed should be written")
+        .output
+        .version_id
+        .expect("observed should be versioned");
+    let encoded_object = encode_dir_object(object);
+    let lock = ecstore
+        .new_ns_lock(bucket.as_str(), &encoded_object)
+        .await
+        .expect("namespace lock should be available");
+    let blocker = lock
+        .get_write_lock(Duration::from_secs(10))
+        .await
+        .expect("test should acquire blocker");
+
+    let mut undo_req = build_request(
+        CopyObjectInput::builder()
+            .copy_source(CopySource::Bucket {
+                bucket: bucket.clone().into(),
+                key: object.to_string().into(),
+                version_id: Some(original.into()),
+            })
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .build()
+            .unwrap(),
+        Method::PUT,
+    );
+    undo_req
+        .headers
+        .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&observed).unwrap());
+
+    let undo = tokio::spawn(async move {
+        let usecase = DefaultObjectUsecase::from_global();
+        Box::pin(usecase.execute_copy_object(undo_req)).await
+    });
+    tokio::task::yield_now().await;
+    let put = PutObjectInput::builder()
+        .bucket(bucket.clone())
+        .key(object.to_string())
+        .body(Some(streaming_blob_from_bytes(b"concurrent")))
+        .content_length(Some(10))
+        .build()
+        .unwrap();
+    let concurrent_put = tokio::spawn(async move {
+        let fs = FS::new();
+        let usecase = DefaultObjectUsecase::from_global();
+        Box::pin(usecase.execute_put_object(&fs, build_request(put, Method::PUT))).await
+    });
+
+    drop(blocker);
+    let undo_result = undo.await.expect("undo task should not panic");
+    let put_result = concurrent_put.await.expect("PUT task should not panic");
+    assert!(put_result.is_ok(), "ordinary PUT should eventually succeed: {put_result:?}");
+    if let Err(err) = undo_result {
+        assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed);
+    }
+    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"concurrent");
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 #[serial]
 #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]

rustfs/src/app/object_usecase.rs

@@ -167,6 +167,7 @@ use s3s::stream::{ByteStream, DynByteStream, RemainingLength};
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 
 const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024;
+const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id";
 const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES";
 const DEFAULT_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: usize = 16 * 1024 * 1024;
 const PUT_EAGER_STATUS_ELIGIBLE: &str = "eligible";
@@ -2571,6 +2572,31 @@ fn internal_object_info_lookup_opts(mut opts: ObjectOptions) -> ObjectOptions {
     opts
 }
 
+fn expected_current_version_id(headers: &HeaderMap) -> S3Result<Option<String>> {
+    headers
+        .get(RUSTFS_EXPECTED_CURRENT_VERSION_ID)
+        .map(|value| {
+            let value = value
+                .to_str()
+                .map(str::trim)
+                .map_err(|_| s3_error!(InvalidArgument, "Invalid expected current version ID header"))?;
+            if value.eq_ignore_ascii_case("null") {
+                return Ok(Uuid::nil().to_string());
+            }
+            Uuid::parse_str(value)
+                .map(|version| version.to_string())
+                .map_err(|_| s3_error!(InvalidArgument, "Invalid expected current version ID header"))
+        })
+        .transpose()
+}
+
+fn validate_undo_delete_version(expected: Option<&str>, requested: Option<&str>) -> S3Result<()> {
+    if expected.is_some() && expected != requested {
+        return Err(s3_error!(PreconditionFailed));
+    }
+    Ok(())
+}
+
 fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError {
     match err {
         rustfs_lock::LockError::QuorumNotReached { required, achieved } => StorageError::NamespaceLockQuorumUnavailable {
@@ -6023,12 +6049,21 @@ impl DefaultObjectUsecase {
             .map_err(ApiError::from)?;
 
         let cp_src_dst_same = path_join_buf(&[&src_bucket, &src_key]) == path_join_buf(&[&bucket, &key]);
+        let expected_current_version_id = expected_current_version_id(&req.headers)?;
+        if expected_current_version_id.is_some()
+            && (!cp_src_dst_same || version_id.is_none() || dest_version_id.is_some() || !dst_opts.versioned)
+        {
+            return Err(s3_error!(
+                InvalidRequest,
+                "Expected current version precondition requires a versioned same-object historical copy that creates a new version"
+            ));
+        }
 
         let Some(store) = self.object_store() else {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
 
-        let _self_copy_lock_guard = if cp_src_dst_same {
+        let _self_copy_lock_guard = if cp_src_dst_same && expected_current_version_id.is_none() {
             let guard = acquire_self_copy_namespace_lock(store.as_ref(), &bucket, &key).await?;
             src_opts.no_lock = true;
             src_get_opts.no_lock = true;
@@ -6037,21 +6072,30 @@ impl DefaultObjectUsecase {
         } else {
             None
         };
+        dst_opts.expected_current_version_id = expected_current_version_id.clone();
 
         let mut current_opts: ObjectOptions = internal_object_info_lookup_opts(
             get_opts(&bucket, &key, dest_version_id.clone(), None, &req.headers)
                 .await
                 .map_err(ApiError::from)?,
         );
-        if cp_src_dst_same {
+        if _self_copy_lock_guard.is_some() {
             current_opts.no_lock = true;
         }
         let previous_current_size = match store.get_object_info(&bucket, &key, &current_opts).await {
             Ok(existing_obj_info) => {
                 validate_existing_object_lock_for_write(&existing_obj_info, &dst_opts)?;
+                if let Some(expected) = expected_current_version_id.as_deref()
+                    && existing_obj_info.version_id.unwrap_or_default().to_string() != expected
+                {
+                    return Err(s3_error!(PreconditionFailed));
+                }
                 Some(existing_obj_info.size.max(0) as u64)
             }
             Err(err) => {
+                if expected_current_version_id.is_some() {
+                    return Err(s3_error!(PreconditionFailed));
+                }
                 if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
                     return Err(ApiError::from(err).into());
                 }
@@ -6898,9 +6942,18 @@ impl DefaultObjectUsecase {
             // }
         }
 
+        let expected_current_version_id = expected_current_version_id(&req.headers)?;
+        if expected_current_version_id.is_some() && (force_delete || !opts.versioned) {
+            return Err(s3_error!(
+                InvalidRequest,
+                "Expected current version precondition requires a version-specific delete in a versioned bucket"
+            ));
+        }
+        validate_undo_delete_version(expected_current_version_id.as_deref(), opts.version_id.as_deref())?;
         let Some(store) = self.object_store() else {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
+        opts.expected_current_version_id = expected_current_version_id.clone();
 
         let replicate_force_delete = force_delete
             && !replica
@@ -6920,7 +6973,6 @@ impl DefaultObjectUsecase {
         let get_opts: ObjectOptions = get_opts(&bucket, &key, version_id_clone, None, &req.headers)
             .await
             .map_err(ApiError::from)?;
-
         let existing_object_info = match store.get_object_info(&bucket, &key, &get_opts).await {
             Ok(obj_info) => {
                 // Check for bypass governance retention header (permission already verified in access.rs)
@@ -12930,6 +12982,70 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
     }
 
+    #[test]
+    fn expected_current_version_header_normalizes_uuid_and_null() {
+        let version = Uuid::new_v4();
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            RUSTFS_EXPECTED_CURRENT_VERSION_ID,
+            HeaderValue::from_str(&version.to_string().to_uppercase()).unwrap(),
+        );
+        assert_eq!(expected_current_version_id(&headers).unwrap(), Some(version.to_string()));
+
+        headers.insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_static(" null "));
+        assert_eq!(expected_current_version_id(&headers).unwrap(), Some(Uuid::nil().to_string()));
+    }
+
+    #[test]
+    fn expected_current_version_header_rejects_empty_and_malformed_values() {
+        for value in ["", "not-a-version"] {
+            let mut headers = HeaderMap::new();
+            headers.insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(value).unwrap());
+            assert_eq!(expected_current_version_id(&headers).unwrap_err().code(), &S3ErrorCode::InvalidArgument);
+        }
+    }
+
+    #[tokio::test]
+    async fn execute_copy_object_rejects_expected_version_for_different_destination() {
+        let input = CopyObjectInput::builder()
+            .copy_source(CopySource::Bucket {
+                bucket: "test-bucket".into(),
+                key: "source-key".into(),
+                version_id: Some(Uuid::new_v4().to_string().into()),
+            })
+            .bucket("test-bucket".to_string())
+            .key("destination-key".to_string())
+            .build()
+            .unwrap();
+        let mut req = build_request(input, Method::PUT);
+        req.headers.insert(
+            RUSTFS_EXPECTED_CURRENT_VERSION_ID,
+            HeaderValue::from_str(&Uuid::new_v4().to_string()).unwrap(),
+        );
+
+        let err = Box::pin(DefaultObjectUsecase::without_context().execute_copy_object(req))
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+    }
+
+    #[test]
+    fn undo_delete_requires_version_id_to_match_expected_current() {
+        let expected = Uuid::new_v4().to_string();
+        assert!(validate_undo_delete_version(Some(&expected), Some(&expected)).is_ok());
+        assert_eq!(
+            validate_undo_delete_version(Some(&expected), Some(&Uuid::new_v4().to_string()))
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::PreconditionFailed
+        );
+        assert_eq!(
+            validate_undo_delete_version(Some(&expected), None).unwrap_err().code(),
+            &S3ErrorCode::PreconditionFailed
+        );
+        assert!(validate_undo_delete_version(None, None).is_ok());
+    }
+
     #[tokio::test]
     async fn execute_delete_objects_rejects_empty_object_list() {
         let input = DeleteObjectsInput::builder()

rustfs/src/app/storage_api.rs

@@ -1065,6 +1065,10 @@ pub(crate) mod test {
             pub(crate) use super::super::super::storage_contracts::MultipartOperations;
         }
 
+        pub(crate) mod namespace {
+            pub(crate) use super::super::super::storage_contracts::NamespaceLocking;
+        }
+
         pub(crate) mod object {
             pub(crate) use super::super::super::storage_contracts::{ObjectIO, ObjectOperations};
         }