refactor(replication): split four oversized hot-path functions into focused helpers

Pure-move decomposition of the four oversized functions flagged by the
replication compatibility review (P1-18), unblocking migration milestone
M2 which requires resyncer moves to stay mechanical:

- resync_bucket (522 lines -> 61-line step sequence): leader lock,
  target resolution, walk/collector/worker spawning, and dispatch loop
  extracted into focused helpers; pure decision helpers (DTO builders,
  HEAD-result classification) separated from IO orchestration.
- replicate_all (411 lines -> 113-line main body): initial target-info
  seeding, read/stat option builders, skip-path notes, target HEAD
  action resolution, and the multipart/single-put payload transport
  extracted as private free functions.
- start_mrf_processor (306 lines -> 46-line spawn body): recovery guard,
  ledger load, per-entry replay (delete/object/metadata), and retained
  entry resolution extracted; retry bookkeeping semantics preserved
  exactly (inner continue-paths push inside helpers, outer Missed push
  stays in the loop).
- apply_iam_item (255 lines -> match dispatch skeleton): one helper per
  IAM item type.

No behavior change: log texts, error paths, event emissions, and metric
counts are byte-identical; existing tests unchanged and green (238
ecstore replication/mrf/resync + 232 rustfs site-replication).

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

@@ -667,6 +667,368 @@ async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
     Err(EcstoreError::PreconditionFailed)
 }
 
