diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs
index 58b3efdca..efe517b99 100644
--- a/crates/ecstore/src/bucket/replication/replication_pool.rs
+++ b/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));
 
diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs
index 30198a75b..d234c40e7 100644
--- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs
+++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs
@@ -15,7 +15,7 @@
 use super::replication_bandwidth_boundary;
 use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt as _};
 use super::replication_config_store::ReplicationConfigStore;
-use super::replication_error_boundary::{Result, is_err_object_not_found, is_err_version_not_found};
+use super::replication_error_boundary::{Error, Result, is_err_object_not_found, is_err_version_not_found};
 use super::replication_event_sink::{EventArgs, send_event, send_local_event};
 use super::replication_filemeta_boundary::{
     MrfReplicateEntry, NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo,
@@ -36,15 +36,16 @@ use super::replication_object_decision_boundary::{
 };
 use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission};
 use super::replication_resync_boundary::ResyncStatusType;
+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, sanitize_resync_error_detail, should_count_head_proxy_failure,
+    resync_state_accepts_update, sanitize_resync_error_detail,
 };
 #[cfg(test)]
 use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX, decode_resync_file};
 use super::replication_storage_boundary::{
-    AdvancedGetOptions, EcstoreObjectOperations, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete,
-    ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, WalkOptions,
+    AdvancedGetOptions, EcstoreObjectOperations, GetObjectReader, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete,
+    ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions,
 };
 use super::replication_target_boundary::{
     PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, replication_action_for_target_head,
@@ -182,15 +183,6 @@ fn has_raw_status(err: &SdkError<HeadObjectError>, status: u16) -> bool {
     err.raw_response().is_some_and(|r| r.status().as_u16() == status)
 }
 
-fn is_head_proxy_failure(err: &SdkError<HeadObjectError>) -> bool {
-    let (is_not_found, code) = err
-        .as_service_error()
-        .map(|service_err| (service_err.is_not_found(), service_err.code()))
-        .unwrap_or((false, None));
-    let raw_status = err.raw_response().map(|resp| resp.status().as_u16());
-    should_count_head_proxy_failure(is_not_found, code, raw_status)
-}
-
 const METRIC_VERSION_IDENTITY_DRIFT_TOTAL: &str = "rustfs_replication_version_identity_drift_total";
 
 /// Targets that already produced a version-identity-drift warning this
@@ -242,6 +234,15 @@ fn audit_target_version_identity(tgt_client: &TargetClient, source_version_id: &
     }
 }
 
+fn is_head_proxy_failure(err: &SdkError<HeadObjectError>) -> bool {
+    let (is_not_found, code) = err
+        .as_service_error()
+        .map(|service_err| (service_err.is_not_found(), service_err.code()))
+        .unwrap_or((false, None));
+    let raw_status = err.raw_response().map(|resp| resp.status().as_u16());
+    should_count_head_proxy_failure(is_not_found, code, raw_status)
+}
+
 async fn record_proxy_request(bucket: &str, api: &str, is_err: bool) {
     if let Some(stats) = runtime_sources::replication_stats() {
         stats.inc_proxy(bucket, api, is_err).await;
@@ -632,31 +633,13 @@ impl ReplicationResyncer {
         }
     }
 
-    #[instrument(skip(cancellation_token, storage))]
-    pub async fn resync_bucket<S: ReplicationStorage>(
-        self: Arc<Self>,
-        cancellation_token: CancellationToken,
-        storage: Arc<S>,
-        heal: bool,
-        opts: ResyncOpts,
-    ) {
-        // Check cancellation before starting the scan.
-        // NOTE: the previous design waited here on `worker_rx.resubscribe().recv()` to
-        // throttle concurrent resyncs, but `resubscribe()` positions the new receiver at
-        // the current write-head of the broadcast ring buffer, so all pre-sent bootstrap
-        // signals (written in `ReplicationResyncer::new`) are invisible to it.  Every
-        // spawned task therefore blocked forever, which is why `resync start` reported
-        // "started" yet objects never moved.  Throttling at this level is also incorrect
-        // for broadcast channels (one send unblocks ALL receivers).  The inner
-        // per-object worker pool (mpsc channels, line ~877) already provides the right
-        // concurrency limit.
-        if cancellation_token.is_cancelled() {
-            return;
-        }
-
-        // Acquire a cluster-wide leader lock for this (bucket, ARN) pair so that only
-        // one node runs the resync scan at a time. Without this, every cluster node would
-        // scan and replicate every object independently, causing N-fold duplicate traffic.
+    /// Acquire a cluster-wide leader lock for this (bucket, ARN) pair so that only
+    /// one node runs the resync scan at a time. Without this, every cluster node would
+    /// scan and replicate every object independently, causing N-fold duplicate traffic.
+    async fn acquire_resync_leader_lock<S: ReplicationStorage>(
+        storage: &Arc<S>,
+        opts: &ResyncOpts,
+    ) -> Option<rustfs_lock::NamespaceLockGuard> {
         let resync_lock_key = ReplicationMetadataStore::resync_lock_key(&opts.bucket, &opts.arn);
         let resync_ns_lock = match storage
             .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), &resync_lock_key)
@@ -674,11 +657,11 @@ impl ReplicationResyncer {
                     reason = "leader_lock_create_failed",
                     "Failed to create resync leader lock — skipping resync"
                 );
-                return;
+                return None;
             }
         };
-        let _resync_leader_guard = match resync_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await {
-            Ok(g) => g,
+        match resync_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await {
+            Ok(g) => Some(g),
             Err(_) => {
                 debug!(
                     event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED,
@@ -689,14 +672,19 @@ impl ReplicationResyncer {
                     reason = "leader_lock_held_by_another_node",
                     "Another node is already running resync for this bucket/ARN — skipping"
                 );
-                return;
+                None
             }
-        };
-
-        let Some(_resync_admission_permit) = self.acquire_resync_admission(&cancellation_token).await else {
-            return;
-        };
+        }
+    }
 
+    /// Resolve and validate the replication config plus the single remote target
+    /// client this resync run replicates to, marking the resync failed (and
+    /// returning `None`) when any lookup or validation step does not hold.
+    async fn resolve_resync_target<S: ReplicationObjectIO>(
+        &self,
+        opts: &ResyncOpts,
+        storage: &Arc<S>,
+    ) -> Option<(ReplicationConfig, Arc<TargetClient>)> {
         let cfg = match get_replication_config(&opts.bucket).await {
             Ok(cfg) => cfg,
             Err(err) => {
@@ -712,7 +700,7 @@ impl ReplicationResyncer {
                 );
                 self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
                     .await;
-                return;
+                return None;
             }
         };
 
@@ -730,7 +718,7 @@ impl ReplicationResyncer {
                 );
                 self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
                     .await;
-                return;
+                return None;
             }
         };
 
@@ -748,7 +736,7 @@ impl ReplicationResyncer {
             );
             self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
                 .await;
-            return;
+            return None;
         }
 
         let target_arns = if let Some(cfg) = cfg {
@@ -773,7 +761,7 @@ impl ReplicationResyncer {
             );
             self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
                 .await;
-            return;
+            return None;
         }
 
         let Some(target_client) = ReplicationTargetStore::remote_target_client(&opts.bucket, &target_arns[0]).await else {
@@ -788,9 +776,15 @@ impl ReplicationResyncer {
             );
             self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
                 .await;
-            return;
+            return None;
         };
 
+        Some((rcfg, target_client))
+    }
+
+    /// Persist the `ResyncStarted` status for non-heal runs, logging (without
+    /// aborting the resync) when the status update fails.
+    async fn mark_resync_started<S: ReplicationObjectIO>(&self, heal: bool, opts: &ResyncOpts, storage: &Arc<S>) {
         if !heal
             && let Err(e) = self
                 .mark_status(ResyncStatusType::ResyncStarted, opts.clone(), storage.clone())
@@ -807,223 +801,152 @@ impl ReplicationResyncer {
                 "Failed to update resync status"
             );
         }
+    }
 
