fix(replication): tolerate orphaned resync intents at startup (#6470)

* fix(replication): tolerate orphaned resync intents at startup

Since #5215 (1.0.0-beta.12) startup reconciles every pending/started
resync intent in resync.bin against the bucket's configured targets and
aborts the whole server when an intent has no matching target ARN. A
resync whose remote target was later removed leaves exactly such an
orphan on disk, so every later start fails with "accepted replication
resync target ... is not configured" regardless of the binary version.

Skip orphaned intents with a warning instead of failing startup; the
resync routine already settles them to ResyncFailed. Cancel the intent
when its remote target is removed so the orphan is not created again.

Fixes #4784

* fix(replication): cancel removed-target resync under the admission lock

Canceling through this node's cached whole-bucket status map could
persist a map that predates another node's admission, erasing that
node's durable restart intent. Reload resync.bin under the bucket
admission lock, publish the fresh map, and only then mark the removed
target's intent canceled. Two-node regression covers the clobber.

* fix(replication): persist resync status via ETag CAS merge

mark_status, the periodic saver, admission, and removed-target
cancellation all persisted their node's cached whole-bucket map, so any
one node's stale cache could resurrect states another node had already
finalized (a canceled intent flipping back to Pending, an admission
vanishing). All resync.bin writers now go through update_resync_status_cas:
load the freshest document with its ETag, apply a per-target mutation
with staleness and canceled-is-terminal guards re-checked against the
persisted entry, and save conditionally, retrying on concurrent writes.
The periodic saver merges per target, letting terminal states and newer
admissions recorded elsewhere win. Cache convergence stays per-target so
locally running resyncs keep their authoritative progress counters.

Regressions: stale_peer_status_write_cannot_resurrect_canceled_intent
(node B's pre-cancel cache marking its own run Started must not revive
node A's canceled intent) plus unit coverage for the periodic-save merge.

* test(ecstore): rename resync test helper off the guarded contract name

fn resync_target is on the architecture guard's reserved list for
crates/replication operation contracts; the merge-test helper now reads
resync_target_state.

* fix(replication): serialize resync status updates

---------

Co-authored-by: overtrue <anzhengchao@gmail.com>

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

@@ -39,7 +39,7 @@ use super::replication_resync_boundary::{
 };
 use super::replication_resyncer::{
     ReplicationResyncer, get_heal_replicate_object_info, replicate_delete, replicate_delete_with_outcome, replicate_object,
-    replicate_object_with_outcome, save_resync_status,
+    replicate_object_with_outcome, update_resync_status_cas,
 };
 use super::replication_state::ReplicationStats;
 use super::replication_storage_boundary::{
@@ -1898,7 +1898,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             }
         };
 
-        let mut bucket_status = load_bucket_resync_metadata(&opts.bucket, self.storage.clone()).await?;
+        let bucket_status = load_bucket_resync_metadata(&opts.bucket, self.storage.clone()).await?;
         if let Some(active) = bucket_status.targets_map.get(&opts.arn) {
             if active.resync_id == opts.resync_id {
                 self.resyncer
@@ -1924,26 +1924,43 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         }
 
         let now = OffsetDateTime::now_utc();
-        bucket_status.last_update = Some(now);
-        bucket_status.targets_map.insert(
-            opts.arn.clone(),
-            TargetReplicationResyncStatus {
-                start_time: Some(now),
-                last_update: Some(now),
-                resync_id: opts.resync_id.clone(),
-                resync_before_date: opts.resync_before,
-                resync_status: ResyncStatusType::ResyncPending,
-                failed_size: 0,
-                failed_count: 0,
-                replicated_size: 0,
-                replicated_count: 0,
-                bucket: opts.bucket.clone(),
-                object: String::new(),
-                error: None,
-            },
-        );
+        let admitted = TargetReplicationResyncStatus {
+            start_time: Some(now),
+            last_update: Some(now),
+            resync_id: opts.resync_id.clone(),
+            resync_before_date: opts.resync_before,
+            resync_status: ResyncStatusType::ResyncPending,
+            failed_size: 0,
+            failed_count: 0,
+            replicated_size: 0,
+            replicated_count: 0,
+            bucket: opts.bucket.clone(),
+            object: String::new(),
+            error: None,
+        };
 
-        save_resync_status(&opts.bucket, &bucket_status, self.storage.clone()).await?;
+        // The admission lock serializes competing admissions, but status
+        // writers (mark_status, the periodic saver) do not take it — write
+        // through the CAS so their concurrent updates to other targets are
+        // never lost, re-checking the conflict gate on each retry.
+        let (bucket_status, _) = update_resync_status_cas(&opts.bucket, self.storage.clone(), |persisted| {
+            if let Some(active) = persisted.targets_map.get(&opts.arn) {
+                if active.resync_id == opts.resync_id {
+                    return Ok(false);
+                }
+                if should_auto_resume_resync(active.resync_status) {
+                    return Err(EcstoreError::other(ResyncActiveConflictError {
+                        bucket: opts.bucket.clone(),
+                        arn: opts.arn.clone(),
+                        active_resync_id: active.resync_id.clone(),
+                    }));
+                }
+            }
+            persisted.last_update = Some(now);
+            persisted.targets_map.insert(opts.arn.clone(), admitted.clone());
+            Ok(true)
+        })
+        .await?;
         self.resyncer
             .status_map
             .write()
@@ -1953,6 +1970,83 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         Ok(true)
     }
 
+    /// Cancel the pending/started resync intent recorded for `arn`, if any,
+    /// because its remote target is being removed. Returns the canceled run.
+    ///
+    /// Runs under the bucket admission lock and reloads `resync.bin` from disk
+    /// before writing, like admission does: this node's `status_map` entry may
+    /// be stale relative to intents admitted by other nodes, and persisting it
+    /// would silently drop their durable restart intents.
+    pub async fn cancel_bucket_resync_for_removed_target(
+        self: Arc<Self>,
+        bucket: &str,
+        arn: &str,
+    ) -> Result<Option<ResyncOpts>, EcstoreError> {
+        let bucket = bucket.to_string();
+        let arn = arn.to_string();
+        tokio::spawn(async move { self.cancel_bucket_resync_for_removed_target_transaction(bucket, arn).await })
+            .await
+            .map_err(|error| EcstoreError::other(format!("replication resync cancellation task failed: {error}")))?
+    }
+
+    async fn cancel_bucket_resync_for_removed_target_transaction(
+        self: Arc<Self>,
+        bucket: String,
+        arn: String,
+    ) -> Result<Option<ResyncOpts>, EcstoreError> {
+        let admission_lock_key = ReplicationMetadataStore::resync_admission_lock_key(&bucket);
+        let admission_lock = self
+            .storage
+            .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), &admission_lock_key)
+            .await?;
+        // Lock order: bucket resync admission lock -> resync status config-object lock.
+        let _admission_guard = admission_lock
+            .get_write_lock(ReplicationLockTiming::acquire_timeout())
+            .await
+            .map_err(EcstoreError::from)?;
+
+        let mut canceled: Option<ResyncOpts> = None;
+        let (final_map, _) = update_resync_status_cas(&bucket, self.storage.clone(), |persisted| {
+            canceled = None;
+            let Some(intent) = persisted.targets_map.get_mut(&arn) else {
+                return Ok(false);
+            };
+            if !should_auto_resume_resync(intent.resync_status) {
+                return Ok(false);
+            }
+            let now = OffsetDateTime::now_utc();
+            canceled = Some(ResyncOpts {
+                bucket: bucket.clone(),
+                arn: arn.clone(),
+                resync_id: intent.resync_id.clone(),
+                resync_before: intent.resync_before_date,
+            });
+            intent.resync_status = ResyncStatusType::ResyncCanceled;
+            intent.last_update = Some(now);
+            persisted.last_update = Some(now);
+            Ok(true)
+        })
+        .await?;
+
+        // Converge only the removed target's cached entry: cached progress
+        // counters for this node's other running targets stay authoritative.
+        {
+            let mut status_map = self.resyncer.status_map.write().await;
+            let cached = status_map
+                .entry(bucket.clone())
+                .or_insert_with(BucketReplicationResyncStatus::new);
+            if let Some(final_target) = final_map.targets_map.get(&arn) {
+                cached.targets_map.insert(arn.clone(), final_target.clone());
+                cached.last_update = final_map.last_update.or(cached.last_update);
+            }
+        }
+
+        if let Some(opts) = &canceled {
+            self.resyncer.cancel(opts).await;
+        }
+        Ok(canceled)
+    }
+
     pub async fn activate_bucket_resync(self: Arc<Self>, opts: ResyncOpts, recovering: bool) -> Result<(), EcstoreError> {
         let bucket_status = load_bucket_resync_metadata(&opts.bucket, self.storage.clone()).await?;
         let Some(target_status) = bucket_status.targets_map.get(&opts.arn) else {
@@ -2710,6 +2804,11 @@ pub trait ReplicationPoolTrait: std::fmt::Debug {
     async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize);
     async fn get_bucket_resync_status(&self, bucket: &str) -> Result<BucketReplicationResyncStatus, EcstoreError>;
     async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError>;
+    async fn cancel_bucket_resync_for_removed_target(
+        self: Arc<Self>,
+        bucket: &str,
+        arn: &str,
+    ) -> Result<Option<ResyncOpts>, EcstoreError>;
     async fn admit_bucket_resync(self: Arc<Self>, opts: ResyncOpts) -> Result<bool, EcstoreError>;
     async fn activate_bucket_resync(self: Arc<Self>, opts: ResyncOpts, recovering: bool) -> Result<(), EcstoreError>;
     async fn start_bucket_resync(self: Arc<Self>, opts: ResyncOpts) -> Result<(), EcstoreError>;
@@ -2763,6 +2862,14 @@ impl<S: ReplicationStorage> ReplicationPoolTrait for ReplicationPool<S> {
         self.cancel_bucket_resync(opts).await
     }
 
+    async fn cancel_bucket_resync_for_removed_target(
+        self: Arc<Self>,
+        bucket: &str,
+        arn: &str,
+    ) -> Result<Option<ResyncOpts>, EcstoreError> {
+        ReplicationPool::<S>::cancel_bucket_resync_for_removed_target(self, bucket, arn).await
+    }
+
     async fn admit_bucket_resync(self: Arc<Self>, opts: ResyncOpts) -> Result<bool, EcstoreError> {
         self.admit_bucket_resync(opts).await
     }
@@ -3241,11 +3348,17 @@ mod tests {
 
         async fn put_object(
             &self,
-            _bucket: &str,
+            bucket: &str,
             object: &str,
             data: &mut Self::PutObjectReader,
             opts: &Self::ObjectOptions,
         ) -> Result<Self::ObjectInfo, Self::Error> {
+            let _lock_guard = if opts.no_lock {
+                None
+            } else {
+                let lock = self.new_ns_lock(bucket, object).await?;
+                Some(lock.get_write_lock(Duration::from_secs(10)).await?)
+            };
             if opts.http_preconditions.is_some()
                 && let Some(replacement) = self
                     .shared
@@ -3891,6 +4004,157 @@ mod tests {
         assert_eq!(pool.resyncer.cancel_tokens.read().await.len(), 1);
     }
 
+    /// Removing a target on node A must cancel only A's intent. Node A's cached
+    /// status map predates node B's admission, so a cancel that persisted the
+    /// cache would erase B's durable restart intent for `arn:second`.
+    #[tokio::test]
+    async fn removed_target_cancel_preserves_intents_admitted_on_other_nodes() {
+        let shared = empty_resync_shared_state();
+        let node_a = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await;
+        let node_b = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-b", shared.clone()))).await;
+        let bucket = "removed-target-cancel";
+
+        assert!(
+            node_a
+                .clone()
+                .admit_bucket_resync(test_resync_opts(bucket, "arn:first", "run-a"))
+                .await
+                .expect("node A admission should persist")
+        );
+        assert!(
+            node_b
+                .clone()
+                .admit_bucket_resync(test_resync_opts(bucket, "arn:second", "run-b"))
+                .await
+                .expect("node B admission should persist")
+        );
+        assert!(
+            !node_a.resyncer.status_map.read().await[bucket]
+                .targets_map
+                .contains_key("arn:second"),
+            "precondition: node A's cache must be stale relative to node B's admission"
+        );
+
+        let canceled = node_a
+            .clone()
+            .cancel_bucket_resync_for_removed_target(bucket, "arn:first")
+            .await
+            .expect("cancel should succeed");
+        assert_eq!(canceled.map(|opts| opts.resync_id), Some("run-a".to_string()));
+
+        let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned"))
+            .expect("persisted status should decode");
+        assert_eq!(persisted.targets_map["arn:first"].resync_status, ResyncStatusType::ResyncCanceled);
+        assert_eq!(persisted.targets_map["arn:second"].resync_status, ResyncStatusType::ResyncPending);
+        assert_eq!(persisted.targets_map["arn:second"].resync_id, "run-b");
+        // Cache convergence is per-target: only the removed ARN is written
+        // back (a running target's cached progress counters stay
+        // authoritative), so node A's cache reflects the cancel while the
+        // persisted document remains the authority for `arn:second`.
+        assert_eq!(
+            node_a.resyncer.status_map.read().await[bucket].targets_map["arn:first"].resync_status,
+            ResyncStatusType::ResyncCanceled
+        );
+
+        let untouched = node_a
+            .clone()
+            .cancel_bucket_resync_for_removed_target(bucket, "arn:first")
+            .await
+            .expect("cancel of a terminal intent should be a no-op");
+        assert!(untouched.is_none());
+        assert!(
+            node_a
+                .clone()
+                .cancel_bucket_resync_for_removed_target(bucket, "arn:missing")
+                .await
+                .expect("cancel of an unknown arn should be a no-op")
+                .is_none()
+        );
+    }
+
+    /// The reviewer's resurrect scenario: after node A cancels `arn:first`, a
+    /// status write from node B — whose cache still holds the pre-cancel map —
+    /// must not flip `arn:first` back to `Pending` on disk. `mark_status` now
+    /// persists through the CAS with per-target guards instead of blind-saving
+    /// its cached whole-bucket map.
+    #[tokio::test]
+    async fn stale_peer_status_write_cannot_resurrect_canceled_intent() {
+        let shared = empty_resync_shared_state();
+        let node_a = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await;
+        let node_b = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-b", shared.clone()))).await;
+        let bucket = "stale-peer-write";
+
+        assert!(
+            node_a
+                .clone()
+                .admit_bucket_resync(test_resync_opts(bucket, "arn:first", "run-a"))
+                .await
+                .expect("node A admission should persist")
+        );
+        assert!(
+            node_b
+                .clone()
+                .admit_bucket_resync(test_resync_opts(bucket, "arn:second", "run-b"))
+                .await
+                .expect("node B admission should persist")
+        );
+        // Seed node B's stale cache: it saw the map before A's cancel.
+        let pre_cancel = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned"))
+            .expect("pre-cancel status should decode");
+        node_b
+            .resyncer
+            .status_map
+            .write()
+            .await
+            .insert(bucket.to_string(), pre_cancel);
+
+        node_a
+            .clone()
+            .cancel_bucket_resync_for_removed_target(bucket, "arn:first")
+            .await
+            .expect("cancel should succeed")
+            .expect("cancel should report the canceled run");
+
+        node_b
+            .resyncer
+            .mark_status(
+                ResyncStatusType::ResyncStarted,
+                test_resync_opts(bucket, "arn:second", "run-b"),
+                node_b.storage.clone(),
+            )
+            .await
+            .expect("peer status write should succeed");
+
+        let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned"))
+            .expect("persisted status should decode");
+        assert_eq!(
+            persisted.targets_map["arn:first"].resync_status,
+            ResyncStatusType::ResyncCanceled,
+            "peer's stale cache must not resurrect the canceled intent"
+        );
+        assert_eq!(persisted.targets_map["arn:second"].resync_status, ResyncStatusType::ResyncStarted);
+
+        // And the reverse guard: a stale write for the canceled target itself
+        // is refused outright.
+        node_b
+            .resyncer
+            .mark_status(
+                ResyncStatusType::ResyncStarted,
+                test_resync_opts(bucket, "arn:first", "run-a"),
+                node_b.storage.clone(),
+            )
+            .await
+            .expect("guarded status write should be skipped, not fail");
+        let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned"))
+            .expect("persisted status should decode");
+        assert_eq!(persisted.targets_map["arn:first"].resync_status, ResyncStatusType::ResyncCanceled);
+        assert_eq!(
+            node_b.resyncer.status_map.read().await[bucket].targets_map["arn:first"].resync_status,
+            ResyncStatusType::ResyncCanceled,
+            "the refused writer's cache must converge to the persisted terminal state"
+        );
+    }
+
     #[tokio::test]
     async fn admitted_resync_waits_for_target_metadata_commit_before_activation() {
         let shared = empty_resync_shared_state();
@@ -4030,6 +4294,77 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn concurrent_resync_status_cas_preserves_both_mutations() {
+        let shared = empty_resync_shared_state();
+        let mut seeded = BucketReplicationResyncStatus::new();
+        for arn in ["arn:a", "arn:b"] {
+            seeded.targets_map.insert(
+                arn.to_string(),
+                TargetReplicationResyncStatus {
+                    bucket: "cas-race".to_string(),
+                    resync_id: format!("run-{arn}"),
+                    resync_status: ResyncStatusType::ResyncPending,
+                    ..Default::default()
+                },
+            );
+        }
+        *shared.data.lock().expect("test data lock should not be poisoned") =
+            encode_resync_file(&seeded).expect("seeded resync status should encode");
+        shared.empty_object_exists.store(true, Ordering::SeqCst);
+        shared.etag_revision.store(1, Ordering::SeqCst);
+        shared.block_next_write.store(true, Ordering::SeqCst);
+        let node_a = Arc::new(LoadResyncNodeStore::new("cas-node-a", shared.clone()));
+        let node_b = Arc::new(LoadResyncNodeStore::new("cas-node-b", shared.clone()));
+
+        let writer_a = tokio::spawn(async move {
+            update_resync_status_cas("cas-race", node_a, |status| {
+                status
+                    .targets_map
+                    .get_mut("arn:a")
+                    .expect("seeded target A should exist")
+                    .resync_status = ResyncStatusType::ResyncCanceled;
+                Ok(true)
+            })
+            .await
+        });
+
+        tokio::time::timeout(Duration::from_secs(10), shared.write_started.notified())
+            .await
+            .expect("writer A should pause after its precondition check");
+
+        let mut writer_b = tokio::spawn(async move {
+            update_resync_status_cas("cas-race", node_b, |status| {
+                status
+                    .targets_map
+                    .get_mut("arn:b")
+                    .expect("seeded target B should exist")
+                    .resync_status = ResyncStatusType::ResyncCompleted;
+                Ok(true)
+            })
+            .await
+        });
+        let writer_b_before_release = tokio::time::timeout(Duration::from_millis(250), &mut writer_b).await.ok();
+
+        shared.allow_write.notify_one();
+        writer_a
+            .await
+            .expect("writer A task should finish")
+            .expect("writer A should report a successful conditional save");
+        match writer_b_before_release {
+            Some(result) => result,
+            None => writer_b.await,
+        }
+        .expect("writer B task should finish")
+        .expect("writer B should retry and save its mutation");
+
+        let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned"))
+            .expect("persisted resync status should decode");
+        assert_eq!(persisted.targets_map["arn:a"].resync_status, ResyncStatusType::ResyncCanceled);
+        assert_eq!(persisted.targets_map["arn:b"].resync_status, ResyncStatusType::ResyncCompleted);
+        assert!(!shared.last_put_no_lock.load(Ordering::SeqCst));
+    }
+
     #[test]
     fn replication_queue_admission_combines_target_results() {
         let mut admission = ReplicationQueueAdmission::Skipped;

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

@@ -40,16 +40,17 @@ use super::replication_resync_boundary::ResyncStatusType;
 #[cfg(test)]
 use super::replication_resync_boundary::should_count_head_proxy_failure;
 use super::replication_resync_boundary::{
-    BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus, encode_resync_file, is_version_id_mismatch,
-    resync_state_accepts_update, resync_status_duration, sanitize_resync_error_detail,
+    BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus, decode_resync_file, encode_resync_file,
+    is_version_id_mismatch, resync_state_accepts_update, resync_status_duration, sanitize_resync_error_detail,
+    should_auto_resume_resync,
 };
 #[cfg(test)]
-use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX, decode_resync_file};
+use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX};
 #[cfg(test)]
 use super::replication_storage_boundary::ReplicationDeletedObject;
 use super::replication_storage_boundary::{
-    AdvancedGetOptions, EcstoreObjectOperations, GetObjectReader, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete,
-    ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions,
+    AdvancedGetOptions, EcstoreObjectOperations, GetObjectReader, HTTPPreconditions, HTTPRangeSpec, ObjectInfo, ObjectOptions,
+    ObjectToDelete, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions,
 };
 use super::replication_target_boundary::{
     ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions,
@@ -428,7 +429,7 @@ impl ReplicationResyncer {
     where
         S: ReplicationObjectIO,
     {
-        let (bucket_status, status_duration) = {
+        let (updated_target, status_duration) = {
             let mut status_map = self.status_map.write().await;
             let now = OffsetDateTime::now_utc();
 
@@ -499,28 +500,62 @@ impl ReplicationResyncer {
 
             bucket_status.last_update = Some(now);
 
-            (bucket_status.clone(), status_duration)
+            (state.clone(), status_duration)
         };
 
-        save_resync_status(&opts.bucket, &bucket_status, obj_layer.clone()).await?;
-        if status != ResyncStatusType::ResyncCanceled {
-            let canceled_status = self
-                .status_map
-                .read()
-                .await
-                .get(&opts.bucket)
-                .filter(|current| {
-                    current.targets_map.get(&opts.arn).is_some_and(|target| {
-                        target.resync_id == opts.resync_id && target.resync_status == ResyncStatusType::ResyncCanceled
-                    })
-                })
-                .cloned();
-            if let Some(canceled_status) = canceled_status {
-                save_resync_status(&opts.bucket, &canceled_status, obj_layer).await?;
-                return Ok(());
+        // Persist through the CAS so a stale cached map can never clobber
+        // states other nodes finalized for other targets; re-run the staleness
+        // and canceled-is-terminal guards against the freshest persisted entry.
+        let updated_last_update = updated_target.last_update;
+        let (final_map, saved) = update_resync_status_cas(&opts.bucket, obj_layer, |persisted| {
+            if let Some(current) = persisted.targets_map.get(&opts.arn) {
+                if !resync_state_accepts_update(current, &opts) {
+                    debug!(
+                        event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                        bucket = %opts.bucket,
+                        arn = %opts.arn,
+                        incoming_resync_id = %opts.resync_id,
+                        current_resync_id = %current.resync_id,
+                        reason = "stale_status_update",
+                        "Skipped persisting stale resync status update"
+                    );
+                    return Ok(false);
+                }
+                if current.resync_status == ResyncStatusType::ResyncCanceled && status != ResyncStatusType::ResyncCanceled {
+                    debug!(
+                        event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                        bucket = %opts.bucket,
+                        arn = %opts.arn,
+                        incoming_status = %status,
+                        reason = "canceled_status_is_terminal",
+                        "Skipped resync status update after cancellation"
+                    );
+                    return Ok(false);
+                }
+            }
+            persisted.targets_map.insert(opts.arn.clone(), updated_target.clone());
+            persisted.last_update = updated_last_update;
+            Ok(true)
+        })
+        .await?;
+
+        // Converge this target's cached entry with what the persisted document
+        // decided (our update, or the newer/terminal state that outranked it).
+        {
+            let mut status_map = self.status_map.write().await;
+            if let Some(cached) = status_map.get_mut(&opts.bucket)
+                && let Some(final_target) = final_map.targets_map.get(&opts.arn)
+            {
+                cached.targets_map.insert(opts.arn.clone(), final_target.clone());
+                cached.last_update = final_map.last_update.or(cached.last_update);
             }
         }
-        if let Some(stats) = runtime_sources::replication_stats() {
+
+        if saved && let Some(stats) = runtime_sources::replication_stats() {
             stats.record_resync_status(&opts.bucket, status, status_duration).await;
         }
 
@@ -612,10 +647,16 @@ impl ReplicationResyncer {
                 }
                 _ = interval.tick() => {
 
-                    let status_map = self.status_map.read().await;
+                    let snapshot: Vec<(String, BucketReplicationResyncStatus)> = self
+                        .status_map
+                        .read()
+                        .await
+                        .iter()
+                        .map(|(bucket, status)| (bucket.clone(), status.clone()))
+                        .collect();
 
                     let mut update = false;
-                    for (bucket, status) in status_map.iter() {
+                    for (bucket, status) in &snapshot {
                         for target in status.targets_map.values() {
                             if target.last_update.is_none() {
                                 update = true;
@@ -631,7 +672,14 @@ impl ReplicationResyncer {
                             }
 
                         if update {
-                            if let Err(err) = save_resync_status(bucket, status, api.clone()).await {
+                            // CAS-merge instead of a blind whole-map save: this
+                            // cache may lag other nodes' admissions and
+                            // cancellations, which must not be overwritten.
+                            let result = update_resync_status_cas(bucket, api.clone(), |persisted| {
+                                Ok(merge_local_resync_into_persisted(persisted, status))
+                            })
+                            .await;
+                            if let Err(err) = result {
                                 error!(
                                     event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED,
                                     component = LOG_COMPONENT_ECSTORE,
@@ -641,8 +689,8 @@ impl ReplicationResyncer {
                                     error = %err,
                                     "Failed to persist resync status"
                                 );
-                            } else {
-                                last_update_times.insert(bucket.clone(), status.last_update.expect("last_update should be set"));
+                            } else if let Some(last_update) = status.last_update {
+                                last_update_times.insert(bucket.clone(), last_update);
                             }
                         }
                     }
@@ -1451,17 +1499,109 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC
     })
 }
 
-pub(crate) async fn save_resync_status<S: ReplicationObjectIO>(
+/// Upper bound on optimistic retries for a `resync.bin` compare-and-swap
+/// update before giving up; contention on one bucket's status is a handful of
+/// writers (status transitions, the periodic saver, admissions), not a crowd.
+const RESYNC_STATUS_CAS_MAX_ATTEMPTS: usize = 32;
+
+/// Read-merge-write `resync.bin` under an ETag compare-and-swap.
+///
+/// Every writer used to persist its node's cached whole-bucket map, so one
+/// node's stale cache could silently resurrect a state another node had
+/// already finalized (e.g. flip a just-canceled intent back to `Pending`).
+/// `apply` receives the freshest persisted map and mutates it in place,
+/// returning `Ok(false)` to skip the write. On a concurrent write the load +
+/// apply + save cycle is retried against the new document. Returns the final
+/// map and whether this call wrote it.
+pub(crate) async fn update_resync_status_cas<S, F>(
     bucket: &str,
-    status: &BucketReplicationResyncStatus,
     api: Arc<S>,
-) -> Result<()> {
-    let data = encode_resync_file(status)?;
-
+    mut apply: F,
+) -> Result<(BucketReplicationResyncStatus, bool)>
+where
+    S: ReplicationObjectIO,
+    F: FnMut(&mut BucketReplicationResyncStatus) -> Result<bool>,
+{
     let config_file = ReplicationMetadataStore::bucket_resync_file_path(bucket);
-    ReplicationConfigStore::save(api, &config_file, data).await?;
+    for _ in 0..RESYNC_STATUS_CAS_MAX_ATTEMPTS {
+        let (mut status, preconditions) =
+            match ReplicationConfigStore::read_no_lock_with_metadata(api.clone(), &config_file).await {
+                Ok((data, object_info)) => {
+                    let etag = object_info
+                        .etag
+                        .filter(|etag| !etag.trim().is_empty())
+                        .ok_or_else(|| Error::other("replication resync status has no ETag for conditional update"))?;
+                    let status = if data.is_empty() {
+                        BucketReplicationResyncStatus::new()
+                    } else {
+                        decode_resync_file(&data)?
+                    };
+                    (
+                        status,
+                        HTTPPreconditions {
+                            if_match: Some(etag),
+                            ..Default::default()
+                        },
+                    )
+                }
+                Err(Error::ConfigNotFound) => (
+                    BucketReplicationResyncStatus::new(),
+                    HTTPPreconditions {
+                        if_none_match: Some("*".to_string()),
+                        ..Default::default()
+                    },
+                ),
+                Err(err) => return Err(err),
+            };
+        if !apply(&mut status)? {
+            return Ok((status, false));
+        }
+        match ReplicationConfigStore::save_conditional(api.clone(), &config_file, encode_resync_file(&status)?, preconditions)
+            .await
+        {
+            Ok(()) => return Ok((status, true)),
+            Err(Error::PreconditionFailed) => continue,
+            Err(err) => return Err(err),
+        }
+    }
+    Err(Error::other("replication resync status conditional update did not converge"))
+}
 
-    Ok(())
+/// Merge this node's cached bucket resync map into the persisted map for the
+/// periodic saver. Per target: same run id overlays the fresher local state
+/// unless the persisted state is already terminal and the local one is not
+/// (a cancel/completion recorded by another node must stick); a different
+/// persisted run id means a newer admission elsewhere and is kept; targets
+/// unknown to disk are added. Returns whether `persisted` changed.
+pub(crate) fn merge_local_resync_into_persisted(
+    persisted: &mut BucketReplicationResyncStatus,
+    local: &BucketReplicationResyncStatus,
+) -> bool {
+    let mut changed = false;
+    for (arn, local_state) in &local.targets_map {
+        match persisted.targets_map.get(arn) {
+            Some(current) if current.resync_id == local_state.resync_id => {
+                let persisted_terminal = !should_auto_resume_resync(current.resync_status);
+                let local_terminal = !should_auto_resume_resync(local_state.resync_status);
+                if persisted_terminal && !local_terminal {
+                    continue;
+                }
+                if current != local_state {
+                    persisted.targets_map.insert(arn.clone(), local_state.clone());
+                    changed = true;
+                }
+            }
+            Some(_) => {}
+            None => {
+                persisted.targets_map.insert(arn.clone(), local_state.clone());
+                changed = true;
+            }
+        }
+    }
+    if changed && local.last_update.is_some() {
+        persisted.last_update = local.last_update;
+    }
+    changed
 }
 
 pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicationInfo, storage: Arc<S>) {
@@ -3913,6 +4053,87 @@ async fn replicate_object_with_multipart<S: ReplicationObjectIO>(ctx: MultipartR
 #[cfg(test)]
 mod tests {
     use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
+    fn resync_target_state(resync_id: &str, status: ResyncStatusType, replicated_count: i64) -> TargetReplicationResyncStatus {
+        TargetReplicationResyncStatus {
+            resync_id: resync_id.to_string(),
+            resync_status: status,
+            replicated_count,
+            ..Default::default()
+        }
+    }
+
+    /// Periodic-saver merge: fresher local progress overlays the same run,
+    /// but a terminal state persisted by another node must stick, a newer
+    /// admission elsewhere is kept, and locally-known targets are added.
+    #[test]
+    fn merge_local_resync_keeps_peer_terminal_and_newer_states() {
+        let mut persisted = BucketReplicationResyncStatus::new();
+        persisted.targets_map.insert(
+            "arn:same-run".to_string(),
+            resync_target_state("run-1", ResyncStatusType::ResyncStarted, 1),
+        );
+        persisted.targets_map.insert(
+            "arn:canceled".to_string(),
+            resync_target_state("run-1", ResyncStatusType::ResyncCanceled, 0),
+        );
+        persisted.targets_map.insert(
+            "arn:new-run".to_string(),
+            resync_target_state("run-2", ResyncStatusType::ResyncPending, 0),
+        );
+
+        let mut local = BucketReplicationResyncStatus::new();
+        local.targets_map.insert(
+            "arn:same-run".to_string(),
+            resync_target_state("run-1", ResyncStatusType::ResyncStarted, 9),
+        );
+        local.targets_map.insert(
+            "arn:canceled".to_string(),
+            resync_target_state("run-1", ResyncStatusType::ResyncPending, 0),
+        );
+        local.targets_map.insert(
+            "arn:new-run".to_string(),
+            resync_target_state("run-1", ResyncStatusType::ResyncStarted, 3),
+        );
+        local.targets_map.insert(
+            "arn:local-only".to_string(),
+            resync_target_state("run-1", ResyncStatusType::ResyncPending, 0),
+        );
+        local.last_update = Some(OffsetDateTime::now_utc());
+
+        assert!(merge_local_resync_into_persisted(&mut persisted, &local));
+        assert_eq!(persisted.targets_map["arn:same-run"].replicated_count, 9, "fresher local progress wins");
+        assert_eq!(
+            persisted.targets_map["arn:canceled"].resync_status,
+            ResyncStatusType::ResyncCanceled,
+            "peer terminal state must stick"
+        );
+        assert_eq!(
+            persisted.targets_map["arn:new-run"].resync_id, "run-2",
+            "newer admission elsewhere is kept"
+        );
+        assert!(persisted.targets_map.contains_key("arn:local-only"));
+        assert_eq!(persisted.last_update, local.last_update);
+    }
+
+    /// A terminal local state for the same run (completion/failure recorded by
+    /// this node) still overlays a non-terminal persisted state.
+    #[test]
+    fn merge_local_resync_reports_no_change_when_maps_agree() {
+        let mut persisted = BucketReplicationResyncStatus::new();
+        persisted
+            .targets_map
+            .insert("arn:same".to_string(), resync_target_state("run-1", ResyncStatusType::ResyncStarted, 5));
+        let local = persisted.clone();
+        assert!(!merge_local_resync_into_persisted(&mut persisted, &local));
+
+        let mut local = local.clone();
+        local
+            .targets_map
+            .insert("arn:same".to_string(), resync_target_state("run-1", ResyncStatusType::ResyncCompleted, 5));
+        assert!(merge_local_resync_into_persisted(&mut persisted, &local));
+        assert_eq!(persisted.targets_map["arn:same"].resync_status, ResyncStatusType::ResyncCompleted);
+    }
+
     use super::super::replication_target_boundary::{BucketTarget, BucketTargets};
     use super::*;
     use s3s::dto::{

crates/replication/src/resync.rs

@@ -153,7 +153,7 @@ pub struct ResyncOpts {
     pub resync_before: Option<OffsetDateTime>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
 pub struct TargetReplicationResyncStatus {
     pub start_time: Option<OffsetDateTime>,
     pub last_update: Option<OffsetDateTime>,