+/// Acquires the MRF recovery leader lock for the startup replay.
+/// Returns `None` (after logging) when the lock cannot be created or another
+/// node is already processing the backlog.
+async fn acquire_mrf_recovery_guard<S: ReplicationStorage>(storage: &Arc<S>) -> Option<rustfs_lock::NamespaceLockGuard> {
+    let recovery_lock = match storage
+        .new_ns_lock(
+            ReplicationMetadataStore::rustfs_meta_bucket(),
+            ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
+        )
+        .await
+    {
+        Ok(lock) => lock,
+        Err(error) => {
+            warn!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                error = %error,
+                "Failed to create the MRF recovery leader lock"
+            );
+            return None;
+        }
+    };
+    match recovery_lock
+        .get_write_lock_quiet(ReplicationLockTiming::acquire_timeout())
+        .await
+    {
+        Ok(guard) => Some(guard),
+        Err(_) => {
+            debug!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                "Another node is already processing the MRF recovery backlog"
+            );
+            None
+        }
+    }
+}
+
+/// Reads and decodes the on-disk MRF recovery file.
+/// Returns `None` when there is nothing to replay: missing file (publishes an
+/// empty available summary), read failure, or corrupt data (quarantined).
+async fn load_mrf_recovery_entries<S: ReplicationStorage>(storage: &Arc<S>) -> Option<Vec<MrfReplicateEntry>> {
+    let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
+        Ok(d) => d,
+        Err(EcstoreError::ConfigNotFound) => {
+            set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
+                available: true,
+                buckets: Vec::new(),
+            });
+            return None;
+        }
+        Err(e) => {
+            warn!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                error = %e,
+                "Failed to load MRF recovery file"
+            );
+            return None;
+        }
+    };
+
+    match decode_mrf_file(&data) {
+        Ok(v) => Some(v),
+        Err(e) => {
+            warn!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                error = %e,
+                "Failed to decode MRF recovery file — preserving corrupt data"
+            );
+            quarantine_mrf_file(storage, &data).await;
+            None
+        }
+    }
+}
+
+/// Replays one MRF recovery entry by operation kind.
+/// Returns `None` when the entry is skipped entirely (no admission outcome);
+/// entries that must be retried later are pushed onto `retry_entries`.
+async fn replay_mrf_entry<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    match entry.op {
+        MrfOpKind::Delete => replay_mrf_delete_entry(entry, storage, retry_entries).await,
+        MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
+            replay_mrf_object_entry(entry, storage, retry_entries).await
+        }
+        MrfOpKind::Metadata => replay_mrf_metadata_entry(entry, storage, retry_entries).await,
+    }
+}
+
+/// Replays a delete-kind MRF entry: force-delete intents replay directly,
+/// stale force-delete generations are skipped, and plain deletes are
+/// reconstructed as heal deletes.
+async fn replay_mrf_delete_entry<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    if should_replay_force_delete_intent(entry) {
+        let operation_id = entry.force_delete_id?;
+        let delete = force_delete_heal_replication_info(entry, operation_id);
+        if replicate_delete_with_outcome(delete, storage.clone()).await {
+            Some(ReplicationQueueAdmission::Queued)
+        } else {
+            Some(ReplicationQueueAdmission::Missed)
+        }
+    } else if entry.force_delete_id.is_some() {
+        Some(ReplicationQueueAdmission::Skipped)
+    } else {
+        replay_mrf_reconstructed_delete(entry, storage, retry_entries).await
+    }
+}
+
+/// Pure DTO construction: heal replication info for a replayed force-delete intent.
+fn force_delete_heal_replication_info(entry: &MrfReplicateEntry, operation_id: uuid::Uuid) -> DeletedObjectReplicationInfo {
+    DeletedObjectReplicationInfo {
+        delete_object: ReplicationDeletedObject {
+            object_name: entry.object.clone(),
+            force_delete: true,
+            force_delete_id: Some(operation_id),
+            force_delete_target_arns: entry.target_arns.clone(),
+            force_delete_generation: entry.force_delete_generation,
+            ..Default::default()
+        },
+        bucket: entry.bucket.clone(),
+        op_type: ReplicationType::Heal,
+        event_type: REPLICATE_HEAL_DELETE.to_string(),
+        ..Default::default()
+    }
+}
+
+/// Reconstruct a heal delete and re-queue it.  We do NOT call
+/// get_object_info here because the delete-marker or version may
+/// already be absent from the local store — that is expected.
+async fn replay_mrf_reconstructed_delete<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    let versioned = ReplicationVersioningStore::prefix_enabled(&entry.bucket, &entry.object).await;
+    let oi = ObjectInfo {
+        bucket: entry.bucket.clone(),
+        name: entry.object.clone(),
+        version_id: entry.version_id,
+        delete_marker: entry.delete_marker,
+        ..Default::default()
+    };
+    let dsc = resolve_mrf_delete_replicate_decision(entry, &oi, versioned, retry_entries).await?;
+    let dv = reconstructed_heal_delete_info(entry, &oi, &dsc);
+    if replicate_delete_with_outcome(dv, storage.clone()).await {
+        Some(ReplicationQueueAdmission::Queued)
+    } else {
+        Some(ReplicationQueueAdmission::Missed)
+    }
+}
+
+/// The MRF entry does not persist the replication decision and the
+/// source object is gone, so re-derive the decision from the live
+/// bucket config (mirroring get_heal_replicate_object_info) and set
+/// it on the reconstructed delete. Without this the decision string
+/// is empty and the delete replicates to zero targets — a silent
+/// no-op that leaves replicas diverged (backlog#858 / #799 B9).
+async fn resolve_mrf_delete_replicate_decision(
+    entry: &MrfReplicateEntry,
+    oi: &ObjectInfo,
+    versioned: bool,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicateDecision> {
+    if entry.target_arns.is_empty() {
+        match ReplicationMetadataStore::optional_replication_config(&entry.bucket).await {
+            Ok(None) => None,
+            Err(_) => {
+                retry_entries.push(entry.clone());
+                None
+            }
+            Ok(Some(_)) => match check_replicate_delete_strict(
+                &entry.bucket,
+                &ObjectToDelete {
+                    object_name: entry.object.clone(),
+                    version_id: entry.version_id,
+                    ..Default::default()
+                },
+                oi,
+                &ObjectOptions {
+                    versioned,
+                    ..Default::default()
+                },
+                None,
+            )
+            .await
+            {
+                Ok(dsc) => Some(dsc),
+                Err(_) => {
+                    retry_entries.push(entry.clone());
+                    None
+                }
+            },
+        }
+    } else {
+        Some(replicate_decision_for_admitted_targets(&entry.target_arns))
+    }
+}
+
+/// Pure DTO construction: reconstructed heal delete carrying the re-derived
+/// replication decision.
+fn reconstructed_heal_delete_info(
+    entry: &MrfReplicateEntry,
+    oi: &ObjectInfo,
+    dsc: &ReplicateDecision,
+) -> DeletedObjectReplicationInfo {
+    let mut rstate = oi.replication_state();
+    rstate.replicate_decision_str = dsc.to_string();
+
+    let delete_marker_mtime = entry
+        .delete_marker_mtime
+        .and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(i128::from(nanos)).ok());
+
+    DeletedObjectReplicationInfo {
+        delete_object: ReplicationDeletedObject {
+            object_name: entry.object.clone(),
+            version_id: entry.version_id,
+            delete_marker_version_id: entry.delete_marker_version_id,
+            delete_marker: entry.delete_marker,
+            delete_marker_mtime,
+            force_delete: entry.force_delete,
+            replication_state: Some(rstate),
+            ..Default::default()
+        },
+        bucket: entry.bucket.clone(),
+        op_type: ReplicationType::Heal,
+        event_type: REPLICATE_HEAL_DELETE.to_string(),
+        ..Default::default()
+    }
+}
+
+/// Replays an Object/Heal/ExistingObject MRF entry against the live source object.
+async fn replay_mrf_object_entry<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    let opts = ObjectOptions {
+        version_id: entry.version_id.map(|u| u.to_string()),
+        ..Default::default()
+    };
+    let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
+        Ok(oi) => oi,
+        Err(e) => {
+            debug!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                bucket = %entry.bucket,
+                object = %entry.object,
+                error = %e,
+                "MRF recovery: source object lookup failed"
+            );
+            if should_retry_mrf_source_lookup(&e) {
+                retry_entries.push(entry.clone());
+            }
+            return None;
+        }
+    };
+    if entry.target_arns.is_empty() {
+        // Legacy entries predate target admission persistence. They cannot
+        // be safely attributed, so retain the old live-config fallback.
+        Some(queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
+    } else {
+        let roi = admitted_mrf_replicate_object(oi, entry, entry.op.replication_type());
+        if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+            Some(ReplicationQueueAdmission::Queued)
+        } else {
+            Some(ReplicationQueueAdmission::Missed)
+        }
+    }
+}
+
+/// Replays a metadata-kind MRF entry against the live source object.
+async fn replay_mrf_metadata_entry<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    let opts = ObjectOptions {
+        version_id: entry.version_id.map(|u| u.to_string()),
+        ..Default::default()
+    };
+    let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
+        Ok(oi) => oi,
+        Err(e) => {
+            debug!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                bucket = %entry.bucket,
+                object = %entry.object,
+                error = %e,
+                "MRF metadata recovery: source object lookup failed"
+            );
+            if should_retry_mrf_source_lookup(&e) {
+                retry_entries.push(entry.clone());
+            }
+            return None;
+        }
+    };
+    if entry.target_arns.is_empty() {
+        Some(queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
+    } else {
+        let roi = admitted_mrf_replicate_object(oi, entry, ReplicationType::Metadata);
+        if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+            Some(ReplicationQueueAdmission::Queued)
+        } else {
+            Some(ReplicationQueueAdmission::Missed)
+        }
+    }
+}
+
+/// Pure DTO construction: replicate-object info for an entry with persisted
+/// admitted targets, carrying over the entry's retry count.
+fn admitted_mrf_replicate_object(oi: ObjectInfo, entry: &MrfReplicateEntry, op_type: ReplicationType) -> ReplicateObjectInfo {
+    let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
+    let mut roi = replicate_object_info_from_object_info(oi, dsc, op_type);
+    roi.retry_count = entry.retry_count.max(0) as u32;
+    roi
+}
+
+/// Acknowledges the replayed MRF prefix and returns the retained backlog.
+/// On acknowledgement failure the backlog is preserved for the next startup and
+/// re-read (falling back to the replayed snapshot) so the published summary stays accurate.
+async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
+    storage: &Arc<S>,
+    recovery_guard: &rustfs_lock::NamespaceLockGuard,
+    entries: &[MrfReplicateEntry],
+    retry_entries: &[MrfReplicateEntry],
+) -> Vec<MrfReplicateEntry> {
+    match acknowledge_mrf_recovery(storage.clone(), recovery_guard, entries, retry_entries).await {
+        Ok(retained) => retained,
+        Err(error) => {
+            warn!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                error = %error,
+                "Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
+            );
+            match read_mrf_entries(storage.clone()).await {
+                Ok(current) => current,
+                Err(read_error) => {
+                    warn!(
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION,
+                        error = %read_error,
+                        "Failed to refresh the MRF backlog after acknowledgement failure"
+                    );
+                    entries.to_vec()
+                }
+            }
+        }
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")]
 struct ResyncActiveConflictError {
@@ -1221,71 +1583,12 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         let storage = self.storage.clone();
 
         let handle = tokio::spawn(async move {
-            let recovery_lock = match storage
-                .new_ns_lock(
-                    ReplicationMetadataStore::rustfs_meta_bucket(),
-                    ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
-                )
-                .await
-            {
-                Ok(lock) => lock,
-                Err(error) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %error,
-                        "Failed to create the MRF recovery leader lock"
-                    );
-                    return;
-                }
-            };
-            let recovery_guard = match recovery_lock
-                .get_write_lock_quiet(ReplicationLockTiming::acquire_timeout())
-                .await
-            {
-                Ok(guard) => guard,
-                Err(_) => {
-                    debug!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        "Another node is already processing the MRF recovery backlog"
-                    );
-                    return;
-                }
+            let Some(recovery_guard) = acquire_mrf_recovery_guard(&storage).await else {
+                return;
             };
 
-            let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
-                Ok(d) => d,
-                Err(EcstoreError::ConfigNotFound) => {
-                    set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
-                        available: true,
-                        buckets: Vec::new(),
-                    });
-                    return;
-                }
-                Err(e) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %e,
-                        "Failed to load MRF recovery file"
-                    );
-                    return;
-                }
-            };
-
-            let entries = match decode_mrf_file(&data) {
-                Ok(v) => v,
-                Err(e) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %e,
-                        "Failed to decode MRF recovery file — preserving corrupt data"
-                    );
-                    quarantine_mrf_file(&storage, &data).await;
-                    return;
-                }
+            let Some(entries) = load_mrf_recovery_entries(&storage).await else {
+                return;
             };
             set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
 