-        let (tx, mut rx) = tokio::sync::mpsc::channel(100);
-        let walk_failed = Arc::new(AtomicBool::new(false));
-        let walk_failed_task = walk_failed.clone();
-        let walk_storage = storage.clone();
-        let walk_cancellation = cancellation_token.clone();
-        let walk_bucket = opts.bucket.clone();
-        let walk_arn = opts.arn.clone();
-        let walk_task = tokio::spawn(async move {
-            if let Err(err) = walk_storage
-                .walk(
-                    walk_cancellation,
-                    &walk_bucket,
-                    "",
-                    tx,
-                    WalkOptions::default().with_walkdir_timeouts(BACKGROUND_WALKDIR_TIMEOUT),
-                )
-                .await
-            {
-                walk_failed_task.store(true, Ordering::Relaxed);
-                error!(
-                    event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                    bucket = %walk_bucket,
-                    arn = %walk_arn,
-                    reason = "walk_failed",
-                    error = %err,
-                    "Replication resync bucket walk failed"
-                );
-            }
-        });
+    /// Drain and join the resync worker tasks after a fatal dispatch error,
+    /// logging any observed task failure and persisting the failed status.
+    async fn finish_resync_failed<S: ReplicationObjectIO>(
+        &self,
+        worker_txs: Vec<tokio::sync::mpsc::Sender<ReplicateObjectInfo>>,
+        results_tx: tokio::sync::mpsc::Sender<TargetReplicationResyncStatus>,
+        futures: Vec<JoinHandle<()>>,
+        join_failure_reason: &str,
+        opts: &ResyncOpts,
+        storage: &Arc<S>,
+    ) {
+        let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await;
+        if worker_failed {
+            error!(
+                event = EVENT_RESYNC_TASK_FAILED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %opts.bucket,
+                arn = %opts.arn,
+                reason = join_failure_reason,
+                "Replication resync worker cleanup observed task failure"
+            );
+        }
+        self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
+            .await;
+    }
 
