fix: guard operator candidate delete by exact version

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/ecstore/src/bucket/lifecycle/transition_transaction.rs

@@ -667,6 +667,11 @@ pub enum TransitionOperatorError {
     RemoteVersionRequired,
     #[error("remote candidate is not proven missing: {0:?}")]
     CandidateNotMissing(TransitionOperatorProbe),
+    #[error("remote candidate version does not match requested exact version: expected {expected}, observed {actual:?}")]
+    CandidateVersionMismatch {
+        expected: String,
+        actual: TransitionOperatorProbe,
+    },
     #[error("transition transaction store failed: {0}")]
     Store(#[source] Error),
     #[error("remote tier reconciliation failed: {0}")]
@@ -757,6 +762,17 @@ pub async fn delete_transition_candidate_for_operator(
     lease
         .validate_remote_version_id(remote_version_id)
         .map_err(TransitionOperatorError::Remote)?;
+    let before_delete_probe = lease
+        .probe_transition_candidate_for(&transaction.remote_object, transaction.transaction_id)
+        .await
+        .map(TransitionOperatorProbe::from)
+        .map_err(TransitionOperatorError::Remote)?;
+    if !matches!(&before_delete_probe, TransitionOperatorProbe::VersionedPresent(version_id) if version_id == remote_version_id) {
+        return Err(TransitionOperatorError::CandidateVersionMismatch {
+            expected: remote_version_id.to_string(),
+            actual: before_delete_probe,
+        });
+    }
     delete_confirmed_transition_candidate_exact_with_lease_idempotent(&transaction.remote_object, remote_version_id, &lease)
         .await
         .map_err(TransitionOperatorError::Remote)?;

crates/ecstore/src/services/tier/warm_backend_s3.rs

@@ -459,7 +459,7 @@ mod tests {
         let metadata = candidate_metadata(identity);
         assert!(transition_candidate_metadata_matches(&metadata, identity).unwrap());
 
-        let mut adjacent = metadata.clone();
+        let mut adjacent = metadata;
         rustfs_utils::http::metadata_compat::insert_str(
             &mut adjacent,
             rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID,

crates/ecstore/src/store/init.rs

@@ -2903,15 +2903,18 @@ mod tests {
         assert_eq!(status.probe, TransitionOperatorProbe::VersionedPresent(remote_version.clone()));
 
         let wrong_version = uuid::Uuid::new_v4().to_string();
-        let result = delete_transition_candidate_for_operator(store.clone(), transaction.transaction_id, &wrong_version)
+        let err = delete_transition_candidate_for_operator(store.clone(), transaction.transaction_id, &wrong_version)
             .await
-            .expect("an exact delete of an absent version should be idempotent");
-        assert_eq!(
-            result.status.probe,
-            TransitionOperatorProbe::VersionedPresent(remote_version.clone()),
-            "an incorrect exact version must not delete the provider-confirmed candidate"
-        );
-        assert!(result.journal_observed_after_delete);
+            .expect_err("a mismatched exact version must fail before deleting a candidate");
+        assert!(matches!(
+            err,
+            TransitionOperatorError::CandidateVersionMismatch {
+                expected,
+                actual: TransitionOperatorProbe::VersionedPresent(ref observed),
+            } if expected == wrong_version && observed == &remote_version
+        ));
+        assert!(backend.contains(&transaction.remote_object).await);
+        assert_eq!(backend.exact_remove_count(), 0);
         load_transition_transaction_record(store.clone(), transaction.transaction_id)
             .await
             .expect("an incorrect exact version must retain the transaction journal");
@@ -2921,7 +2924,7 @@ mod tests {
             .expect("operator-confirmed exact candidate should be deleted");
         assert_eq!(result.status.probe, TransitionOperatorProbe::Missing);
         assert!(result.journal_observed_after_delete);
-        assert_eq!(backend.exact_remove_count(), 2);
+        assert_eq!(backend.exact_remove_count(), 1);
         assert_eq!(backend.remove_versions().await, vec![(transaction.remote_object.clone(), remote_version)]);
         load_transition_transaction_record(store.clone(), transaction.transaction_id)
             .await

rustfs/src/admin/handlers/ilm_transition.rs

@@ -444,6 +444,9 @@ fn map_transition_operator_error(err: TransitionOperatorError) -> S3Error {
         TransitionOperatorError::CandidateNotMissing(_) => {
             s3_error!(OperationAborted, "remote candidate is not proven missing")
         }
+        TransitionOperatorError::CandidateVersionMismatch { .. } => {
+            s3_error!(OperationAborted, "remote candidate version does not match requested exact version")
+        }
         TransitionOperatorError::Store(_) | TransitionOperatorError::Remote(_) => {
             s3_error!(InternalError, "transition reconciliation failed")
         }