@@ -1294,187 +1597,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             let mut retry_entries = Vec::new();
 
             for entry in entries.iter() {
-                let admission = match entry.op {
-                    MrfOpKind::Delete => {
-                        if should_replay_force_delete_intent(entry) {
-                            let Some(operation_id) = entry.force_delete_id else {
-                                continue;
-                            };
-                            let delete = DeletedObjectReplicationInfo {
-                                delete_object: ReplicationDeletedObject {
-                                    object_name: entry.object.clone(),
-                                    force_delete: true,
-                                    force_delete_id: Some(operation_id),
-                                    force_delete_target_arns: entry.target_arns.clone(),
-                                    force_delete_generation: entry.force_delete_generation,
-                                    ..Default::default()
-                                },
-                                bucket: entry.bucket.clone(),
-                                op_type: ReplicationType::Heal,
-                                event_type: REPLICATE_HEAL_DELETE.to_string(),
-                                ..Default::default()
-                            };
-                            if replicate_delete_with_outcome(delete, storage.clone()).await {
-                                ReplicationQueueAdmission::Queued
-                            } else {
-                                ReplicationQueueAdmission::Missed
-                            }
-                        } else if entry.force_delete_id.is_some() {
-                            ReplicationQueueAdmission::Skipped
-                        } else {
-                            // Reconstruct a heal delete and re-queue it.  We do NOT call
-                            // get_object_info here because the delete-marker or version may
-                            // already be absent from the local store — that is expected.
-                            //
-                            // The MRF entry does not persist the replication decision and the
-                            // source object is gone, so re-derive the decision from the live
-                            // bucket config (mirroring get_heal_replicate_object_info) and set
-                            // it on the reconstructed delete. Without this the decision string
-                            // is empty and the delete replicates to zero targets — a silent
-                            // no-op that leaves replicas diverged (backlog#858 / #799 B9).
-                            let versioned = ReplicationVersioningStore::prefix_enabled(&entry.bucket, &entry.object).await;
-                            let oi = ObjectInfo {
-                                bucket: entry.bucket.clone(),
-                                name: entry.object.clone(),
-                                version_id: entry.version_id,
-                                delete_marker: entry.delete_marker,
-                                ..Default::default()
-                            };
-                            let dsc = if entry.target_arns.is_empty() {
-                                match ReplicationMetadataStore::optional_replication_config(&entry.bucket).await {
-                                    Ok(None) => continue,
-                                    Err(_) => {
-                                        retry_entries.push(entry.clone());
-                                        continue;
-                                    }
-                                    Ok(Some(_)) => match check_replicate_delete_strict(
-                                        &entry.bucket,
-                                        &ObjectToDelete {
-                                            object_name: entry.object.clone(),
-                                            version_id: entry.version_id,
-                                            ..Default::default()
-                                        },
-                                        &oi,
-                                        &ObjectOptions {
-                                            versioned,
-                                            ..Default::default()
-                                        },
-                                        None,
-                                    )
-                                    .await
-                                    {
-                                        Ok(dsc) => dsc,
-                                        Err(_) => {
-                                            retry_entries.push(entry.clone());
-                                            continue;
-                                        }
-                                    },
-                                }
-                            } else {
-                                replicate_decision_for_admitted_targets(&entry.target_arns)
-                            };
-                            let mut rstate = oi.replication_state();
-                            rstate.replicate_decision_str = dsc.to_string();
-
-                            let delete_marker_mtime = entry
-                                .delete_marker_mtime
-                                .and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(i128::from(nanos)).ok());
-
-                            let dv = DeletedObjectReplicationInfo {
-                                delete_object: ReplicationDeletedObject {
-                                    object_name: entry.object.clone(),
-                                    version_id: entry.version_id,
-                                    delete_marker_version_id: entry.delete_marker_version_id,
-                                    delete_marker: entry.delete_marker,
-                                    delete_marker_mtime,
-                                    force_delete: entry.force_delete,
-                                    replication_state: Some(rstate),
-                                    ..Default::default()
-                                },
-                                bucket: entry.bucket.clone(),
-                                op_type: ReplicationType::Heal,
-                                event_type: REPLICATE_HEAL_DELETE.to_string(),
-                                ..Default::default()
-                            };
-                            if replicate_delete_with_outcome(dv, storage.clone()).await {
-                                ReplicationQueueAdmission::Queued
-                            } else {
-                                ReplicationQueueAdmission::Missed
-                            }
-                        }
-                    }
-                    MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
-                        let opts = ObjectOptions {
-                            version_id: entry.version_id.map(|u| u.to_string()),
-                            ..Default::default()
-                        };
-                        let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
-                            Ok(oi) => oi,
-                            Err(e) => {
-                                debug!(
-                                    component = LOG_COMPONENT_ECSTORE,
-                                    subsystem = LOG_SUBSYSTEM_REPLICATION,
-                                    bucket = %entry.bucket,
-                                    object = %entry.object,
-                                    error = %e,
-                                    "MRF recovery: source object lookup failed"
-                                );
-                                if should_retry_mrf_source_lookup(&e) {
-                                    retry_entries.push(entry.clone());
-                                }
-                                continue;
-                            }
-                        };
-                        if entry.target_arns.is_empty() {
-                            // Legacy entries predate target admission persistence. They cannot
-                            // be safely attributed, so retain the old live-config fallback.
-                            queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
-                        } else {
-                            let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
-                            let mut roi = replicate_object_info_from_object_info(oi, dsc, entry.op.replication_type());
-                            roi.retry_count = entry.retry_count.max(0) as u32;
-                            if replicate_object_with_outcome(roi, storage.clone()).await.1 {
-                                ReplicationQueueAdmission::Queued
-                            } else {
-                                ReplicationQueueAdmission::Missed
-                            }
-                        }
-                    }
-                    MrfOpKind::Metadata => {
-                        let opts = ObjectOptions {
-                            version_id: entry.version_id.map(|u| u.to_string()),
-                            ..Default::default()
-                        };
-                        let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
-                            Ok(oi) => oi,
-                            Err(e) => {
-                                debug!(
-                                    component = LOG_COMPONENT_ECSTORE,
-                                    subsystem = LOG_SUBSYSTEM_REPLICATION,
-                                    bucket = %entry.bucket,
-                                    object = %entry.object,
-                                    error = %e,
-                                    "MRF metadata recovery: source object lookup failed"
-                                );
-                                if should_retry_mrf_source_lookup(&e) {
-                                    retry_entries.push(entry.clone());
-                                }
-                                continue;
-                            }
-                        };
-                        if entry.target_arns.is_empty() {
-                            queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
-                        } else {
-                            let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
-                            let mut roi = replicate_object_info_from_object_info(oi, dsc, ReplicationType::Metadata);
-                            roi.retry_count = entry.retry_count.max(0) as u32;
-                            if replicate_object_with_outcome(roi, storage.clone()).await.1 {
-                                ReplicationQueueAdmission::Queued
-                            } else {
-                                ReplicationQueueAdmission::Missed
-                            }
-                        }
-                    }
+                let Some(admission) = replay_mrf_entry(entry, &storage, &mut retry_entries).await else {
+                    continue;
                 };
 
                 if admission == ReplicationQueueAdmission::Missed {
@@ -1484,29 +1608,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                 }
             }
 