-        let mut worker_txs = Vec::new();
+    /// Abort the resync worker tasks after cancellation and persist the
+    /// canceled status.
+    async fn finish_resync_canceled<S: ReplicationObjectIO>(
+        &self,
+        worker_txs: Vec<tokio::sync::mpsc::Sender<ReplicateObjectInfo>>,
+        results_tx: tokio::sync::mpsc::Sender<TargetReplicationResyncStatus>,
+        futures: Vec<JoinHandle<()>>,
+        opts: &ResyncOpts,
+        storage: &Arc<S>,
+    ) {
+        finish_resync_workers(worker_txs, results_tx, futures, true).await;
+        self.resync_bucket_mark_status(ResyncStatusType::ResyncCanceled, opts.clone(), storage.clone())
+            .await;
+    }
+
+    /// Spawn the collector task that folds per-object resync results into the
+    /// aggregated resync stats.
+    fn spawn_resync_results_collector(
+        resyncer: Arc<Self>,
+        opts: &ResyncOpts,
+    ) -> (tokio::sync::mpsc::Sender<TargetReplicationResyncStatus>, JoinHandle<()>) {
         // mpsc, not broadcast: a lagging broadcast receiver returns Err(Lagged) which
         // would end the collector and silently drop every subsequent worker result.
         let (results_tx, mut results_rx) = tokio::sync::mpsc::channel::<TargetReplicationResyncStatus>(RESYNC_WORKER_COUNT * 4);
 
         let opts_clone = opts.clone();
-        let self_clone = self.clone();
-
-        let mut futures = vec![walk_task];
 
         let results_fut = tokio::spawn(async move {
             while let Some(st) = results_rx.recv().await {
-                self_clone.inc_stats(&st, opts_clone.clone()).await;
+                resyncer.inc_stats(&st, opts_clone.clone()).await;
             }
         });
 
+        (results_tx, results_fut)
+    }
+
+    #[instrument(skip(cancellation_token, storage))]
+    pub async fn resync_bucket<S: ReplicationStorage>(
+        self: Arc<Self>,
+        cancellation_token: CancellationToken,
+        storage: Arc<S>,
+        heal: bool,
+        opts: ResyncOpts,
+    ) {
+        // Check cancellation before starting the scan.
+        // NOTE: the previous design waited here on `worker_rx.resubscribe().recv()` to
+        // throttle concurrent resyncs, but `resubscribe()` positions the new receiver at
+        // the current write-head of the broadcast ring buffer, so all pre-sent bootstrap
+        // signals (written in `ReplicationResyncer::new`) are invisible to it.  Every
+        // spawned task therefore blocked forever, which is why `resync start` reported
+        // "started" yet objects never moved.  Throttling at this level is also incorrect
+        // for broadcast channels (one send unblocks ALL receivers).  The inner
+        // per-object worker pool (mpsc channels, `spawn_resync_object_workers`) already
+        // provides the right concurrency limit.
+        if cancellation_token.is_cancelled() {
+            return;
+        }
+
+        let Some(_resync_leader_guard) = Self::acquire_resync_leader_lock(&storage, &opts).await else {
+            return;
+        };
+
+        let Some(_resync_admission_permit) = self.acquire_resync_admission(&cancellation_token).await else {
+            return;
+        };
+
+        let Some((rcfg, target_client)) = self.resolve_resync_target(&opts, &storage).await else {
+            return;
+        };
+
+        self.mark_resync_started(heal, &opts, &storage).await;
+
+        let (rx, walk_failed, walk_task) = spawn_resync_walk_task(&storage, &cancellation_token, &opts);
+
+        let mut futures = vec![walk_task];
+
+        let (results_tx, results_fut) = Self::spawn_resync_results_collector(self.clone(), &opts);
+
         futures.push(results_fut);
 
-        for _ in 0..RESYNC_WORKER_COUNT {
-            let (tx, mut rx) = tokio::sync::mpsc::channel::<ReplicateObjectInfo>(100);
-            worker_txs.push(tx);
+        let worker_txs =
+            spawn_resync_object_workers(&cancellation_token, &target_client, &storage, &opts, &results_tx, &mut futures);
 
-            let cancel_token = cancellation_token.clone();
-            let target_client = target_client.clone();
-            let storage = storage.clone();
-            let results_tx = results_tx.clone();
-            let bucket_name = opts.bucket.clone();
-            let target_arn = opts.arn.clone();
+        self.drive_resync_dispatch(
+            &cancellation_token,
+            rx,
+            &rcfg,
+            ResyncRunState {
+                worker_txs,
+                results_tx,
+                futures,
+                walk_failed,
+            },
+            &opts,
+            &storage,
+        )
+        .await;
+    }
 
-            let f = tokio::spawn(async move {
-                while let Some(mut roi) = rx.recv().await {
-                    if cancel_token.is_cancelled() {
-                        return;
-                    }
-
-                    if roi.delete_marker || !roi.version_purge_status.is_empty() {
-                        let (version_id, dm_version_id) = if roi.version_purge_status.is_empty() {
-                            (None, roi.version_id)
-                        } else {
-                            (roi.version_id, None)
-                        };
-
-                        let doi = DeletedObjectReplicationInfo {
-                            delete_object: ReplicationDeletedObject {
-                                object_name: roi.name.clone(),
-                                delete_marker_version_id: dm_version_id,
-                                version_id,
-                                replication_state: roi.replication_state.clone(),
-                                delete_marker: roi.delete_marker,
-                                delete_marker_mtime: roi.mod_time,
-                                ..Default::default()
-                            },
-                            bucket: roi.bucket.clone(),
-                            event_type: REPLICATE_EXISTING_DELETE.to_string(),
-                            op_type: ReplicationType::ExistingObject,
-                            target_arn: target_arn.clone(),
-                            ..Default::default()
-                        };
-                        replicate_delete(doi, storage.clone()).await;
-                    } else {
-                        roi.op_type = ReplicationType::ExistingObject;
-                        roi.event_type = REPLICATE_EXISTING.to_string();
-                        replicate_object(roi.clone(), storage.clone()).await;
-                    }
-
-                    let mut st = TargetReplicationResyncStatus {
-                        object: roi.name.clone(),
-                        bucket: roi.bucket.clone(),
-                        ..Default::default()
-                    };
-
-                    let reset_id = target_client.reset_id.clone();
-
-                    let head_result = head_object_with_proxy_stats(
-                        &bucket_name,
-                        target_client.as_ref(),
-                        &target_client.bucket,
-                        &roi.name,
-                        roi.version_id.map(|v| v.to_string()),
-                    )
-                    .await;
-                    let (size, err) = match head_result {
-                        Ok(_) => {
-                            st.replicated_count += 1;
-                            st.replicated_size += roi.size;
-                            (roi.size, None)
-                        }
-                        Err(err) if roi.delete_marker => {
-                            // Verifying a replicated delete marker: only a
-                            // definitive 404/NoSuchKey or 405/MethodNotAllowed
-                            // confirms the marker propagated. Any other
-                            // (retryable/ambiguous) HEAD error leaves the outcome
-                            // unverified, so it must count as failed — not as a
-                            // blanket success (backlog#862 / #799 B13).
-                            let retryable = {
-                                let (is_not_found, code) = err
-                                    .as_service_error()
-                                    .map(|se| (se.is_not_found(), se.code()))
-                                    .unwrap_or((false, None));
-                                is_retryable_delete_replication_head_error(is_not_found, code)
-                            };
-                            if retryable {
-                                st.failed_count += 1;
-                                (0, Some(err))
-                            } else {
-                                st.replicated_count += 1;
-                                (0, None)
-                            }
-                        }
-                        Err(err) if is_version_id_format_mismatch(&err) => {
-                            // AWS-style target rejects the RustFS UUID versionId
-                            // (400). Re-verify without the versionId before
-                            // concluding the object failed to replicate, instead
-                            // of counting a well-replicated object as failed.
-                            match head_object_fallback(&bucket_name, target_client.as_ref(), &roi.name).await {
-                                Ok(Some(_)) => {
-                                    st.replicated_count += 1;
-                                    st.replicated_size += roi.size;
-                                    (roi.size, None)
-                                }
-                                Ok(None) => {
-                                    st.failed_count += 1;
-                                    (0, Some(err))
-                                }
-                                Err(e2) => {
-                                    st.failed_count += 1;
-                                    (0, Some(e2))
-                                }
-                            }
-                        }
-                        Err(err) => {
-                            st.failed_count += 1;
-                            (0, Some(err))
-                        }
-                    };
-
-                    if err.is_some() {
-                        debug!(
-                            event = EVENT_RESYNC_OBJECT_PROCESSED,
-                            component = LOG_COMPONENT_ECSTORE,
-                            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                            reset_id = %reset_id,
-                            bucket = %bucket_name,
-                            object = %roi.name,
-                            version_id = %roi.version_id.unwrap_or_default(),
-                            size,
-                            error = ?err,
-                            "Processed resync object with verification error"
-                        );
-                    } else {
-                        trace!(
-                            event = EVENT_RESYNC_OBJECT_PROCESSED,
-                            component = LOG_COMPONENT_ECSTORE,
-                            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                            reset_id = %reset_id,
-                            bucket = %bucket_name,
-                            object = %roi.name,
-                            version_id = %roi.version_id.unwrap_or_default(),
-                            size,
-                            "Processed resync object"
-                        );
-                    }
-                    st.error = err.as_ref().and_then(resync_target_error_detail);
-
-                    if cancel_token.is_cancelled() {
-                        return;
-                    }
-
-                    if let Err(err) = results_tx.send(st).await {
-                        error!(
-                            event = EVENT_RESYNC_RUNTIME_CHANNEL_FAILED,
-                            component = LOG_COMPONENT_ECSTORE,
-                            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                            bucket = %bucket_name,
-                            reason = "status_channel_send_failed",
-                            error = %err,
-                            "Failed to send resync status"
-                        );
-                    }
-                }
-            });
-
-            futures.push(f);
-        }
+    /// Pump walked objects through classification into the hashed worker
+    /// queues, finalizing the resync status on dispatch error, cancellation,
+    /// or completion of the walk.
+    async fn drive_resync_dispatch<S: ReplicationStorage>(
+        &self,
+        cancellation_token: &CancellationToken,
+        mut rx: tokio::sync::mpsc::Receiver<StorageObjectInfoOrErr<ObjectInfo, Error>>,
+        rcfg: &ReplicationConfig,
+        state: ResyncRunState,
+        opts: &ResyncOpts,
+        storage: &Arc<S>,
+    ) {
+        let ResyncRunState {
+            worker_txs,
+            results_tx,
+            futures,
+            walk_failed,
+        } = state;
 
         while let Some(res) = rx.recv().await {
             if let Some(err) = res.err {
@@ -1039,27 +962,21 @@ impl ReplicationResyncer {
                 );
                 cancellation_token.cancel();
                 drop(rx);
-                let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await;
-                if worker_failed {
-                    error!(
-                        event = EVENT_RESYNC_TASK_FAILED,
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                        bucket = %opts.bucket,
-                        arn = %opts.arn,
-                        reason = "worker_join_failed_after_object_info_error",
-                        "Replication resync worker cleanup observed task failure"
-                    );
-                }
-                self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
-                    .await;
+                self.finish_resync_failed(
+                    worker_txs,
+                    results_tx,
+                    futures,
+                    "worker_join_failed_after_object_info_error",
+                    opts,
+                    storage,
+                )
+                .await;
                 return;
             }
 
             if cancellation_token.is_cancelled() {
                 drop(rx);
-                finish_resync_workers(worker_txs, results_tx, futures, true).await;
-                self.resync_bucket_mark_status(ResyncStatusType::ResyncCanceled, opts.clone(), storage.clone())
+                self.finish_resync_canceled(worker_txs, results_tx, futures, opts, storage)
                     .await;
                 return;
             }
@@ -1068,7 +985,7 @@ impl ReplicationResyncer {
                 continue;
             };
 
-            let roi = match get_heal_replicate_object_info(&object, &rcfg).await {
+            let roi = match get_heal_replicate_object_info(&object, rcfg).await {
                 Ok(roi) => roi,
                 Err(err) => {
                     error!(
@@ -1083,20 +1000,15 @@ impl ReplicationResyncer {
                     );
                     cancellation_token.cancel();
                     drop(rx);
-                    let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await;
-                    if worker_failed {
-                        error!(
-                            event = EVENT_RESYNC_TASK_FAILED,
-                            component = LOG_COMPONENT_ECSTORE,
-                            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                            bucket = %opts.bucket,
-                            arn = %opts.arn,
-                            reason = "worker_join_failed_after_classification_error",
-                            "Replication resync worker cleanup observed task failure"
-                        );
-                    }
-                    self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
-                        .await;
+                    self.finish_resync_failed(
+                        worker_txs,
+                        results_tx,
+                        futures,
+                        "worker_join_failed_after_classification_error",
+                        opts,
+                        storage,
+                    )
+                    .await;
                     return;
                 }
             };
@@ -1106,8 +1018,7 @@ impl ReplicationResyncer {
 
             if cancellation_token.is_cancelled() {
                 drop(rx);
-                finish_resync_workers(worker_txs, results_tx, futures, true).await;
-                self.resync_bucket_mark_status(ResyncStatusType::ResyncCanceled, opts.clone(), storage.clone())
+                self.finish_resync_canceled(worker_txs, results_tx, futures, opts, storage)
                     .await;
                 return;
             }
@@ -1127,26 +1038,21 @@ impl ReplicationResyncer {
                 );
                 cancellation_token.cancel();
                 drop(rx);
-                let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await;
-                if worker_failed {
-                    error!(
-                        event = EVENT_RESYNC_TASK_FAILED,
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                        bucket = %opts.bucket,
-                        arn = %opts.arn,
-                        reason = "worker_join_failed_after_queue_send_error",
-                        "Replication resync worker cleanup observed task failure"
-                    );
-                }
-                self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
-                    .await;
+                self.finish_resync_failed(
+                    worker_txs,
+                    results_tx,
+                    futures,
+                    "worker_join_failed_after_queue_send_error",
+                    opts,
+                    storage,
+                )
+                .await;
                 return;
             }
         }
 
         let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await;
-        let target_failed = self.target_has_resync_failures(&opts).await;
+        let target_failed = self.target_has_resync_failures(opts).await;
         let status = if walk_failed.load(Ordering::Relaxed) || worker_failed || target_failed {
             ResyncStatusType::ResyncFailed
         } else {
@@ -1157,6 +1063,275 @@ impl ReplicationResyncer {
     }
 }
 
+/// Worker-pool channel and task state for one resync run, handed from setup to
+/// the dispatch loop.
+struct ResyncRunState {
+    worker_txs: Vec<tokio::sync::mpsc::Sender<ReplicateObjectInfo>>,
+    results_tx: tokio::sync::mpsc::Sender<TargetReplicationResyncStatus>,
+    futures: Vec<JoinHandle<()>>,
+    walk_failed: Arc<AtomicBool>,
+}
+
+/// Spawn the bucket walk task that feeds object listings into the resync
+/// dispatch loop, surfacing walk failures through the returned flag.
+fn spawn_resync_walk_task<S: ReplicationStorage>(
+    storage: &Arc<S>,
+    cancellation_token: &CancellationToken,
+    opts: &ResyncOpts,
+) -> (
+    tokio::sync::mpsc::Receiver<StorageObjectInfoOrErr<ObjectInfo, Error>>,
+    Arc<AtomicBool>,
+    JoinHandle<()>,
+) {
+    let (tx, rx) = tokio::sync::mpsc::channel(100);
+    let walk_failed = Arc::new(AtomicBool::new(false));
+    let walk_failed_task = walk_failed.clone();
+    let walk_storage = storage.clone();
+    let walk_cancellation = cancellation_token.clone();
+    let walk_bucket = opts.bucket.clone();
+    let walk_arn = opts.arn.clone();
+    let walk_task = tokio::spawn(async move {
+        if let Err(err) = walk_storage
+            .walk(
+                walk_cancellation,
+                &walk_bucket,
+                "",
+                tx,
+                WalkOptions::default().with_walkdir_timeouts(BACKGROUND_WALKDIR_TIMEOUT),
+            )
+            .await
+        {
+            walk_failed_task.store(true, Ordering::Relaxed);
+            error!(
+                event = EVENT_RESYNC_RUNTIME_SKIPPED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %walk_bucket,
+                arn = %walk_arn,
+                reason = "walk_failed",
+                error = %err,
+                "Replication resync bucket walk failed"
+            );
+        }
+    });
+    (rx, walk_failed, walk_task)
+}
+
+/// Build the delete-replication work item for an existing delete marker or
+/// version purge discovered during a resync scan.
+fn resync_existing_delete_replication_info(roi: &ReplicateObjectInfo, target_arn: &str) -> DeletedObjectReplicationInfo {
+    let (version_id, dm_version_id) = if roi.version_purge_status.is_empty() {
+        (None, roi.version_id)
+    } else {
+        (roi.version_id, None)
+    };
+
+    DeletedObjectReplicationInfo {
+        delete_object: ReplicationDeletedObject {
+            object_name: roi.name.clone(),
+            delete_marker_version_id: dm_version_id,
+            version_id,
+            replication_state: roi.replication_state.clone(),
+            delete_marker: roi.delete_marker,
+            delete_marker_mtime: roi.mod_time,
+            ..Default::default()
+        },
+        bucket: roi.bucket.clone(),
+        event_type: REPLICATE_EXISTING_DELETE.to_string(),
+        op_type: ReplicationType::ExistingObject,
+        target_arn: target_arn.to_string(),
+        ..Default::default()
+    }
+}
+
+/// Classify the target HEAD verification result for one resynced object,
+/// updating the per-object status counters and returning the accounted size
+/// together with any verification error.
+async fn verify_resync_head_result(
+    head_result: std::result::Result<HeadObjectOutput, SdkError<HeadObjectError>>,
+    roi: &ReplicateObjectInfo,
+    st: &mut TargetReplicationResyncStatus,
+    target_client: &Arc<TargetClient>,
+) -> (i64, Option<SdkError<HeadObjectError>>) {
+    match head_result {
+        Ok(_) => {
+            st.replicated_count += 1;
+            st.replicated_size += roi.size;
+            (roi.size, None)
+        }
+        Err(err) if roi.delete_marker => {
+            // Verifying a replicated delete marker: only a
+            // definitive 404/NoSuchKey or 405/MethodNotAllowed
+            // confirms the marker propagated. Any other
+            // (retryable/ambiguous) HEAD error leaves the outcome
+            // unverified, so it must count as failed — not as a
+            // blanket success (backlog#862 / #799 B13).
+            let retryable = {
+                let (is_not_found, code) = err
+                    .as_service_error()
+                    .map(|se| (se.is_not_found(), se.code()))
+                    .unwrap_or((false, None));
+                is_retryable_delete_replication_head_error(is_not_found, code)
+            };
+            if retryable {
+                st.failed_count += 1;
+                (0, Some(err))
+            } else {
+                st.replicated_count += 1;
+                (0, None)
+            }
+        }
+        Err(err) if is_version_id_format_mismatch(&err) => {
+            // AWS-style target rejects the RustFS UUID versionId
+            // (400). Re-verify without the versionId before
+            // concluding the object failed to replicate, instead
+            // of counting a well-replicated object as failed.
+            match head_object_fallback(&roi.bucket, target_client.as_ref(), &roi.name).await {
+                Ok(Some(_)) => {
+                    st.replicated_count += 1;
+                    st.replicated_size += roi.size;
+                    (roi.size, None)
+                }
+                Ok(None) => {
+                    st.failed_count += 1;
+                    (0, Some(err))
+                }
+                Err(e2) => {
+                    st.failed_count += 1;
+                    (0, Some(e2))
+                }
+            }
+        }
+        Err(err) => {
+            st.failed_count += 1;
+            (0, Some(err))
+        }
+    }
+}
+
+/// Replicate one existing object (or delete marker / version purge) to the
+/// resync target, verify the outcome via a target HEAD, and produce the
+/// per-object resync status update.
+async fn resync_worker_process_object<S: ReplicationStorage>(
+    mut roi: ReplicateObjectInfo,
+    storage: &Arc<S>,
+    target_client: &Arc<TargetClient>,
+    bucket_name: &str,
+    target_arn: &str,
+) -> TargetReplicationResyncStatus {
+    if roi.delete_marker || !roi.version_purge_status.is_empty() {
+        let doi = resync_existing_delete_replication_info(&roi, target_arn);
+        replicate_delete(doi, storage.clone()).await;
+    } else {
+        roi.op_type = ReplicationType::ExistingObject;
+        roi.event_type = REPLICATE_EXISTING.to_string();
+        replicate_object(roi.clone(), storage.clone()).await;
+    }
+
+    let mut st = TargetReplicationResyncStatus {
+        object: roi.name.clone(),
+        bucket: roi.bucket.clone(),
+        ..Default::default()
+    };
+
+    let reset_id = target_client.reset_id.clone();
+
+    let head_result = head_object_with_proxy_stats(
+        bucket_name,
+        target_client.as_ref(),
+        &target_client.bucket,
+        &roi.name,
+        roi.version_id.map(|v| v.to_string()),
+    )
+    .await;
+    let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, target_client).await;
+
+    if err.is_some() {
+        debug!(
+            event = EVENT_RESYNC_OBJECT_PROCESSED,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+            reset_id = %reset_id,
+            bucket = %bucket_name,
+            object = %roi.name,
+            version_id = %roi.version_id.unwrap_or_default(),
+            size,
+            error = ?err,
+            "Processed resync object with verification error"
+        );
+    } else {
+        trace!(
+            event = EVENT_RESYNC_OBJECT_PROCESSED,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+            reset_id = %reset_id,
+            bucket = %bucket_name,
+            object = %roi.name,
+            version_id = %roi.version_id.unwrap_or_default(),
+            size,
+            "Processed resync object"
+        );
+    }
+    st.error = err.as_ref().and_then(resync_target_error_detail);
+
+    st
+}
+
+/// Spawn the per-object resync worker pool, wiring every worker to the shared
+/// results channel and registering its task handle for cleanup.
+fn spawn_resync_object_workers<S: ReplicationStorage>(
+    cancellation_token: &CancellationToken,
+    target_client: &Arc<TargetClient>,
+    storage: &Arc<S>,
+    opts: &ResyncOpts,
+    results_tx: &tokio::sync::mpsc::Sender<TargetReplicationResyncStatus>,
+    futures: &mut Vec<JoinHandle<()>>,
+) -> Vec<tokio::sync::mpsc::Sender<ReplicateObjectInfo>> {
+    let mut worker_txs = Vec::new();
+
+    for _ in 0..RESYNC_WORKER_COUNT {
+        let (tx, mut rx) = tokio::sync::mpsc::channel::<ReplicateObjectInfo>(100);
+        worker_txs.push(tx);
+
+        let cancel_token = cancellation_token.clone();
+        let target_client = target_client.clone();
+        let storage = storage.clone();
+        let results_tx = results_tx.clone();
+        let bucket_name = opts.bucket.clone();
+        let target_arn = opts.arn.clone();
+
+        let f = tokio::spawn(async move {
+            while let Some(roi) = rx.recv().await {
+                if cancel_token.is_cancelled() {
+                    return;
+                }
+
+                let st = resync_worker_process_object(roi, &storage, &target_client, &bucket_name, &target_arn).await;
+
+                if cancel_token.is_cancelled() {
+                    return;
+                }
+
+                if let Err(err) = results_tx.send(st).await {
+                    error!(
+                        event = EVENT_RESYNC_RUNTIME_CHANNEL_FAILED,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                        bucket = %bucket_name,
+                        reason = "status_channel_send_failed",
+                        error = %err,
+                        "Failed to send resync status"
+                    );
+                }
+            }
+        });
+
+        futures.push(f);
+    }
+
+    worker_txs
+}
+
 pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationConfig) -> Result<ReplicateObjectInfo> {
     let mut oi = oi.clone();
     let mut user_defined = (*oi.user_defined).clone();
@@ -2984,80 +3159,25 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
         let bucket = self.bucket.clone();
         let object = self.name.clone();
 
-        let mut replication_action = ReplicationAction::Metadata;
-        let mut rinfo = ReplicatedTargetInfo {
-            arn: tgt_client.arn.clone(),
-            size: self.actual_size,
-            replication_action,
-            op_type: self.op_type,
-            replication_status: ReplicationStatusType::Failed,
-            prev_replication_status: self.target_replication_status(&tgt_client.arn),
-            endpoint: tgt_client.endpoint.clone(),
-            secure: tgt_client.secure,
-            ..Default::default()
-        };
+        let mut rinfo = replicate_all_target_info(self, &tgt_client);
 
         if ReplicationTargetStore::target_is_offline(&tgt_client).await {
-            debug!(
-                event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                bucket = %bucket,
-                arn = %tgt_client.arn,
-                target = %tgt_client.to_url(),
-                reason = "target_offline",
-                "Skipped replication because target is offline"
-            );
-            send_local_event(EventArgs {
-                event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                bucket_name: bucket.clone(),
-                object: self.to_object_info(),
-                user_agent: "Internal: [Replication]".to_string(),
-                ..Default::default()
-            });
+            note_replicate_all_target_offline(self, &bucket, &tgt_client);
             return rinfo;
         }
 
         let versioned = ReplicationVersioningStore::prefix_enabled(&bucket, &object).await;
         let version_suspended = ReplicationVersioningStore::prefix_suspended(&bucket, &object).await;
 
-        let obj_opts = ObjectOptions {
-            version_id: self.version_id.map(|v| v.to_string()),
-            version_suspended,
-            versioned,
-            replication_request: true,
-            // SSE-C passthrough reads the stored ciphertext verbatim; the
-            // decrypting reader cannot serve it (no customer key server-side).
-            raw_data_movement_read: self.ssec,
-            ..Default::default()
-        };
+        let obj_opts = replicate_all_read_options(self, versioned, version_suspended);
 
-        let mut gr = match storage
+        let gr = match storage
             .get_object_reader(&bucket, &object, None, HeaderMap::new(), &obj_opts)
             .await
         {
             Ok(gr) => gr,
             Err(e) => {
-                if !(is_err_object_not_found(&e) || is_err_version_not_found(&e)) {
-                    debug!(
-                        event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                        bucket = %bucket,
-                        arn = %tgt_client.arn,
-                        error = %e,
-                        reason = "object_reader_unavailable",
-                        "Skipped replication because object reader is unavailable"
-                    );
-                    send_local_event(EventArgs {
-                        event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                        bucket_name: bucket.clone(),
-                        object: self.to_object_info(),
-                        user_agent: "Internal: [Replication]".to_string(),
-                        ..Default::default()
-                    });
-                }
-
+                note_replicate_all_reader_unavailable(self, &bucket, &tgt_client, &e);
                 return rinfo;
             }
         };
@@ -3069,23 +3189,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
         let size = match object_info.get_actual_size() {
             Ok(size) => size,
             Err(e) => {
-                debug!(
-                    event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                    bucket = %bucket,
-                    arn = %tgt_client.arn,
-                    error = %e,
-                    reason = "actual_size_unavailable",
-                    "Skipped replication because actual object size is unavailable"
-                );
-                send_local_event(EventArgs {
-                    event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                    bucket_name: bucket.clone(),
-                    object: object_info,
-                    user_agent: "Internal: [Replication]".to_string(),
-                    ..Default::default()
-                });
+                note_replicate_all_size_unavailable(&bucket, &tgt_client, object_info, &e);
                 return rinfo;
             }
         };
@@ -3094,174 +3198,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
         let transfer_size = if self.ssec { object_info.size } else { size };
 
         if tgt_client.bucket.is_empty() {
-            debug!(
-                event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                bucket = %bucket,
-                arn = %tgt_client.arn,
-                reason = "target_bucket_empty",
-                "Skipped replication because target bucket is empty"
-            );
-            send_local_event(EventArgs {
-                event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                bucket_name: bucket.clone(),
-                object: object_info,
-                user_agent: "Internal: [Replication]".to_string(),
-                ..Default::default()
-            });
+            note_replicate_all_target_bucket_empty(&bucket, &tgt_client, object_info);
             return rinfo;
         }
 
-        let mut sopts = StatObjectOptions {
-            version_id: object_info.version_id.map(|v| v.to_string()).unwrap_or_default(),
-            internal: AdvancedGetOptions {
-                replication_proxy_request: "false".to_string(),
-                ..Default::default()
-            },
-            ..Default::default()
-        };
+        let _sopts = replicate_all_stat_options(&object_info, &bucket, &tgt_client);
 
-        if let Err(err) = sopts.set(AMZ_TAGGING_DIRECTIVE, "ACCESS") {
-            debug!(
-                event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                bucket = %bucket,
-                arn = %tgt_client.arn,
-                error = %err,
-                reason = "tagging_directive_header_invalid",
-                "Skipped replication tagging directive header detail"
-            );
-        }
-
-        match head_object_with_proxy_stats(
-            &bucket,
-            tgt_client.as_ref(),
-            &tgt_client.bucket,
-            &object,
-            self.version_id.map(|v| v.to_string()),
-        )
-        .await
-        {
-            Ok(oi) => {
-                replication_action = replication_action_for_target_head(&object_info, &oi, self.op_type);
-                rinfo.replication_status = ReplicationStatusType::Completed;
-                if replication_action == ReplicationAction::None {
-                    if self.op_type == ReplicationType::ExistingObject
-                        && replication_target_head_is_newer_null_version(&object_info, &oi)
-                    {
-                        warn!(
-                            event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                            component = LOG_COMPONENT_ECSTORE,
-                            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                            bucket = %bucket,
-                            object = %object,
-                            arn = %tgt_client.arn,
-                            endpoint = %tgt_client.to_url(),
-                            reason = "target_newer_than_source_null_version",
-                            "Skipping replication because newer target version exists"
-                        );
-                        send_local_event(EventArgs {
-                            event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                            bucket_name: bucket.clone(),
-                            object: object_info.clone(),
-                            user_agent: "Internal: [Replication]".to_string(),
-                            ..Default::default()
-                        });
-                    }
-
-                    if object_info.target_replication_status(&tgt_client.arn) == ReplicationStatusType::Pending
-                        || object_info.target_replication_status(&tgt_client.arn) == ReplicationStatusType::Failed
-                        || self.op_type == ReplicationType::ExistingObject
-                    {
-                        rinfo.replication_action = replication_action;
-                        rinfo.replication_status = ReplicationStatusType::Completed;
-                    }
-
-                    if rinfo.replication_status == ReplicationStatusType::Completed
-                        && self.op_type == ReplicationType::ExistingObject
-                        && !tgt_client.reset_id.is_empty()
-                    {
-                        rinfo.resync_timestamp = format!(
-                            "{};{}",
-                            OffsetDateTime::now_utc()
-                                .format(&Rfc3339)
-                                .unwrap_or_else(|_| "invalid-time".to_string()),
-                            tgt_client.reset_id
-                        );
-                        rinfo.replication_resynced = true;
-                    }
-
-                    rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
-
-                    return rinfo;
-                }
-            }
-            Err(e) => {
-                if is_version_id_format_mismatch(&e) {
-                    // Version-ID format mismatch: retry without versionId and compare ETags.
-                    match head_object_fallback(&bucket, &tgt_client, &object).await {
-                        Ok(Some(oi)) => {
-                            replication_action = if replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()) {
-                                ReplicationAction::None
-                            } else {
-                                ReplicationAction::All
-                            };
-                        }
-                        Ok(None) => {
-                            replication_action = ReplicationAction::All;
-                        }
-                        Err(e2) => {
-                            rinfo.error = Some(e2.to_string());
-                            debug!(
-                                event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                                component = LOG_COMPONENT_ECSTORE,
-                                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                                bucket = %bucket,
-                                arn = %tgt_client.arn,
-                                error = %e2,
-                                reason = "head_object_fallback_failed",
-                                "Failed replication head-object fallback"
-                            );
-                            send_local_event(EventArgs {
-                                event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                                bucket_name: bucket.clone(),
-                                object: object_info,
-                                user_agent: "Internal: [Replication]".to_string(),
-                                ..Default::default()
-                            });
-                            rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
-                            return rinfo;
-                        }
-                    }
-                } else if e.as_service_error().is_some_and(|se| se.is_not_found()) {
-                    replication_action = ReplicationAction::All;
-                } else {
-                    rinfo.error = Some(e.to_string());
-                    debug!(
-                        event = EVENT_RESYNC_RUNTIME_SKIPPED,
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                        bucket = %bucket,
-                        arn = %tgt_client.arn,
-                        error = %e,
-                        reason = "head_object_failed",
-                        "Skipped replication because head-object failed"
-                    );
-
-                    send_local_event(EventArgs {
-                        event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                        bucket_name: bucket.clone(),
-                        object: object_info,
-                        user_agent: "Internal: [Replication]".to_string(),
-                        ..Default::default()
-                    });
-
-                    rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
-                    return rinfo;
-                }
-            }
+        let Some((replication_action, object_info)) =
+            resolve_replicate_all_action(self, &tgt_client, &bucket, &object, object_info, start_time, &mut rinfo).await
+        else {
+            return rinfo;
         };
 
         rinfo.replication_status = ReplicationStatusType::Completed;
@@ -3276,14 +3222,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
             // AWS-style targets permanently FAILED and never converging
             // (backlog#860 / #799 B11).
             if self.op_type == ReplicationType::ExistingObject && !tgt_client.reset_id.is_empty() {
-                rinfo.resync_timestamp = format!(
-                    "{};{}",
-                    OffsetDateTime::now_utc()
-                        .format(&Rfc3339)
-                        .unwrap_or_else(|_| "invalid-time".to_string()),
-                    tgt_client.reset_id
-                );
-                rinfo.replication_resynced = true;
+                apply_replication_resync_timestamp(&mut rinfo, &tgt_client.reset_id);
             }
             rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
             return rinfo;
@@ -3295,91 +3234,29 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
         let (put_opts, is_multipart) = match replication_put_object_options(&tgt_client.storage_class, &object_info) {
             Ok((put_opts, is_mp)) => (put_opts, is_mp),
             Err(e) => {
-                // Unsupported source metadata (e.g. managed SSE) is a fail-closed
-                // condition: report FAILED so the composite status and the
-                // OperationFailedReplication event reflect that nothing reached
-                // the target, instead of leaking the optimistic Completed above.
-                rinfo.replication_status = ReplicationStatusType::Failed;
-                rinfo.error = Some(e.to_string());
-                warn!(
-                    event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                    bucket = %bucket,
-                    arn = %tgt_client.arn,
-                    operation = "build_put_options",
-                    error = %e,
-                    "Replication target operation failed"
-                );
-                send_local_event(EventArgs {
-                    event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                    bucket_name: bucket.clone(),
-                    object: object_info,
-                    user_agent: "Internal: [Replication]".to_string(),
-                    ..Default::default()
-                });
-
-                rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
+                fail_replicate_all_put_options(&mut rinfo, &tgt_client, &bucket, object_info, &e, start_time);
                 return rinfo;
             }
         };
 
-        let has_tagging_replication = !put_opts.user_tags.is_empty();
-        if let Some(err) = if is_multipart {
-            drop(gr);
-            let result = replicate_object_with_multipart(MultipartReplicationContext {
-                storage: storage.clone(),
-                cli: tgt_client.clone(),
-                src_bucket: &bucket,
-                dst_bucket: &tgt_client.bucket,
+        if let Some(err) = replicate_all_payload_to_target(
+            ReplicateAllPayloadContext {
+                storage: &storage,
+                tgt_client: &tgt_client,
+                bucket: &bucket,
                 object: &object,
                 object_info: &object_info,
                 obj_opts: &obj_opts,
                 arn: &rinfo.arn,
+                transfer_size,
+                is_multipart,
                 put_opts,
-            })
-            .await;
-            record_proxy_request(&bucket, "PutObject", result.is_err()).await;
-            if has_tagging_replication {
-                record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await;
-            }
-            result.err()
-        } else {
-            gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn);
-            let byte_stream = async_read_to_bytestream(gr.stream);
-            let result = tgt_client
-                .put_object(&tgt_client.bucket, &object, transfer_size, byte_stream, &put_opts)
-                .await
-                .map(|assigned_version_id| {
-                    audit_target_version_identity(
-                        &tgt_client,
-                        &put_opts.internal.source_version_id,
-                        assigned_version_id.as_deref(),
-                    )
-                })
-                .map_err(|e| std::io::Error::other(e.to_string()));
-            record_proxy_request(&bucket, "PutObject", result.is_err()).await;
-            if has_tagging_replication {
-                record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await;
-            }
-            result.err()
-        } {
-            rinfo.replication_status = ReplicationStatusType::Failed;
-            rinfo.error = Some(err.to_string());
-            warn!(
-                event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                bucket = %bucket,
-                arn = %tgt_client.arn,
-                object = %object,
-                operation = "put_object",
-                error = ?err,
-                "Replication target operation failed"
-            );
-            rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
-
-            mark_replication_target_offline_if_needed(&tgt_client, &err).await;
+            },
+            gr,
+        )
+        .await
+        {
+            fail_replicate_all_put_object(&mut rinfo, &tgt_client, &bucket, &object, &err, start_time).await;
             return rinfo;
         }
 
@@ -3406,6 +3283,434 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
     }
 }
 
