fix(replication): fence force-delete journal updates (#5661)

* feat(replication): add conditional config store APIs

* fix(replication): fence force-delete journal updates

* fix(replication): retry durable force-delete commits

* test(replication): fence lost force-delete journal leases

* fix(replication): bound force-delete journal retries

crates/ecstore/src/bucket/replication/replication_config_store.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use super::replication_error_boundary::Result;
-use super::replication_storage_boundary::ReplicationObjectIO;
+use super::replication_storage_boundary::{HTTPPreconditions, ObjectInfo, ObjectOptions, ReplicationObjectIO};
 use crate::config::{com, storageclass};
 use std::sync::Arc;
 
@@ -37,6 +37,21 @@ impl ReplicationConfigStore {
         com::read_config_no_lock(api, file).await
     }
 
+    pub(crate) async fn read_no_lock_with_metadata<S>(api: Arc<S>, file: &str) -> Result<(Vec<u8>, ObjectInfo)>
+    where
+        S: ReplicationObjectIO,
+    {
+        com::read_config_with_metadata(
+            api,
+            file,
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+        )
+        .await
+    }
+
     pub(crate) async fn save<S>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()>
     where
         S: ReplicationObjectIO,
@@ -50,4 +65,26 @@ impl ReplicationConfigStore {
     {
         com::save_config_no_lock(api, file, data).await
     }
+
+    pub(crate) async fn save_conditional<S>(
+        api: Arc<S>,
+        file: &str,
+        data: Vec<u8>,
+        http_preconditions: HTTPPreconditions,
+    ) -> Result<()>
+    where
+        S: ReplicationObjectIO,
+    {
+        com::save_config_with_opts_quiet(
+            api,
+            file,
+            data,
+            &ObjectOptions {
+                max_parity: true,
+                http_preconditions: Some(http_preconditions),
+                ..Default::default()
+            },
+        )
+        .await
+    }
 }

crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs

@@ -33,6 +33,7 @@ pub(crate) struct ReplicationMetadataStore;
 impl ReplicationMetadataStore {
     pub(crate) const MRF_REPLICATION_FILE: &'static str = "config/replication/mrf.bin";
     pub(crate) const FORCE_DELETE_REPLICATION_FILE: &'static str = "config/replication/force-delete.bin";
+    pub(crate) const FORCE_DELETE_REPLICATION_TRANSACTION_LOCK: &'static str = "config/replication/force-delete.bin.transaction";
 
     pub(crate) async fn replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
         metadata_sys::get_replication_config(bucket).await
@@ -114,5 +115,9 @@ mod tests {
             ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE,
             "config/replication/force-delete.bin"
         );
+        assert_eq!(
+            ReplicationMetadataStore::FORCE_DELETE_REPLICATION_TRANSACTION_LOCK,
+            "config/replication/force-delete.bin.transaction"
+        );
     }
 }

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -42,7 +42,8 @@ use super::replication_resyncer::{
 };
 use super::replication_state::ReplicationStats;
 use super::replication_storage_boundary::{
-    ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage,
+    HTTPPreconditions, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationDeletedObject, ReplicationObjectIO,
+    ReplicationStorage,
 };
 use super::replication_target_boundary::{ReplicationTargetStore, replication_object_is_ssec_encrypted};
 use super::replication_versioning_boundary::ReplicationVersioningStore;