-            let retained = match acknowledge_mrf_recovery(storage.clone(), &recovery_guard, &entries, &retry_entries).await {
-                Ok(retained) => retained,
-                Err(error) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %error,
-                        "Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
-                    );
-                    match read_mrf_entries(storage.clone()).await {
-                        Ok(current) => current,
-                        Err(read_error) => {
-                            warn!(
-                                component = LOG_COMPONENT_ECSTORE,
-                                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                                error = %read_error,
-                                "Failed to refresh the MRF backlog after acknowledgement failure"
-                            );
-                            entries.clone()
-                        }
-                    }
-                }
-            };
+            let retained = resolve_retained_mrf_entries(&storage, &recovery_guard, &entries, &retry_entries).await;
             let retained_count = retained.len();
             set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
 

rustfs/src/admin/handlers/site_replication.rs

@@ -66,18 +66,20 @@ use rustfs_config::{
 };
 use rustfs_iam::error::is_err_no_such_service_account;
 use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
+use rustfs_iam::store::object::ObjectStore;
 use rustfs_iam::store::{MappedPolicy, UserType, sr_wire_user_type, user_type_from_sr_wire};
 use rustfs_iam::sys::{
-    NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret,
+    IamSys, NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret,
 };
 use rustfs_madmin::{
     AddOrUpdateUserReq, BucketBandwidth, GroupAddRemove, GroupStatus, IDPSettings, InProgressMetric, InQueueMetric,
     LDAPConfigSettings, LDAPSettings, OpenIDProviderSettings, PeerInfo, PeerSite, QStat, ReplProxyMetric, ReplicateAddStatus,
     ReplicateEditStatus, ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC,
     SR_IAM_ITEM_STS_ACC_LEGACY, SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem,
-    SRIAMPolicy, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation,
-    SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRSiteSummary,
-    SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
+    SRIAMPolicy, SRIAMUser, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq,
+    SRPendingOperation, SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRRetryStats, SRSTSCredential,
+    SRSessionPolicy, SRSiteSummary, SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate,
+    SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
 };
 use rustfs_policy::policy::{
     Policy,
@@ -9278,247 +9280,16 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
     let incoming_updated_at = item.updated_at;
 
     match item.r#type.as_str() {
-        "policy" => {
-            if let Some(policy) = item.policy {
-                let policy: Policy =
-                    serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
-                iam_sys.set_policy(&item.name, policy).await.map_err(ApiError::from)?;
-            } else {
-                iam_sys.delete_policy(&item.name, true).await.map_err(ApiError::from)?;
-            }
-            Ok(())
-        }
-        "policy-mapping" => {
-            let Some(mapping) = item.policy_mapping else {
-                return Err(s3_error!(InvalidRequest, "policyMapping is required"));
-            };
-            let user_type =
-                user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
-            iam_sys
-                .policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
-                .await
-                .map_err(ApiError::from)?;
-            Ok(())
-        }
-        "group-info" => {
-            let Some(group_info) = item.group_info else {
-                return Err(s3_error!(InvalidRequest, "groupInfo is required"));
-            };
-            let update = group_info.update_req;
-            if !group_info_requires_upsert(&update) {
-                iam_sys
-                    .remove_users_from_group(&update.group, update.members)
-                    .await
-                    .map_err(ApiError::from)?;
-                return Ok(());
-            }
-
-            iam_sys
-                .add_users_to_group(&update.group, update.members)
-                .await
-                .map_err(ApiError::from)?;
-            iam_sys
-                .set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
-                .await
-                .map_err(ApiError::from)?;
-            Ok(())
-        }
+        "policy" => apply_iam_policy_item(&iam_sys, &item.name, item.policy).await,
+        "policy-mapping" => apply_iam_policy_mapping_item(&iam_sys, item.policy_mapping).await,
+        "group-info" => apply_iam_group_info_item(&iam_sys, item.group_info).await,
         // MinIO madmin-go sends `SRIAMItemSTSAcc = "sts-account"`. The legacy alias
         // `sts-credential` (emitted by older RustFS releases) stays accepted permanently
         // so mixed-version RustFS sites keep replicating STS credentials during rolling
         // upgrades; it is a compatibility layer, not temporary code.
-        SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => {
-            let Some(sts_credential) = item.sts_credential else {
-                return Err(s3_error!(InvalidRequest, "stsCredential is required"));
-            };
-            let Some(secret) = current_token_signing_key() else {
-                return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
-            };
-            let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
-                .map_err(|e| s3_error!(InvalidRequest, "invalid STS session token: {e}"))?;
-            let expiration = claims
-                .get("exp")
-                .and_then(claims_unix_timestamp)
-                .map(OffsetDateTime::from_unix_timestamp)
-                .transpose()
-                .map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?;
-            let groups = string_list_claim(&claims, "groups");
-            let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping);
-            let cred = rustfs_credentials::Credentials {
-                access_key: sts_credential.access_key.clone(),
-                secret_key: sts_credential.secret_key.clone(),
-                session_token: sts_credential.session_token.clone(),
-                expiration,
-                status: "on".to_string(),
-                parent_user: sts_credential.parent_user.clone(),
-                groups,
-                claims: Some(claims),
-                ..Default::default()
-            };
-            iam_sys
-                .set_temp_user(&sts_credential.access_key, &cred, compatibility_policy)
-                .await
-                .map_err(ApiError::from)?;
-            Ok(())
-        }
-        "iam-user" => {
-            let Some(user) = item.iam_user else {
-                return Err(s3_error!(InvalidRequest, "iamUser is required"));
-            };
-            if let Some(local) = iam_sys.get_user(&user.access_key).await
-                && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
-            {
-                return Ok(());
-            }
-            if user.is_delete_req {
-                iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
-            } else {
-                let Some(user_req) = user.user_req else {
-                    return Err(s3_error!(InvalidRequest, "userReq is required"));
-                };
-                let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
-                if is_status_only_update {
-                    iam_sys
-                        .set_user_status(&user.access_key, user_req.status)
-                        .await
-                        .map_err(ApiError::from)?;
-                } else {
-                    iam_sys
-                        .create_user(&user.access_key, &user_req)
-                        .await
-                        .map_err(ApiError::from)?;
-                }
-            }
-            Ok(())
-        }
-        "service-account" => {
-            let Some(change) = item.svc_acc_change else {
-                return Err(s3_error!(InvalidRequest, "serviceAccountChange is required"));
-            };
-            let envelope = change.oidc_service_account_envelope;
-            if let Some(create) = change.create {
-                let local_updated_at = iam_sys
-                    .get_user(&create.access_key)
-                    .await
-                    .map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
-                let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
-                    if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
-                        return Ok(());
-                    }
-                    ReplicatedServiceAccountPolicy {
-                        policy: Some(site_replicator_service_account_policy()?),
-                        is_envelope: false,
-                    }
-                } else {
-                    let Some(replicated_policy) = decode_service_account_replication_policy(
-                        &create,
-                        envelope.as_ref(),
-                        incoming_updated_at,
-                        local_updated_at,
-                    )?
-                    else {
-                        return Ok(());
-                    };
-                    replicated_policy
-                };
-                match iam_sys.get_service_account(&create.access_key).await {
-                    Ok((existing, _)) => {
-                        if existing.parent_user != create.parent {
-                            return Err(s3_error!(
-                                InvalidRequest,
-                                "service account {} already exists with a different parent user",
-                                create.access_key
-                            ));
-                        }
-                        iam_sys
-                            .update_service_account(
-                                &create.access_key,
-                                UpdateServiceAccountOpts {
-                                    name: replicated_policy.metadata_for_existing_account(create.name),
-                                    description: replicated_policy.metadata_for_existing_account(create.description),
-                                    session_policy: replicated_policy.for_existing_account(),
-                                    secret_key: Some(create.secret_key),
-                                    expiration: create.expiration,
-                                    status: (!create.status.is_empty()).then_some(create.status),
-                                    parent_user: None,
-                                    allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
-                                },
-                            )
-                            .await
-                            .map_err(ApiError::from)?;
-                    }
-                    Err(err) if is_err_no_such_service_account(&err) => {
-                        iam_sys
-                            .new_service_account(
-                                &create.parent,
-                                Some(create.groups),
-                                NewServiceAccountOpts {
-                                    session_policy: replicated_policy.policy,
-                                    access_key: create.access_key,
-                                    secret_key: create.secret_key,
-                                    name: (!create.name.is_empty()).then_some(create.name),
-                                    description: (!create.description.is_empty()).then_some(create.description),
-                                    expiration: create.expiration,
-                                    allow_site_replicator_account: true,
-                                    claims: Some(create.claims),
-                                },
-                            )
-                            .await
-                            .map_err(ApiError::from)?;
-                    }
-                    Err(err) => return Err(ApiError::from(err).into()),
-                }
-                return Ok(());
-            }
-
-            if let Some(update) = change.update {
-                if let Some(local) = iam_sys.get_user(&update.access_key).await
-                    && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
-                {
-                    return Ok(());
-                }
-                let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT;
-                let session_policy = if allow_site_replicator_account {
-                    Some(site_replicator_service_account_policy()?)
-                } else {
-                    update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok())
-                };
-                iam_sys
-                    .update_service_account(
-                        &update.access_key,
-                        UpdateServiceAccountOpts {
-                            session_policy,
-                            secret_key: (!update.secret_key.is_empty()).then_some(update.secret_key),
-                            name: (!update.name.is_empty()).then_some(update.name),
-                            description: (!update.description.is_empty()).then_some(update.description),
-                            expiration: update.expiration,
-                            status: (!update.status.is_empty()).then_some(update.status),
-                            // Peers replicate credentials, never the local parent binding:
-                            // each site resolves its own parent from its own IAM.
-                            parent_user: None,
-                            allow_site_replicator_account,
-                        },
-                    )
-                    .await
-                    .map_err(ApiError::from)?;
-                return Ok(());
-            }
-
-            if let Some(delete) = change.delete {
-                if let Some(local) = iam_sys.get_user(&delete.access_key).await
-                    && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
-                {
-                    return Ok(());
-                }
-                iam_sys
-                    .delete_service_account(&delete.access_key, true)
-                    .await
-                    .map_err(ApiError::from)?;
-                return Ok(());
-            }
-
-            Err(s3_error!(InvalidRequest, "serviceAccountChange is empty"))
-        }
+        SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => apply_iam_sts_account_item(&iam_sys, item.sts_credential).await,
+        "iam-user" => apply_iam_user_item(&iam_sys, item.iam_user, incoming_updated_at).await,
+        "service-account" => apply_iam_service_account_item(&iam_sys, item.svc_acc_change, incoming_updated_at).await,
         _ => Err(s3_error!(
             NotImplemented,
             "site replication IAM item type `{}` is not supported",
@@ -9527,6 +9298,252 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
     }
 }
 