+/// Build the initial replication outcome DTO for `replicate_all`, seeded with
+/// the metadata-only action and a failed status until the target confirms
+/// otherwise.
+fn replicate_all_target_info(roi: &ReplicateObjectInfo, tgt_client: &TargetClient) -> ReplicatedTargetInfo {
+    ReplicatedTargetInfo {
+        arn: tgt_client.arn.clone(),
+        size: roi.actual_size,
+        replication_action: ReplicationAction::Metadata,
+        op_type: roi.op_type,
+        replication_status: ReplicationStatusType::Failed,
+        prev_replication_status: roi.target_replication_status(&tgt_client.arn),
+        endpoint: tgt_client.endpoint.clone(),
+        secure: tgt_client.secure,
+        ..Default::default()
+    }
+}
+
+/// Log and notify that replication was skipped because the target is offline.
+fn note_replicate_all_target_offline(roi: &ReplicateObjectInfo, bucket: &str, tgt_client: &TargetClient) {
+    debug!(
+        event = EVENT_RESYNC_RUNTIME_SKIPPED,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+        bucket = %bucket,
+        arn = %tgt_client.arn,
+        target = %tgt_client.to_url(),
+        reason = "target_offline",
+        "Skipped replication because target is offline"
+    );
+    send_local_event(EventArgs {
+        event_name: EventName::ObjectReplicationNotTracked.to_string(),
+        bucket_name: bucket.to_string(),
+        object: roi.to_object_info(),
+        user_agent: "Internal: [Replication]".to_string(),
+        ..Default::default()
+    });
+}
+
+/// Build the source-side read options for `replicate_all`.
+fn replicate_all_read_options(roi: &ReplicateObjectInfo, versioned: bool, version_suspended: bool) -> ObjectOptions {
+    ObjectOptions {
+        version_id: roi.version_id.map(|v| v.to_string()),
+        version_suspended,
+        versioned,
+        replication_request: true,
+        // SSE-C passthrough reads the stored ciphertext verbatim; the
+        // decrypting reader cannot serve it (no customer key server-side).
+        raw_data_movement_read: roi.ssec,
+        ..Default::default()
+    }
+}
+
+/// Log and notify that replication was skipped because the source object
+/// reader is unavailable; missing objects/versions stay silent.
+fn note_replicate_all_reader_unavailable(roi: &ReplicateObjectInfo, bucket: &str, tgt_client: &TargetClient, e: &Error) {
+    if !(is_err_object_not_found(e) || is_err_version_not_found(e)) {
+        debug!(
+            event = EVENT_RESYNC_RUNTIME_SKIPPED,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+            bucket = %bucket,
+            arn = %tgt_client.arn,
+            error = %e,
+            reason = "object_reader_unavailable",
+            "Skipped replication because object reader is unavailable"
+        );
+        send_local_event(EventArgs {
+            event_name: EventName::ObjectReplicationNotTracked.to_string(),
+            bucket_name: bucket.to_string(),
+            object: roi.to_object_info(),
+            user_agent: "Internal: [Replication]".to_string(),
+            ..Default::default()
+        });
+    }
+}
+
+/// Log and notify that replication was skipped because the actual object size
+/// is unavailable.
+fn note_replicate_all_size_unavailable(bucket: &str, tgt_client: &TargetClient, object_info: ObjectInfo, e: &std::io::Error) {
+    debug!(
+        event = EVENT_RESYNC_RUNTIME_SKIPPED,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+        bucket = %bucket,
+        arn = %tgt_client.arn,
+        error = %e,
+        reason = "actual_size_unavailable",
+        "Skipped replication because actual object size is unavailable"
+    );
+    send_local_event(EventArgs {
+        event_name: EventName::ObjectReplicationNotTracked.to_string(),
+        bucket_name: bucket.to_string(),
+        object: object_info,
+        user_agent: "Internal: [Replication]".to_string(),
+        ..Default::default()
+    });
+}
+
+/// Log and notify that replication was skipped because the target bucket is
+/// empty.
+fn note_replicate_all_target_bucket_empty(bucket: &str, tgt_client: &TargetClient, object_info: ObjectInfo) {
+    debug!(
+        event = EVENT_RESYNC_RUNTIME_SKIPPED,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+        bucket = %bucket,
+        arn = %tgt_client.arn,
+        reason = "target_bucket_empty",
+        "Skipped replication because target bucket is empty"
+    );
+    send_local_event(EventArgs {
+        event_name: EventName::ObjectReplicationNotTracked.to_string(),
+        bucket_name: bucket.to_string(),
+        object: object_info,
+        user_agent: "Internal: [Replication]".to_string(),
+        ..Default::default()
+    });
+}
+
+/// Build the stat options for the target metadata comparison, logging (without
+/// failing) when the tagging directive header cannot be set.
+fn replicate_all_stat_options(object_info: &ObjectInfo, bucket: &str, tgt_client: &TargetClient) -> StatObjectOptions {
+    let mut sopts = StatObjectOptions {
+        version_id: object_info.version_id.map(|v| v.to_string()).unwrap_or_default(),
+        internal: AdvancedGetOptions {
+            replication_proxy_request: "false".to_string(),
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+
+    if let Err(err) = sopts.set(AMZ_TAGGING_DIRECTIVE, "ACCESS") {
+        debug!(
+            event = EVENT_RESYNC_RUNTIME_SKIPPED,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+            bucket = %bucket,
+            arn = %tgt_client.arn,
+            error = %err,
+            reason = "tagging_directive_header_invalid",
+            "Skipped replication tagging directive header detail"
+        );
+    }
+
+    sopts
+}
+
+/// Record a failed payload transfer: mark the outcome FAILED, log the target
+/// operation failure, and take the target offline when the error is a network
+/// failure.
+async fn fail_replicate_all_put_object(
+    rinfo: &mut ReplicatedTargetInfo,
+    tgt_client: &Arc<TargetClient>,
+    bucket: &str,
+    object: &str,
+    err: &std::io::Error,
+    start_time: OffsetDateTime,
+) {
+    rinfo.replication_status = ReplicationStatusType::Failed;
+    rinfo.error = Some(err.to_string());
+    warn!(
+        event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+        bucket = %bucket,
+        arn = %tgt_client.arn,
+        object = %object,
+        operation = "put_object",
+        error = ?err,
+        "Replication target operation failed"
+    );
+    rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
+
+    mark_replication_target_offline_if_needed(tgt_client, err).await;
+}
+
+/// Stamp the replication outcome as resynced against the target's current
+/// reset id.
+fn apply_replication_resync_timestamp(rinfo: &mut ReplicatedTargetInfo, reset_id: &str) {
+    rinfo.resync_timestamp = format!(
+        "{};{}",
+        OffsetDateTime::now_utc()
+            .format(&Rfc3339)
+            .unwrap_or_else(|_| "invalid-time".to_string()),
+        reset_id
+    );
+    rinfo.replication_resynced = true;
+}
+
+/// Compare the source object against the target via HEAD and decide which
+/// replication action is still required. Returns `None` after fully settling
+/// `rinfo` when replication must stop here — either because the target already
+/// matches or because the comparison failed.
+async fn resolve_replicate_all_action(
+    roi: &ReplicateObjectInfo,
+    tgt_client: &Arc<TargetClient>,
+    bucket: &str,
+    object: &str,
+    object_info: ObjectInfo,
+    start_time: OffsetDateTime,
+    rinfo: &mut ReplicatedTargetInfo,
+) -> Option<(ReplicationAction, ObjectInfo)> {
+    let replication_action;
+    match head_object_with_proxy_stats(
+        bucket,
+        tgt_client.as_ref(),
+        &tgt_client.bucket,
+        object,
+        roi.version_id.map(|v| v.to_string()),
+    )
+    .await
+    {
+        Ok(oi) => {
+            replication_action = replication_action_for_target_head(&object_info, &oi, roi.op_type);
+            rinfo.replication_status = ReplicationStatusType::Completed;
+            if replication_action == ReplicationAction::None {
+                if roi.op_type == ReplicationType::ExistingObject
+                    && replication_target_head_is_newer_null_version(&object_info, &oi)
+                {
+                    warn!(
+                        event = EVENT_RESYNC_RUNTIME_SKIPPED,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                        bucket = %bucket,
+                        object = %object,
+                        arn = %tgt_client.arn,
+                        endpoint = %tgt_client.to_url(),
+                        reason = "target_newer_than_source_null_version",
+                        "Skipping replication because newer target version exists"
+                    );
+                    send_local_event(EventArgs {
+                        event_name: EventName::ObjectReplicationNotTracked.to_string(),
+                        bucket_name: bucket.to_string(),
+                        object: object_info.clone(),
+                        user_agent: "Internal: [Replication]".to_string(),
+                        ..Default::default()
+                    });
+                }
+
+                if object_info.target_replication_status(&tgt_client.arn) == ReplicationStatusType::Pending
+                    || object_info.target_replication_status(&tgt_client.arn) == ReplicationStatusType::Failed
+                    || roi.op_type == ReplicationType::ExistingObject
+                {
+                    rinfo.replication_action = replication_action;
+                    rinfo.replication_status = ReplicationStatusType::Completed;
+                }
+
+                if rinfo.replication_status == ReplicationStatusType::Completed
+                    && roi.op_type == ReplicationType::ExistingObject
+                    && !tgt_client.reset_id.is_empty()
+                {
+                    apply_replication_resync_timestamp(rinfo, &tgt_client.reset_id);
+                }
+
+                rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
+
+                return None;
+            }
+        }
+        Err(e) => {
+            if is_version_id_format_mismatch(&e) {
+                // Version-ID format mismatch: retry without versionId and compare ETags.
+                match head_object_fallback(bucket, tgt_client, object).await {
+                    Ok(Some(oi)) => {
+                        replication_action = if replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()) {
+                            ReplicationAction::None
+                        } else {
+                            ReplicationAction::All
+                        };
+                    }
+                    Ok(None) => {
+                        replication_action = ReplicationAction::All;
+                    }
+                    Err(e2) => {
+                        rinfo.error = Some(e2.to_string());
+                        debug!(
+                            event = EVENT_RESYNC_RUNTIME_SKIPPED,
+                            component = LOG_COMPONENT_ECSTORE,
+                            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                            bucket = %bucket,
+                            arn = %tgt_client.arn,
+                            error = %e2,
+                            reason = "head_object_fallback_failed",
+                            "Failed replication head-object fallback"
+                        );
+                        send_local_event(EventArgs {
+                            event_name: EventName::ObjectReplicationNotTracked.to_string(),
+                            bucket_name: bucket.to_string(),
+                            object: object_info,
+                            user_agent: "Internal: [Replication]".to_string(),
+                            ..Default::default()
+                        });
+                        rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
+                        return None;
+                    }
+                }
+            } else if e.as_service_error().is_some_and(|se| se.is_not_found()) {
+                replication_action = ReplicationAction::All;
+            } else {
+                rinfo.error = Some(e.to_string());
+                debug!(
+                    event = EVENT_RESYNC_RUNTIME_SKIPPED,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                    bucket = %bucket,
+                    arn = %tgt_client.arn,
+                    error = %e,
+                    reason = "head_object_failed",
+                    "Skipped replication because head-object failed"
+                );
+
+                send_local_event(EventArgs {
+                    event_name: EventName::ObjectReplicationNotTracked.to_string(),
+                    bucket_name: bucket.to_string(),
+                    object: object_info,
+                    user_agent: "Internal: [Replication]".to_string(),
+                    ..Default::default()
+                });
+
+                rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
+                return None;
+            }
+        }
+    };
+
+    Some((replication_action, object_info))
+}
+
+/// Record a fail-closed put-options failure.
+/// Unsupported source metadata (e.g. managed SSE) is a fail-closed
+/// condition: report FAILED so the composite status and the
+/// OperationFailedReplication event reflect that nothing reached
+/// the target, instead of leaking the optimistic Completed set earlier.
+fn fail_replicate_all_put_options(
+    rinfo: &mut ReplicatedTargetInfo,
+    tgt_client: &TargetClient,
+    bucket: &str,
+    object_info: ObjectInfo,
+    e: &Error,
+    start_time: OffsetDateTime,
+) {
+    rinfo.replication_status = ReplicationStatusType::Failed;
+    rinfo.error = Some(e.to_string());
+    warn!(
+        event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+        bucket = %bucket,
+        arn = %tgt_client.arn,
+        operation = "build_put_options",
+        error = %e,
+        "Replication target operation failed"
+    );
+    send_local_event(EventArgs {
+        event_name: EventName::ObjectReplicationNotTracked.to_string(),
+        bucket_name: bucket.to_string(),
+        object: object_info,
+        user_agent: "Internal: [Replication]".to_string(),
+        ..Default::default()
+    });
+
+    rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
+}
+
+/// Borrowed inputs shared by both transports of the `replicate_all` payload
+/// transfer step.
+struct ReplicateAllPayloadContext<'a, S: ReplicationObjectIO> {
+    storage: &'a Arc<S>,
+    tgt_client: &'a Arc<TargetClient>,
+    bucket: &'a str,
+    object: &'a str,
+    object_info: &'a ObjectInfo,
+    obj_opts: &'a ObjectOptions,
+    arn: &'a str,
+    transfer_size: i64,
+    is_multipart: bool,
+    put_opts: PutObjectOptions,
+}
+
+/// Ship the object payload to the replication target over the multipart or
+/// single-put transport, returning the transport error when the upload fails.
+async fn replicate_all_payload_to_target<S: ReplicationObjectIO>(
+    ctx: ReplicateAllPayloadContext<'_, S>,
+    mut gr: GetObjectReader,
+) -> Option<std::io::Error> {
+    let has_tagging_replication = !ctx.put_opts.user_tags.is_empty();
+    if ctx.is_multipart {
+        drop(gr);
+        let result = replicate_object_with_multipart(MultipartReplicationContext {
+            storage: ctx.storage.clone(),
+            cli: ctx.tgt_client.clone(),
+            src_bucket: ctx.bucket,
+            dst_bucket: &ctx.tgt_client.bucket,
+            object: ctx.object,
+            object_info: ctx.object_info,
+            obj_opts: ctx.obj_opts,
+            arn: ctx.arn,
+            put_opts: ctx.put_opts,
+        })
+        .await;
+        record_proxy_request(ctx.bucket, "PutObject", result.is_err()).await;
+        if has_tagging_replication {
+            record_proxy_request(ctx.bucket, "PutObjectTagging", result.is_err()).await;
+        }
+        result.err()
+    } else {
+        gr.stream = wrap_with_bandwidth_monitor(gr.stream, &ctx.put_opts, ctx.bucket, ctx.arn);
+        let byte_stream = async_read_to_bytestream(gr.stream);
+        let result = ctx
+            .tgt_client
+            .put_object(&ctx.tgt_client.bucket, ctx.object, ctx.transfer_size, byte_stream, &ctx.put_opts)
+            .await
+            .map(|assigned_version_id| {
+                audit_target_version_identity(
+                    ctx.tgt_client,
+                    &ctx.put_opts.internal.source_version_id,
+                    assigned_version_id.as_deref(),
+                )
+            })
+            .map_err(|e| std::io::Error::other(e.to_string()));
+        record_proxy_request(ctx.bucket, "PutObject", result.is_err()).await;
+        if has_tagging_replication {
+            record_proxy_request(ctx.bucket, "PutObjectTagging", result.is_err()).await;
+        }
+        result.err()
+    }
+}
+
 fn wrap_with_bandwidth_monitor_with_header(
     stream: Box<dyn AsyncRead + Unpin + Send + Sync>,
     bucket: &str,
diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs
index fba311ba9..b2aee8b10 100644
--- a/rustfs/src/admin/handlers/site_replication.rs
+++ b/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(),