@@ -403,75 +404,143 @@ pub async fn persist_force_delete_intent<S: ReplicationStorage>(
     mut entry: MrfReplicateEntry,
 ) -> Result<(), EcstoreError> {
     entry.force_delete_local_commit = false;
-    let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE;
-    let lock = storage
-        .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file)
-        .await?;
-    let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?;
-
-    let mut entries = match ReplicationConfigStore::read_no_lock(storage.clone(), file).await {
-        Ok(data) => decode_mrf_file(&data)?,
-        Err(EcstoreError::ConfigNotFound) => Vec::new(),
-        Err(err) => return Err(err),
-    };
-
-    if entries
-        .iter()
-        .any(|existing| existing.force_delete_id == entry.force_delete_id)
-    {
-        return Ok(());
-    }
-
-    entries.push(entry);
-    let data = encode_mrf_file(&entries)?;
-    ReplicationConfigStore::save_no_lock(storage, file, data).await
+    update_force_delete_intents(storage, move |entries, _exists| {
+        if entries
+            .iter()
+            .any(|existing| existing.force_delete_id == entry.force_delete_id)
+        {
+            return Ok(false);
+        }
+        entries.push(entry.clone());
+        Ok(true)
+    })
+    .await
 }
 
 pub async fn commit_force_delete_intent<S: ReplicationStorage>(
     storage: Arc<S>,
     operation_id: uuid::Uuid,
 ) -> Result<(), EcstoreError> {
-    let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE;
-    let lock = storage
-        .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file)
-        .await?;
-    let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?;
-
-    let data = ReplicationConfigStore::read_no_lock(storage.clone(), file).await?;
-    let mut entries = decode_mrf_file(&data)?;
-    let Some(entry) = entries.iter_mut().find(|entry| entry.force_delete_id == Some(operation_id)) else {
-        return Err(EcstoreError::ConfigNotFound);
-    };
-    if entry.force_delete_local_commit {
-        return Ok(());
-    }
-    entry.force_delete_local_commit = true;
-    ReplicationConfigStore::save_no_lock(storage, file, encode_mrf_file(&entries)?).await
+    update_force_delete_intents(storage, move |entries, exists| {
+        if !exists {
+            return Err(EcstoreError::ConfigNotFound);
+        }
+        let Some(entry) = entries.iter_mut().find(|entry| entry.force_delete_id == Some(operation_id)) else {
+            return Err(EcstoreError::ConfigNotFound);
+        };
+        if entry.force_delete_local_commit {
+            return Ok(false);
+        }
+        entry.force_delete_local_commit = true;
+        Ok(true)
+    })
+    .await
 }
 
 pub async fn complete_force_delete_intent<S: ReplicationStorage>(
     storage: Arc<S>,
     operation_id: uuid::Uuid,
 ) -> Result<(), EcstoreError> {
-    let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE;
-    let lock = storage
-        .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file)
-        .await?;
-    let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?;
+    update_force_delete_intents(storage, move |entries, exists| {
+        if !exists {
+            return Ok(false);
+        }
+        let original_len = entries.len();
+        entries.retain(|entry| entry.force_delete_id != Some(operation_id));
+        Ok(entries.len() != original_len)
+    })
+    .await
+}
 