+async fn apply_iam_policy_item(iam_sys: &IamSys<ObjectStore>, name: &str, policy: Option<Value>) -> S3Result<()> {
+    if let Some(policy) = policy {
+        let policy: Policy =
+            serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
+        iam_sys.set_policy(name, policy).await.map_err(ApiError::from)?;
+    } else {
+        iam_sys.delete_policy(name, true).await.map_err(ApiError::from)?;
+    }
+    Ok(())
+}
+
+async fn apply_iam_policy_mapping_item(iam_sys: &IamSys<ObjectStore>, policy_mapping: Option<SRPolicyMapping>) -> S3Result<()> {
+    let Some(mapping) = policy_mapping else {
+        return Err(s3_error!(InvalidRequest, "policyMapping is required"));
+    };
+    let user_type = user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
+    iam_sys
+        .policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
+        .await
+        .map_err(ApiError::from)?;
+    Ok(())
+}
+
+async fn apply_iam_group_info_item(iam_sys: &IamSys<ObjectStore>, group_info: Option<SRGroupInfo>) -> S3Result<()> {
+    let Some(group_info) = group_info else {
+        return Err(s3_error!(InvalidRequest, "groupInfo is required"));
+    };
+    let update = group_info.update_req;
+    if !group_info_requires_upsert(&update) {
+        iam_sys
+            .remove_users_from_group(&update.group, update.members)
+            .await
+            .map_err(ApiError::from)?;
+        return Ok(());
+    }
+
+    iam_sys
+        .add_users_to_group(&update.group, update.members)
+        .await
+        .map_err(ApiError::from)?;
+    iam_sys
+        .set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
+        .await
+        .map_err(ApiError::from)?;
+    Ok(())
+}
+
+async fn apply_iam_sts_account_item(iam_sys: &IamSys<ObjectStore>, sts_credential: Option<SRSTSCredential>) -> S3Result<()> {
+    let Some(sts_credential) = sts_credential else {
+        return Err(s3_error!(InvalidRequest, "stsCredential is required"));
+    };
+    let Some(secret) = current_token_signing_key() else {
+        return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
+    };
+    let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
+        .map_err(|e| s3_error!(InvalidRequest, "invalid STS session token: {e}"))?;
+    let expiration = claims
+        .get("exp")
+        .and_then(claims_unix_timestamp)
+        .map(OffsetDateTime::from_unix_timestamp)
+        .transpose()
+        .map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?;
+    let groups = string_list_claim(&claims, "groups");
+    let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping);
+    let cred = rustfs_credentials::Credentials {
+        access_key: sts_credential.access_key.clone(),
+        secret_key: sts_credential.secret_key.clone(),
+        session_token: sts_credential.session_token.clone(),
+        expiration,
+        status: "on".to_string(),
+        parent_user: sts_credential.parent_user.clone(),
+        groups,
+        claims: Some(claims),
+        ..Default::default()
+    };
+    iam_sys
+        .set_temp_user(&sts_credential.access_key, &cred, compatibility_policy)
+        .await
+        .map_err(ApiError::from)?;
+    Ok(())
+}
+
+async fn apply_iam_user_item(
+    iam_sys: &IamSys<ObjectStore>,
+    iam_user: Option<SRIAMUser>,
+    incoming_updated_at: Option<OffsetDateTime>,
+) -> S3Result<()> {
+    let Some(user) = iam_user else {
+        return Err(s3_error!(InvalidRequest, "iamUser is required"));
+    };
+    if let Some(local) = iam_sys.get_user(&user.access_key).await
+        && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
+    {
+        return Ok(());
+    }
+    if user.is_delete_req {
+        iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
+    } else {
+        let Some(user_req) = user.user_req else {
+            return Err(s3_error!(InvalidRequest, "userReq is required"));
+        };
+        let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
+        if is_status_only_update {
+            iam_sys
+                .set_user_status(&user.access_key, user_req.status)
+                .await
+                .map_err(ApiError::from)?;
+        } else {
+            iam_sys
+                .create_user(&user.access_key, &user_req)
+                .await
+                .map_err(ApiError::from)?;
+        }
+    }
+    Ok(())
+}
+
+async fn apply_iam_service_account_item(
+    iam_sys: &IamSys<ObjectStore>,
+    svc_acc_change: Option<SRSvcAccChange>,
+    incoming_updated_at: Option<OffsetDateTime>,
+) -> S3Result<()> {
+    let Some(change) = svc_acc_change else {
+        return Err(s3_error!(InvalidRequest, "serviceAccountChange is required"));
+    };
+    let envelope = change.oidc_service_account_envelope;
+    if let Some(create) = change.create {
+        let local_updated_at = iam_sys
+            .get_user(&create.access_key)
+            .await
+            .map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
+        let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
+            if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
+                return Ok(());
+            }
+            ReplicatedServiceAccountPolicy {
+                policy: Some(site_replicator_service_account_policy()?),
+                is_envelope: false,
+            }
+        } else {
+            let Some(replicated_policy) =
+                decode_service_account_replication_policy(&create, envelope.as_ref(), incoming_updated_at, local_updated_at)?
+            else {
+                return Ok(());
+            };
+            replicated_policy
+        };
+        match iam_sys.get_service_account(&create.access_key).await {
+            Ok((existing, _)) => {
+                if existing.parent_user != create.parent {
+                    return Err(s3_error!(
+                        InvalidRequest,
+                        "service account {} already exists with a different parent user",
+                        create.access_key
+                    ));
+                }
+                iam_sys
+                    .update_service_account(
+                        &create.access_key,
+                        UpdateServiceAccountOpts {
+                            name: replicated_policy.metadata_for_existing_account(create.name),
+                            description: replicated_policy.metadata_for_existing_account(create.description),
+                            session_policy: replicated_policy.for_existing_account(),
+                            secret_key: Some(create.secret_key),
+                            expiration: create.expiration,
+                            status: (!create.status.is_empty()).then_some(create.status),
+                            parent_user: None,
+                            allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
+                        },
+                    )
+                    .await
+                    .map_err(ApiError::from)?;
+            }
+            Err(err) if is_err_no_such_service_account(&err) => {
+                iam_sys
+                    .new_service_account(
+                        &create.parent,
+                        Some(create.groups),
+                        NewServiceAccountOpts {
+                            session_policy: replicated_policy.policy,
+                            access_key: create.access_key,
+                            secret_key: create.secret_key,
+                            name: (!create.name.is_empty()).then_some(create.name),
+                            description: (!create.description.is_empty()).then_some(create.description),
+                            expiration: create.expiration,
+                            allow_site_replicator_account: true,
+                            claims: Some(create.claims),
+                        },
+                    )
+                    .await
+                    .map_err(ApiError::from)?;
+            }
+            Err(err) => return Err(ApiError::from(err).into()),
+        }
+        return Ok(());
+    }
+
+    if let Some(update) = change.update {
+        if let Some(local) = iam_sys.get_user(&update.access_key).await
+            && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
+        {
+            return Ok(());
+        }
+        let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT;
+        let session_policy = if allow_site_replicator_account {
+            Some(site_replicator_service_account_policy()?)
+        } else {
+            update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok())
+        };
+        iam_sys
+            .update_service_account(
+                &update.access_key,
+                UpdateServiceAccountOpts {
+                    session_policy,
+                    secret_key: (!update.secret_key.is_empty()).then_some(update.secret_key),
+                    name: (!update.name.is_empty()).then_some(update.name),
+                    description: (!update.description.is_empty()).then_some(update.description),
+                    expiration: update.expiration,
+                    status: (!update.status.is_empty()).then_some(update.status),
+                    // Peers replicate credentials, never the local parent binding:
+                    // each site resolves its own parent from its own IAM.
+                    parent_user: None,
+                    allow_site_replicator_account,
+                },
+            )
+            .await
+            .map_err(ApiError::from)?;
+        return Ok(());
+    }
+
+    if let Some(delete) = change.delete {
+        if let Some(local) = iam_sys.get_user(&delete.access_key).await
+            && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
+        {
+            return Ok(());
+        }
+        iam_sys
+            .delete_service_account(&delete.access_key, true)
+            .await
+            .map_err(ApiError::from)?;
+        return Ok(());
+    }
+
+    Err(s3_error!(InvalidRequest, "serviceAccountChange is empty"))
+}
+
 fn claims_unix_timestamp(value: &Value) -> Option<i64> {
     match value {
         Value::Number(number) => number.as_i64(),