-    let data = match ReplicationConfigStore::read_no_lock(storage.clone(), file).await {
-        Ok(data) => data,
-        Err(EcstoreError::ConfigNotFound) => return Ok(()),
-        Err(err) => return Err(err),
-    };
-    let mut entries = decode_mrf_file(&data)?;
-    let original_len = entries.len();
-    entries.retain(|entry| entry.force_delete_id != Some(operation_id));
-    if entries.len() == original_len {
-        return Ok(());
+const FORCE_DELETE_INTENT_CAS_RETRIES: usize = 3;
+
+fn is_retryable_force_delete_error(error: &EcstoreError) -> bool {
+    matches!(error, EcstoreError::PreconditionFailed) || error.to_string().contains("force-delete journal lock lost")
+}
+
+async fn update_force_delete_intents<S, F>(storage: Arc<S>, mut update: F) -> Result<(), EcstoreError>
+where
+    S: ReplicationStorage,
+    F: FnMut(&mut Vec<MrfReplicateEntry>, bool) -> Result<bool, EcstoreError>,
+{
+    let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE;
+    for attempt in 0..=FORCE_DELETE_INTENT_CAS_RETRIES {
+        let result = {
+            let lock = storage
+                .new_ns_lock(
+                    ReplicationMetadataStore::rustfs_meta_bucket(),
+                    ReplicationMetadataStore::FORCE_DELETE_REPLICATION_TRANSACTION_LOCK,
+                )
+                .await?;
+            // Lock order is transaction namespace lock -> force-delete journal object lock.
+            // Keep the transaction guard alive through the conditional write so legacy
+            // writers cannot interleave a read-modify-write transition within this process.
+            let guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?;
+            let (mut entries, preconditions, exists) = read_force_delete_intents(storage.clone(), file).await?;
+            if !update(&mut entries, exists)? {
+                return Ok(());
+            }
+            save_force_delete_intents(storage.clone(), file, &guard, entries, preconditions).await
+        };
+
+        match result {
+            Err(error) if is_retryable_force_delete_error(&error) && attempt < FORCE_DELETE_INTENT_CAS_RETRIES => {
+                tokio::time::sleep(Duration::from_millis(25)).await;
+            }
+            result => return result,
+        }
     }
 
-    ReplicationConfigStore::save_no_lock(storage, file, encode_mrf_file(&entries)?).await
+    Err(EcstoreError::other("force-delete journal update retries exhausted"))
+}
+
+async fn read_force_delete_intents<S: ReplicationObjectIO>(
+    storage: Arc<S>,
+    file: &str,
+) -> Result<(Vec<MrfReplicateEntry>, HTTPPreconditions, bool), EcstoreError> {
+    match ReplicationConfigStore::read_no_lock_with_metadata(storage, file).await {
+        Ok((data, object_info)) => {
+            let etag = object_info
+                .etag
+                .filter(|etag| !etag.trim().is_empty())
+                .ok_or_else(|| EcstoreError::other("force-delete journal has no ETag for conditional update"))?;
+            Ok((
+                decode_mrf_file(&data)?,
+                HTTPPreconditions {
+                    if_match: Some(etag),
+                    ..Default::default()
+                },
+                true,
+            ))
+        }
+        Err(EcstoreError::ConfigNotFound) => Ok((
+            Vec::new(),
+            HTTPPreconditions {
+                if_none_match: Some("*".to_string()),
+                ..Default::default()
+            },
+            false,
+        )),
+        Err(err) => Err(err),
+    }
+}
+
+async fn save_force_delete_intents<S: ReplicationStorage>(
+    storage: Arc<S>,
+    file: &str,
+    guard: &rustfs_lock::NamespaceLockGuard,
+    entries: Vec<MrfReplicateEntry>,
+    preconditions: HTTPPreconditions,
+) -> Result<(), EcstoreError> {
+    ensure_force_delete_journal_lock_held(guard.is_lock_lost())?;
+    ReplicationConfigStore::save_conditional(storage, file, encode_mrf_file(&entries)?, preconditions).await
+}
+
+fn ensure_force_delete_journal_lock_held(lock_lost: bool) -> Result<(), EcstoreError> {
+    if lock_lost {
+        return Err(EcstoreError::other("force-delete journal lock lost before conditional update"));
+    }
+    Ok(())
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -2522,7 +2591,7 @@ mod tests {
         StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageNamespaceLocking, StorageObjectInfoOrErr, WalkOptions,
     };
     use super::*;
-    use std::collections::HashMap;
+    use std::collections::{HashMap, VecDeque};
     use std::fmt::{Debug, Formatter};
     use std::io::Cursor;
     use std::sync::Mutex as StdMutex;
@@ -2537,6 +2606,11 @@ mod tests {
 
     struct LoadResyncSharedState {
         data: StdMutex<Vec<u8>>,
+        etag_revision: AtomicUsize,
+        last_put_preconditions: StdMutex<Option<HTTPPreconditions>>,
+        last_put_no_lock: AtomicBool,
+        omit_etag: AtomicBool,
+        conditional_write_replacements: StdMutex<VecDeque<Vec<u8>>>,
         writes: StdMutex<Vec<(String, Vec<u8>)>>,
         lock_manager: Arc<rustfs_lock::GlobalLockManager>,
         first_read_started: Notify,
@@ -2615,6 +2689,8 @@ mod tests {
                 object_info: ObjectInfo {
                     size,
                     actual_size: size,
+                    etag: (!self.shared.omit_etag.load(Ordering::SeqCst))
+                        .then(|| format!("mrf-{}", self.shared.etag_revision.load(Ordering::SeqCst))),
                     ..Default::default()
                 },
                 buffered_body: None,
@@ -2627,8 +2703,44 @@ mod tests {
             _bucket: &str,
             object: &str,
             data: &mut Self::PutObjectReader,
-            _opts: &Self::ObjectOptions,
+            opts: &Self::ObjectOptions,
         ) -> Result<Self::ObjectInfo, Self::Error> {
+            if opts.http_preconditions.is_some()
+                && let Some(replacement) = self
+                    .shared
+                    .conditional_write_replacements
+                    .lock()
+                    .expect("test replacement lock should not be poisoned")
+                    .pop_front()
+            {
+                *self.shared.data.lock().expect("test data lock should not be poisoned") = replacement;
+                self.shared.etag_revision.fetch_add(1, Ordering::SeqCst);
+            }
+            let current_etag = if self
+                .shared
+                .data
+                .lock()
+                .expect("test data lock should not be poisoned")
+                .is_empty()
+            {
+                None
+            } else {
+                Some(format!("mrf-{}", self.shared.etag_revision.load(Ordering::SeqCst)))
+            };
+            if opts.http_preconditions.as_ref().is_some_and(|preconditions| {
+                preconditions.if_none_match_value() == Some("*") && current_etag.is_some()
+                    || preconditions
+                        .if_match_value()
+                        .is_some_and(|expected| current_etag.as_deref() != Some(expected))
+            }) {
+                return Err(EcstoreError::PreconditionFailed);
+            }
+            *self
+                .shared
+                .last_put_preconditions
+                .lock()
+                .expect("test preconditions lock should not be poisoned") = opts.http_preconditions.clone();
+            self.shared.last_put_no_lock.store(opts.no_lock, Ordering::SeqCst);
             if self.shared.fail_next_write.swap(false, Ordering::SeqCst) {
                 return Err(EcstoreError::Unexpected);
             }
@@ -2644,6 +2756,7 @@ mod tests {
                 .expect("test writes lock should not be poisoned")
                 .push((object.to_string(), encoded.clone()));
             *self.shared.data.lock().expect("test data lock should not be poisoned") = encoded;
+            self.shared.etag_revision.fetch_add(1, Ordering::SeqCst);
             self.shared.write_count.fetch_add(1, Ordering::SeqCst);
             Ok(ObjectInfo::default())
         }
@@ -3096,6 +3209,11 @@ mod tests {
     fn empty_resync_shared_state() -> Arc<LoadResyncSharedState> {
         Arc::new(LoadResyncSharedState {
             data: StdMutex::new(Vec::new()),
+            etag_revision: AtomicUsize::new(0),
+            last_put_preconditions: StdMutex::new(None),
+            last_put_no_lock: AtomicBool::new(false),
+            omit_etag: AtomicBool::new(false),
+            conditional_write_replacements: StdMutex::new(VecDeque::new()),
             writes: StdMutex::new(Vec::new()),
             lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()),
             first_read_started: Notify::new(),
@@ -3743,6 +3861,11 @@ mod tests {
         temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1"))], async {
             let shared = Arc::new(LoadResyncSharedState {
                 data: StdMutex::new(load_resync_test_metadata()),
+                etag_revision: AtomicUsize::new(1),
+                last_put_preconditions: StdMutex::new(None),
+                last_put_no_lock: AtomicBool::new(false),
+                omit_etag: AtomicBool::new(false),
+                conditional_write_replacements: StdMutex::new(VecDeque::new()),
                 writes: StdMutex::new(Vec::new()),
                 lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()),
                 first_read_started: Notify::new(),
@@ -4393,7 +4516,7 @@ mod tests {
     #[tokio::test]
     async fn force_delete_intent_append_commit_and_cleanup_are_idempotent() {
         let shared = empty_resync_shared_state();
-        let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared));
+        let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared.clone()));
         let operation_id = Uuid::new_v4();
         let entry = MrfReplicateEntry {
             bucket: "source".to_string(),
@@ -4408,6 +4531,15 @@ mod tests {
         persist_force_delete_intent(storage.clone(), entry.clone())
             .await
             .expect("first journal append should succeed");
+        let preconditions = shared
+            .last_put_preconditions
+            .lock()
+            .expect("test preconditions lock should not be poisoned")
+            .clone()
+            .expect("first journal append should be conditional");
+        assert_eq!(preconditions.if_none_match_value(), Some("*"));
+        assert_eq!(preconditions.if_match_value(), None);
+        assert!(!shared.last_put_no_lock.load(Ordering::SeqCst));
         persist_force_delete_intent(storage.clone(), entry)
             .await
             .expect("duplicate journal append should be a no-op");
@@ -4423,6 +4555,15 @@ mod tests {
         commit_force_delete_intent(storage.clone(), operation_id)
             .await
             .expect("commit marker should persist");
+        let preconditions = shared
+            .last_put_preconditions
+            .lock()
+            .expect("test preconditions lock should not be poisoned")
+            .clone()
+            .expect("commit marker should be conditional");
+        assert_eq!(preconditions.if_none_match_value(), None);
+        assert_eq!(preconditions.if_match_value(), Some("mrf-1"));
+        assert!(!shared.last_put_no_lock.load(Ordering::SeqCst));
         commit_force_delete_intent(storage.clone(), operation_id)
             .await
             .expect("duplicate commit marker should be a no-op");
@@ -4440,4 +4581,182 @@ mod tests {
             .await
             .expect("duplicate journal cleanup should be a no-op");
     }
+
+    #[tokio::test]
+    async fn force_delete_intent_cleanup_retries_after_a_stale_journal_snapshot() {
+        let shared = empty_resync_shared_state();
+        let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared.clone()));
+        let operation_id = Uuid::new_v4();
+        let mut entry = MrfReplicateEntry {
+            bucket: "source".to_string(),
+            object: "original".to_string(),
+            force_delete_id: Some(operation_id),
+            op: MrfOpKind::Delete,
+            ..Default::default()
+        };
+        persist_force_delete_intent(storage.clone(), entry.clone())
+            .await
+            .expect("journal append should succeed");
+        commit_force_delete_intent(storage.clone(), operation_id)
+            .await
+            .expect("journal commit should succeed");
+        entry.force_delete_local_commit = true;
+        let concurrent = MrfReplicateEntry {
+            bucket: "source".to_string(),
+            object: "concurrent".to_string(),
+            force_delete_id: Some(Uuid::new_v4()),
+            op: MrfOpKind::Delete,
+            ..Default::default()
+        };
+        shared
+            .conditional_write_replacements
+            .lock()
+            .expect("test replacement lock should not be poisoned")
+            .push_back(encode_mrf_file(&[entry, concurrent.clone()]).expect("concurrent journal entries should encode"));
+
+        complete_force_delete_intent(storage.clone(), operation_id)
+            .await
+            .expect("cleanup should retry after a concurrent journal update");
+
+        let data = ReplicationConfigStore::read(storage, ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE)
+            .await
+            .expect("journal should remain readable");
+        let entries = decode_mrf_file(&data).expect("journal should decode");
+        assert_eq!(entries.len(), 1, "cleanup must preserve only the concurrent journal entry");
+        assert_eq!(entries[0].force_delete_id, concurrent.force_delete_id);
+        assert_eq!(entries[0].object, concurrent.object);
+    }
+
+    #[tokio::test]
+    async fn force_delete_intent_commit_retries_past_the_bounded_cas_conflict_limit() {
+        let shared = empty_resync_shared_state();
+        let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared.clone()));
+        let operation_id = Uuid::new_v4();
+        let entry = MrfReplicateEntry {
+            bucket: "source".to_string(),
+            object: "original".to_string(),
+            force_delete_id: Some(operation_id),
+            op: MrfOpKind::Delete,
+            ..Default::default()
+        };
+        persist_force_delete_intent(storage.clone(), entry.clone())
+            .await
+            .expect("journal append should succeed");
+        {
+            let mut replacements = shared
+                .conditional_write_replacements
+                .lock()
+                .expect("test replacement lock should not be poisoned");
+            for object in ["first", "second", "third"] {
+                let mut replacement = entry.clone();
+                replacement.object = object.to_string();
+                replacements.push_back(encode_mrf_file(&[replacement]).expect("concurrent journal entry should encode"));
+            }
+        }
+
+        commit_force_delete_intent(storage.clone(), operation_id)
+            .await
+            .expect("commit marker must retry until it is durable");
+
+        let data = ReplicationConfigStore::read(storage, ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE)
+            .await
+            .expect("journal should remain readable");
+        let entries = decode_mrf_file(&data).expect("journal should decode");
+        assert_eq!(entries.len(), 1);
+        assert!(entries[0].force_delete_local_commit);
+        assert_eq!(entries[0].force_delete_id, Some(operation_id));
+    }
+
+    #[tokio::test]
+    async fn force_delete_intent_rejects_existing_journal_without_an_etag() {
+        let shared = empty_resync_shared_state();
+        let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared.clone()));
+        let operation_id = Uuid::new_v4();
+        let entry = MrfReplicateEntry {
+            bucket: "source".to_string(),
+            object: "original".to_string(),
+            force_delete_id: Some(operation_id),
+            op: MrfOpKind::Delete,
+            ..Default::default()
+        };
+        persist_force_delete_intent(storage.clone(), entry)
+            .await
+            .expect("journal append should succeed");
+        let writes_before = shared.write_count.load(Ordering::SeqCst);
+        shared.omit_etag.store(true, Ordering::SeqCst);
+
+        let err = commit_force_delete_intent(storage.clone(), operation_id)
+            .await
+            .expect_err("missing ETag must reject journal mutation");
+        assert!(err.to_string().contains("no ETag"));
+        assert_eq!(shared.write_count.load(Ordering::SeqCst), writes_before);
+        let data = ReplicationConfigStore::read(storage, ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE)
+            .await
+            .expect("journal should remain readable");
+        let entries = decode_mrf_file(&data).expect("journal should decode");
+        assert!(!entries[0].force_delete_local_commit);
+    }
+
+    #[test]
+    fn force_delete_journal_rejects_a_lost_transaction_lease() {
+        let err = ensure_force_delete_journal_lock_held(true).expect_err("lost transaction lease must fence the journal write");
+
+        assert!(err.to_string().contains("lock lost"));
+    }
+
+    #[tokio::test]
+    async fn force_delete_journal_rejects_a_stale_conditional_write() {
+        let shared = empty_resync_shared_state();
+        let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared));
+        let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE;
+        let original = MrfReplicateEntry {
+            bucket: "source".to_string(),
+            object: "original".to_string(),
+            force_delete_id: Some(Uuid::new_v4()),
+            op: MrfOpKind::Delete,
+            ..Default::default()
+        };
+        ReplicationConfigStore::save(
+            storage.clone(),
+            file,
+            encode_mrf_file(&[original]).expect("initial journal entry should encode"),
+        )
+        .await
+        .expect("initial journal write should succeed");
+
+        let (_, object_info) = ReplicationConfigStore::read_no_lock_with_metadata(storage.clone(), file)
+            .await
+            .expect("journal snapshot should include an ETag");
+        let stale_preconditions = HTTPPreconditions {
+            if_match: object_info.etag,
+            ..Default::default()
+        };
+        let replacement = MrfReplicateEntry {
+            bucket: "source".to_string(),
+            object: "replacement".to_string(),
+            force_delete_id: Some(Uuid::new_v4()),
+            op: MrfOpKind::Delete,
+            ..Default::default()
+        };
+        let replacement_data = encode_mrf_file(&[replacement]).expect("replacement journal entry should encode");
+        ReplicationConfigStore::save(storage.clone(), file, replacement_data.clone())
+            .await
+            .expect("concurrent journal write should succeed");
+
+        let err = ReplicationConfigStore::save_conditional(
+            storage.clone(),
+            file,
+            encode_mrf_file(&[]).expect("empty journal should encode"),
+            stale_preconditions,
+        )
+        .await
+        .expect_err("stale journal snapshot must not overwrite newer data");
+        assert_eq!(err, EcstoreError::PreconditionFailed);
+        assert_eq!(
+            ReplicationConfigStore::read(storage, file)
+                .await
+                .expect("newer journal data should remain readable"),
+            replacement_data
+        );
+    }
 }

crates/ecstore/src/bucket/replication/replication_storage_boundary.rs

@@ -25,7 +25,7 @@ pub(crate) use crate::storage_api_contracts::list::{
 };
 pub(crate) use crate::storage_api_contracts::namespace::NamespaceLocking as StorageNamespaceLocking;
 pub(crate) use crate::storage_api_contracts::object::{
-    DeletedObject, EcstoreObjectOperations, ObjectIO, ObjectOperations, ObjectToDelete,
+    DeletedObject, EcstoreObjectOperations, HTTPPreconditions, ObjectIO, ObjectOperations, ObjectToDelete,
 };
 pub(crate) use crate::storage_api_contracts::range::HTTPRangeSpec;
 pub(crate) use rustfs_replication::{DeletedObject as ReplicationDeletedObject, ObjectToDelete as ReplicationObjectToDelete};

crates/ecstore/src/config/com.rs

@@ -586,10 +586,47 @@ where
             PutObjectReader = PutObjReader,
         >,
 {
-    save_config_with_opts_and_metadata(api, file, data, opts).await.map(|_| ())
+    save_config_with_opts_inner(api, file, data, opts, true).await.map(|_| ())
+}
+
+/// Saves a configuration object without logging an error for a retryable caller-owned failure.
+pub async fn save_config_with_opts_quiet<S>(api: Arc<S>, file: &str, data: Vec<u8>, opts: &ObjectOptions) -> Result<()>
+where
+    S: ObjectIO<
+            Error = Error,
+            RangeSpec = HTTPRangeSpec,
+            HeaderMap = HeaderMap,
+            ObjectOptions = ObjectOptions,
+            ObjectInfo = ObjectInfo,
+            GetObjectReader = GetObjectReader,
+            PutObjectReader = PutObjReader,
+        >,
+{
+    save_config_with_opts_inner(api, file, data, opts, false).await.map(|_| ())
 }
 
 async fn save_config_with_opts_and_metadata<S>(api: Arc<S>, file: &str, data: Vec<u8>, opts: &ObjectOptions) -> Result<ObjectInfo>
+where
+    S: ObjectIO<
+            Error = Error,
+            RangeSpec = HTTPRangeSpec,
+            HeaderMap = HeaderMap,
+            ObjectOptions = ObjectOptions,
+            ObjectInfo = ObjectInfo,
+            GetObjectReader = GetObjectReader,
+            PutObjectReader = PutObjReader,
+        >,
+{
+    save_config_with_opts_inner(api, file, data, opts, true).await
+}
+
+async fn save_config_with_opts_inner<S>(
+    api: Arc<S>,
+    file: &str,
+    data: Vec<u8>,
+    opts: &ObjectOptions,
+    log_error: bool,
+) -> Result<ObjectInfo>
 where
     S: ObjectIO<
             Error = Error,
@@ -605,7 +642,9 @@ where
     match api.put_object(RUSTFS_META_BUCKET, file, &mut put_data, opts).await {
         Ok(object_info) => Ok(object_info),
         Err(err) => {
-            error!("save_config_with_opts: err: {:?}, file: {}", err, file);
+            if log_error {
+                error!("save_config_with_opts: err: {:?}, file: {}", err, file);
+            }
             Err(err)
         }
     }