fix(admin): converge site replication IAM deletions and retry backoff (#6962)

* fix(admin): replay recorded IAM deletions in site replication retry drain

An IAM deletion whose peer delivery failed during an outage window was
previously unrecoverable without a manual repair: the collapsed retry
entry carries no body, the snapshot resend cannot express "this entity
no longer exists", and the drain escalated the entry into a permanent
marker. The deleted user kept working credentials on the peer until an
operator intervened — a security exposure (backlog#2071).

Record the verbatim deletion body (user/policy/group-removal/
policy-mapping-clear/service-account) in the persisted state, in the
same transaction as the retry-event upsert. The drain now replays the
recorded deletions before the snapshot resend — snapshot-after ordering
restores any entity recreated locally in the meantime — and settles the
collapsed entry when its whole liability is provably replayed. Entries
that predate recording, merge with legacy rows, or overflow the
per-peer record cap keep the escalation semantics: only explicitly
recorded deletion events are ever replayed, never a cross-site diff.

The peer apply handlers become idempotent for deletion shapes (missing
policy/group/member tolerated, matching the existing user-delete
tolerance), so a replayed deletion that already converged settles
instead of wedging the drain. The IAM change hook now attempts every
peer instead of failing fast, so a multi-site broadcast books a retry
entry (and deletion record) for each unreachable peer rather than only
the first. Repair success and peer removal clear the affected peer's
records alongside the entries they accompany.

* fix(admin): probe recovered peers to lift retry drain backoff

A bucket created while a peer was unreachable accumulated three failed
deliveries and entered exponential backoff (2400s and up, capped at a
day). After the peer recovered, the reconcile tick's drain kept
skipping the entry until the backoff elapsed, so the site stayed
diverged — NoSuchBucket resync noise on the source, missing bucket on
the peer — for up to 24 hours with nothing else driving convergence
(backlog#2071, round-four R1.6).

Split reachability from replay: the drain now probes each peer whose
replayable backlog is held back only by backoff (one cheap devnull POST
per peer per tick) and promotes the backlog when the peer answers, so a
recovered peer converges at the next 600s tick. A failed probe advances
nothing — retry counts only move on real delivery attempts, keeping the
exponential schedule intact for a peer that is genuinely down. The base
backoff still floors re-attempts against a reachable peer that keeps
rejecting a delivery. SITE_REPLICATION_RETRY_FAILED_AFTER stays at 3:
the flag is retryStats visibility only, and with the probe in place an
early failed mark is a timely operator signal rather than a dead end.

The drain tick also logs an operator-visible warning whenever the queue
holds failed or escalated entries, instead of backing off in silence.

rustfs/src/admin/handlers/site_replication.rs

@@ -2739,6 +2739,7 @@ fn set_pending_endpoint_refresh(state: &mut SiteReplicationState, pending: Pendi
         last_error: "endpoint target refresh pending".to_string(),
         updated_at: Some(OffsetDateTime::now_utc()),
         edit_generation: None,
+        deletions_recorded: false,
     });
     state.pending_endpoint_refresh = Some(pending);
     Ok(())
@@ -3152,6 +3153,7 @@ fn remove_sites(mut state: SiteReplicationState, req: SRRemoveReq) -> SiteReplic
         state.peers.clear();
         state.resync_status.clear();
         state.retry_queue.clear();
+        state.iam_deletion_replays.clear();
         state.pending_endpoint_refresh = None;
         state.updated_at = Some(OffsetDateTime::now_utc());
         return state;
@@ -3162,6 +3164,7 @@ fn remove_sites(mut state: SiteReplicationState, req: SRRemoveReq) -> SiteReplic
         state.peers.clear();
         state.resync_status.clear();
         state.retry_queue.clear();
+        state.iam_deletion_replays.clear();
         state.pending_endpoint_refresh = None;
         state.updated_at = Some(OffsetDateTime::now_utc());
         return state;
@@ -3182,6 +3185,11 @@ fn remove_sites(mut state: SiteReplicationState, req: SRRemoveReq) -> SiteReplic
             .iter()
             .any(|(deployment_id, endpoint)| &event.peer_deployment_id == deployment_id || &event.peer_endpoint == endpoint)
     });
+    state.iam_deletion_replays.retain(|record| {
+        !removed_peers
+            .iter()
+            .any(|(deployment_id, endpoint)| &record.peer_deployment_id == deployment_id || &record.peer_endpoint == endpoint)
+    });
     state
         .resync_status
         .retain(|deployment_id, _| state.peers.contains_key(deployment_id));
@@ -5407,7 +5415,14 @@ async fn apply_iam_policy_item(iam_sys: &IamSys<ObjectStore>, name: &str, 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)?;
+        // Idempotent delete: the retry drain replays recorded deletions, and
+        // an entity already absent here IS the converged outcome — erroring
+        // would wedge the replay forever (backlog#2071).
+        match iam_sys.delete_policy(name, true).await {
+            Ok(()) => {}
+            Err(err) if rustfs_iam::error::is_err_no_such_policy(&err) => {}
+            Err(err) => return Err(ApiError::from(err).into()),
+        }
     }
     Ok(())
 }
@@ -5430,10 +5445,26 @@ async fn apply_iam_group_info_item(iam_sys: &IamSys<ObjectStore>, group_info: Op
     };
     let update = group_info.update_req;
     if !group_info_requires_upsert(&update) {
-        iam_sys
+        // Idempotent removal: a replayed deletion may find the group or a
-            .remove_users_from_group(&update.group, update.members)
+        // member already gone (deleted here earlier, or the user tombstone
-            .await
+        // was replayed first) — that IS the converged outcome, and erroring
-            .map_err(ApiError::from)?;
+        // would wedge the retry drain forever (backlog#2071). Members absent
+        // here are dropped individually so one missing user cannot veto the
+        // removal of the members that do exist.
+        let mut members = Vec::with_capacity(update.members.len());
+        for member in update.members.iter() {
+            if iam_sys.get_user(member).await.is_some() {
+                members.push(member.clone());
+            }
+        }
+        if members.is_empty() && !update.members.is_empty() {
+            return Ok(());
+        }
+        match iam_sys.remove_users_from_group(&update.group, members).await {
+            Ok(_) => {}
+            Err(err) if rustfs_iam::error::is_err_no_such_group(&err) => {}
+            Err(err) => return Err(ApiError::from(err).into()),
+        }
         return Ok(());
     }
 
@@ -5841,6 +5872,7 @@ impl Operation for SiteReplicationAddHandler {
                 pending_remove: _,
                 pending_endpoint_refresh: _,
                 retry_queue: _,
+                iam_deletion_replays: _,
                 sync_state_initialized,
                 edit_generation: _,
                 applied_edit_generations: _,
@@ -9889,6 +9921,13 @@ mod tests {
                 path: "/rustfs/admin/v3/site-replication/peer/iam-item".to_string(),
                 ..Default::default()
             }],
+            iam_deletion_replays: vec![SiteReplicationIamDeletionReplay {
+                id: "record-1".to_string(),
+                peer_deployment_id: "remote-dep".to_string(),
+                peer_endpoint: "https://remote.example.com".to_string(),
+                entity: "iam-user:alice".to_string(),
+                ..Default::default()
+            }],
             ..Default::default()
         };
 
@@ -9901,6 +9940,10 @@ mod tests {
         );
 
         assert!(state.retry_queue.is_empty());
+        assert!(
+            state.iam_deletion_replays.is_empty(),
+            "a departed peer's recorded deletions can never be replayed"
+        );
     }
 
     #[test]
@@ -12073,6 +12116,7 @@ mod tests {
                 last_error: "site replication is not enabled".to_string(),
                 updated_at: Some(OffsetDateTime::now_utc()),
                 edit_generation: None,
+                deletions_recorded: false,
             }],
             ..Default::default()
         };
@@ -12270,6 +12314,7 @@ mod tests {
                 last_error: "peer offline".to_string(),
                 updated_at: Some(OffsetDateTime::now_utc()),
                 edit_generation: None,
+                deletions_recorded: false,
             }],
             ..Default::default()
         };

rustfs/src/site_replication/hooks.rs

@@ -434,8 +434,43 @@ pub async fn site_replication_bucket_meta_hook(mut item: SRBucketMeta) -> S3Resu
     .await
 }
 
+/// Broadcast one IAM change to every peer. Unlike the generic JSON broadcast
+/// this attempts ALL peers instead of failing fast, and books every failure —
+/// transport construction included — through the deletion-aware recorder: a
+/// deletion body that failed to reach a peer must be persisted for the retry
+/// drain, or the peer keeps the deleted entity forever (backlog#2071).
 pub async fn site_replication_iam_change_hook(item: SRIAMItem) -> S3Result<()> {
-    broadcast_site_replication_json("/rustfs/admin/v3/site-replication/peer/iam-item", &item).await
+    let Some(runtime) = runtime_site_replication_targets().await? else {
+        return Ok(());
+    };
+    let mut first_error: Option<S3Error> = None;
+    for peer in runtime.state.peers.values() {
+        if peer.deployment_id == runtime.local_peer.deployment_id
+            || same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint)
+        {
+            continue;
+        }
+        let sent = async {
+            let transport = PeerTransport::for_runtime_peer(peer).await?;
+            PeerAdminRequest::put(
+                &transport.connection,
+                SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH,
+                &runtime.state.service_account_access_key,
+            )
+            .with_client(&transport.client)
+            .send(&runtime.service_account_secret_key, &item)
+            .await
+        }
+        .await;
+        match sent {
+            Ok(_) => dequeue_site_replication_retry_event(peer, SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH).await,
+            Err(err) => {
+                record_failed_site_replication_iam_delivery(peer, &item, &err).await;
+                first_error.get_or_insert(err);
+            }
+        }
+    }
+    first_error.map_or(Ok(()), Err)
 }
 
 pub(crate) fn raw_config_to_string(raw: &[u8]) -> Option<String> {

rustfs/src/site_replication/repair.rs

@@ -652,6 +652,13 @@ pub(crate) async fn persist_site_replication_repair_task(
             Some(error) => upsert_site_replication_retry_event(&mut state.retry_queue, &peer, &path, error, None),
             None => {
                 dequeue_site_replication_retry_events_including_escalated(&mut state.retry_queue, &peer, &path);
+                // A repair is the operator's accountability transfer for the
+                // possibly-unreplayed deletions too; keeping the records
+                // without their entry would strand them forever (the drain
+                // only visits queued entries).
+                if collapsed_retry_queue_path(&path) == Some(SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH) {
+                    clear_iam_deletion_replays_for_peer(state, &peer);
+                }
             }
         }
         Ok(())

rustfs/src/site_replication/retry.rs

@@ -16,6 +16,10 @@ use super::*;
 
 pub(crate) const SITE_REPLICATION_RETRY_QUEUE_LIMIT: usize = 256;
 
+/// Attempts before an entry reports as `failed` in retryStats. Visibility
+/// only: a `failed` entry stays drain-eligible, and the reachability probe
+/// short-circuits its backoff once the peer answers again — so an early
+/// `failed` mark is a timely operator signal, not a dead end.
 pub(crate) const SITE_REPLICATION_RETRY_FAILED_AFTER: u32 = 3;
 
 pub(crate) const SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH: &str = "internal:endpoint-target-refresh";
@@ -37,6 +41,14 @@ pub(crate) struct SiteReplicationRetryEvent {
     /// [`settle_site_replication_retry_events`].
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub(crate) edit_generation: Option<u64>,
+    /// Whether every failure folded into this collapsed IAM entry had its
+    /// deletion body (if it was a deletion) recorded in
+    /// [`SiteReplicationState::iam_deletion_replays`]. Only then may a
+    /// successful deletion replay plus a stable snapshot resend settle the
+    /// entry; a legacy entry (or one degraded by record overflow) keeps the
+    /// escalation semantics because an unrecorded deletion may hide in it.
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub(crate) deletions_recorded: bool,
 }
 
 pub(crate) fn retry_event_matches(event: &SiteReplicationRetryEvent, peer: &PeerInfo, path: &str) -> bool {
@@ -87,6 +99,10 @@ pub(crate) fn normalize_collapsed_retry_queue_paths(queue: &mut Vec<SiteReplicat
             (Some(_), None) => true,
             _ => false,
         };
+        // Merged rows may span binaries: a row written without deletion
+        // recording taints the merged entry, so only both-recorded merges
+        // stay settleable.
+        let deletions_recorded = existing.deletions_recorded && event.deletions_recorded;
         if event_is_newer {
             let retry_count = existing.retry_count.max(event.retry_count);
             *existing = event;
@@ -94,6 +110,7 @@ pub(crate) fn normalize_collapsed_retry_queue_paths(queue: &mut Vec<SiteReplicat
         } else {
             existing.retry_count = existing.retry_count.max(event.retry_count);
         }
+        existing.deletions_recorded = deletions_recorded;
         existing.failed = existing.retry_count >= SITE_REPLICATION_RETRY_FAILED_AFTER;
     }
     *queue = normalized;
@@ -210,6 +227,7 @@ pub(crate) fn upsert_site_replication_retry_event(
         last_error: detail,
         updated_at: Some(now),
         edit_generation: generation,
+        deletions_recorded: false,
     });
     if queue.len() > SITE_REPLICATION_RETRY_QUEUE_LIMIT {
         let overflow = queue.len() - SITE_REPLICATION_RETRY_QUEUE_LIMIT;
@@ -272,6 +290,342 @@ pub(crate) async fn enqueue_site_replication_retry_event_for_generation(
     }
 }
 
+pub(crate) const SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH: &str = "/rustfs/admin/v3/site-replication/peer/iam-item";
+
+/// Per-peer cap on recorded deletion bodies. Beyond it the peer's collapsed
+/// IAM entry degrades to the escalation semantics (an unrecorded deletion may
+/// exist), so the list stays bounded without silently dropping liability.
+pub(crate) const SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER: usize = 256;
+
+/// One IAM deletion event whose delivery to `peer` failed, kept verbatim so
+/// the retry drain can replay it before the snapshot resend. `entity`
+/// collapses repeated deletions of the same entity into the newest body.
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub(crate) struct SiteReplicationIamDeletionReplay {
+    pub(crate) id: String,
+    pub(crate) peer_deployment_id: String,
+    pub(crate) peer_endpoint: String,
+    pub(crate) entity: String,
+    pub(crate) item: Value,
+    #[serde(default, with = "time::serde::rfc3339::option", skip_serializing_if = "Option::is_none")]
+    pub(crate) recorded_at: Option<OffsetDateTime>,
+}
+
+pub(crate) fn iam_deletion_replay_matches(record: &SiteReplicationIamDeletionReplay, peer: &PeerInfo) -> bool {
+    record.peer_deployment_id == peer.deployment_id || record.peer_endpoint == peer.endpoint
+}
+
+/// The entity a deletion-shaped IAM item removes, or `None` for items that
+/// create or update state (those are faithfully replayed by the snapshot
+/// resend and need no record). Group member removal keys on the removed
+/// member set: two removals from the same group are distinct events, not a
+/// newer revision of one another.
+pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
+    match item.r#type.as_str() {
+        "policy" if item.policy.is_none() => Some(format!("policy:{}", item.name)),
+        "iam-user" => item
+            .iam_user
+            .as_ref()
+            .filter(|user| user.is_delete_req)
+            .map(|user| format!("iam-user:{}", user.access_key)),
+        "group-info" => item
+            .group_info
+            .as_ref()
+            .filter(|group| group.update_req.is_remove)
+            .map(|group| {
+                let mut members = group.update_req.members.clone();
+                members.sort_unstable();
+                format!("group-remove:{}:{}", group.update_req.group, members.join(","))
+            }),
+        "policy-mapping" => item
+            .policy_mapping
+            .as_ref()
+            .filter(|mapping| mapping.policy.is_empty())
+            .map(|mapping| format!("policy-mapping:{}:{}:{}", mapping.user_or_group, mapping.user_type, mapping.is_group)),
+        "service-account" => item
+            .svc_acc_change
+            .as_ref()
+            .and_then(|change| change.delete.as_ref())
+            .map(|delete| format!("svc-acc:{}", delete.access_key)),
+        _ => None,
+    }
+}
+
+/// Failure bookkeeping for one IAM item delivery: upsert the collapsed retry
+/// event and, when the item is a deletion, record its body for replay. Both
+/// live in the same state so the caller commits them in one transaction — a
+/// retry entry can never exist whose deletion body was lost to a separate
+/// failed write.
+pub(crate) fn record_failed_iam_delivery(state: &mut SiteReplicationState, peer: &PeerInfo, item: &SRIAMItem, error: &str) {
+    let existed = state
+        .retry_queue
+        .iter()
+        .any(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH));
+    upsert_site_replication_retry_event(&mut state.retry_queue, peer, SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH, error, None);
+    if !existed
+        && let Some(event) = state
+            .retry_queue
+            .iter_mut()
+            .find(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH))
+    {
+        // Fresh entry: every failure it will ever collapse goes through this
+        // recording path, so a deletion replay plus a stable snapshot resend
+        // can later settle it instead of escalating.
+        event.deletions_recorded = true;
+    }
+
+    let Some(entity) = iam_item_deletion_entity(item) else {
+        return;
+    };
+    let item_value = match serde_json::to_value(item) {
+        Ok(value) => value,
+        Err(_) => {
+            degrade_iam_retry_event_to_escalation(state, peer);
+            return;
+        }
+    };
+    let now = OffsetDateTime::now_utc();
+    if let Some(existing) = state
+        .iam_deletion_replays
+        .iter_mut()
+        .find(|record| iam_deletion_replay_matches(record, peer) && record.entity == entity)
+    {
+        existing.item = item_value;
+        existing.recorded_at = Some(now);
+        return;
+    }
+
+    let per_peer = state
+        .iam_deletion_replays
+        .iter()
+        .filter(|record| iam_deletion_replay_matches(record, peer))
+        .count();
+    if per_peer >= SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER {
+        // The record set is no longer complete for this peer; the entry must
+        // escalate rather than settle. Drop the oldest record to stay
+        // bounded — remaining records are still replayed best-effort.
+        degrade_iam_retry_event_to_escalation(state, peer);
+        if let Some(oldest) = state
+            .iam_deletion_replays
+            .iter()
+            .enumerate()
+            .filter(|(_, record)| iam_deletion_replay_matches(record, peer))
+            .min_by_key(|(_, record)| record.recorded_at)
+            .map(|(index, _)| index)
+        {
+            state.iam_deletion_replays.remove(oldest);
+        }
+    }
+    state.iam_deletion_replays.push(SiteReplicationIamDeletionReplay {
+        id: Uuid::new_v4().to_string(),
+        peer_deployment_id: peer.deployment_id.clone(),
+        peer_endpoint: peer.endpoint.clone(),
+        entity,
+        item: item_value,
+        recorded_at: Some(now),
+    });
+}
+
+pub(crate) fn degrade_iam_retry_event_to_escalation(state: &mut SiteReplicationState, peer: &PeerInfo) {
+    if let Some(event) = state
+        .retry_queue
+        .iter_mut()
+        .find(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH))
+    {
+        event.deletions_recorded = false;
+    }
+}
+
+pub(crate) fn iam_deletion_replays_for_peer(
+    state: &SiteReplicationState,
+    peer: &PeerInfo,
+) -> Vec<SiteReplicationIamDeletionReplay> {
+    let mut records: Vec<SiteReplicationIamDeletionReplay> = state
+        .iam_deletion_replays
+        .iter()
+        .filter(|record| iam_deletion_replay_matches(record, peer))
+        .cloned()
+        .collect();
+    // Oldest first, so a later deletion of a recreated entity lands after
+    // the earlier one.
+    records.sort_by_key(|record| record.recorded_at);
+    records
+}
+
+pub(crate) fn clear_iam_deletion_replays_for_peer(state: &mut SiteReplicationState, peer: &PeerInfo) {
+    state
+        .iam_deletion_replays
+        .retain(|record| !iam_deletion_replay_matches(record, peer));
+}
+
+pub(crate) async fn record_failed_site_replication_iam_delivery(peer: &PeerInfo, item: &SRIAMItem, error: &S3Error) {
+    let peer_owned = peer.clone();
+    let item_owned = item.clone();
+    let error_text = error.to_string();
+    let deletion_entity = iam_item_deletion_entity(item);
+    let result = update_site_replication_state(move |state| {
+        // A departed peer can never drain its entries again (remove_sites
+        // already pruned them) — mirror enqueue_site_replication_retry_event.
+        if state.peers.contains_key(&peer_owned.deployment_id) {
+            record_failed_iam_delivery(state, &peer_owned, &item_owned, &error_text);
+        }
+        Ok(())
+    })
+    .await;
+
+    match result {
+        Ok(()) => {
+            if let Some(entity) = deletion_entity {
+                warn!(
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    peer = %peer.endpoint,
+                    deployment_id = %peer.deployment_id,
+                    entity = %entity,
+                    result = "iam_deletion_recorded_for_replay",
+                    "IAM deletion delivery to peer failed; recorded for retry-drain replay"
+                );
+            }
+        }
+        Err(err) => {
+            warn!(
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                peer = %peer.endpoint,
+                deployment_id = %peer.deployment_id,
+                error = ?err,
+                "failed to persist site replication IAM delivery failure"
+            );
+        }
+    }
+}
+
+/// Post-replay settlement for a collapsed IAM entry: remove the replayed
+/// deletion records and, when the entry's whole liability is provably
+/// replayed (`deletions_recorded` and no residual records), remove the entry.
+/// Anything else falls back to the escalation marker, and a failure stamped
+/// after `snapshot_updated_at` keeps the entry drain-eligible untouched.
+/// Returns whether the entry was fully settled.
+pub(crate) fn settle_replayed_iam_retry_events(
+    state: &mut SiteReplicationState,
+    peer: &PeerInfo,
+    path: &str,
+    snapshot_updated_at: Option<OffsetDateTime>,
+    replayed_record_ids: &[String],
+) -> bool {
+    state
+        .iam_deletion_replays
+        .retain(|record| !(iam_deletion_replay_matches(record, peer) && replayed_record_ids.contains(&record.id)));
+
+    if collapsed_retry_queue_path(path) != Some(SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH) {
+        return false;
+    }
+    let Some(index) = state
+        .retry_queue
+        .iter()
+        .position(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH))
+    else {
+        return false;
+    };
+    let event = &state.retry_queue[index];
+    let newer_failure_recorded = match (event.updated_at, snapshot_updated_at) {
+        (Some(current), Some(seen)) => current > seen,
+        (Some(_), None) => true,
+        (None, _) => false,
+    };
+    if newer_failure_recorded && event.last_error != SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER {
+        // The newer failure's own deletion (if any) has its own record; the
+        // next drain pass replays it.
+        return false;
+    }
+    let residual_records = state
+        .iam_deletion_replays
+        .iter()
+        .any(|record| iam_deletion_replay_matches(record, peer));
+    if event.deletions_recorded && !residual_records {
+        state.retry_queue.remove(index);
+        return true;
+    }
+    escalate_site_replication_retry_events_up_to(&mut state.retry_queue, peer, path, snapshot_updated_at);
+    false
+}
+
+pub(crate) async fn settle_replayed_site_replication_iam_retry_event(
+    peer: &PeerInfo,
+    path: &str,
+    snapshot_updated_at: Option<OffsetDateTime>,
+    replayed_record_ids: Vec<String>,
+) {
+    let peer_owned = peer.clone();
+    let path_owned = path.to_string();
+    let result = update_site_replication_state(move |state| {
+        Ok(settle_replayed_iam_retry_events(
+            state,
+            &peer_owned,
+            &path_owned,
+            snapshot_updated_at,
+            &replayed_record_ids,
+        ))
+    })
+    .await;
+
+    match result {
+        Ok(true) => {
+            info!(
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                peer = %peer.endpoint,
+                deployment_id = %peer.deployment_id,
+                result = "iam_retry_event_settled",
+                "recorded IAM deletions replayed and snapshot stable; collapsed retry entry settled"
+            );
+        }
+        Ok(false) => {}
+        Err(err) => {
+            warn!(
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                peer = %peer.endpoint,
+                deployment_id = %peer.deployment_id,
+                path,
+                error = ?err,
+                "failed to settle replayed site replication IAM retry event"
+            );
+        }
+    }
+}
+
+/// Drop a deletion record whose body no longer deserializes (it can never be
+/// replayed) and degrade the peer's entry to escalation so the liability
+/// stays operator-visible instead of silently settling.
+pub(crate) async fn drop_corrupt_iam_deletion_replay(peer: &PeerInfo, record_id: &str) {
+    let peer_owned = peer.clone();
+    let record_id_owned = record_id.to_string();
+    let result = update_site_replication_state(move |state| {
+        state.iam_deletion_replays.retain(|record| record.id != record_id_owned);
+        degrade_iam_retry_event_to_escalation(state, &peer_owned);
+        Ok(())
+    })
+    .await;
+
+    if let Err(err) = result {
+        warn!(
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            peer = %peer.endpoint,
+            deployment_id = %peer.deployment_id,
+            record_id,
+            error = ?err,
+            "failed to drop corrupt site replication IAM deletion record"
+        );
+    }
+}
+
 pub(crate) fn retry_bucket_operation(path: &str) -> Option<String> {
     let (base_path, query) = path.split_once('?')?;
     if base_path != SITE_REPLICATION_PEER_BUCKET_OPS_PATH {
@@ -292,7 +646,10 @@ pub(crate) fn retry_event_replayed_by_bootstrap(event: &SiteReplicationRetryEven
 /// reconcile cadence (`site_replication_reconcile::RECONCILE_INTERVAL`).
 pub(crate) const SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS: i64 = 600;
 
-/// Backoff ceiling: a permanently failed peer is still probed daily.
+/// Backoff ceiling for *replay attempts*: a permanently failed peer still
+/// gets a full replay daily. Reachability is separately probed every tick
+/// ([`promote_reachable_deferred_retry_events`]), so a peer that recovers
+/// converges at the next tick instead of waiting out this ceiling.
 pub(crate) const SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS: i64 = 86_400;
 
 /// What the background drain may do for one retry event. Everything not
@@ -628,16 +985,150 @@ pub(crate) fn actionable_site_replication_retry_events(
         .collect()
 }
 
+/// Replayable events currently held back only by backoff. The exponential
+/// backoff exists to spare a *dead* peer the expensive replay (plan build,
+/// snapshot resend) — it must not delay convergence to a peer that has
+/// already RECOVERED, or a failure window ends in up to a day of silent
+/// divergence (backlog#2071). The drain probes each such peer with one cheap
+/// request per tick and promotes its backlog when the probe succeeds. The
+/// base backoff still floors individual re-attempts so a reachable peer that
+/// keeps failing a delivery is not hammered faster than before.
+pub(crate) fn deferred_site_replication_retry_events(
+    state: &SiteReplicationState,
+    now: OffsetDateTime,
+) -> Vec<SiteReplicationRetryEvent> {
+    state
+        .retry_queue
+        .iter()
+        .filter(|event| classify_site_replication_retry_event(event).is_some())
+        .filter(|event| state.peers.contains_key(&event.peer_deployment_id))
+        .filter(|event| !site_replication_retry_backoff_elapsed(event, now))
+        .filter(|event| {
+            event.updated_at.is_none_or(|updated_at| {
+                now.unix_timestamp().saturating_sub(updated_at.unix_timestamp()) >= SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS
+            })
+        })
+        .cloned()
+        .collect()
+}
+
+/// Peer link-check upload drain: a bodiless-in-spirit POST every peer accepts
+/// with the replication service account, discarding the payload. The cheapest
+/// authenticated proof that the peer is reachable again.
+pub(crate) const SITE_REPLICATION_PEER_DEVNULL_PATH: &str = "/rustfs/admin/v3/site-replication/devnull";
+
+pub(crate) async fn probe_site_replication_peer_reachable(runtime: &SiteReplicationRuntime, peer: &PeerInfo) -> bool {
+    let Ok(transport) = PeerTransport::for_runtime_peer(peer).await else {
+        return false;
+    };
+    PeerAdminRequest::post(
+        &transport.connection,
+        SITE_REPLICATION_PEER_DEVNULL_PATH,
+        &runtime.state.service_account_access_key,
+    )
+    .with_client(&transport.client)
+    .send(&runtime.service_account_secret_key, &serde_json::json!({}))
+    .await
+    .is_ok()
+}
+
+/// Probe the peers whose whole backlog is deferred and promote the backlog of
+/// every peer that answers. A probe failure advances nothing: retry counts
+/// only move on real delivery attempts, so the per-event backoff is intact
+/// when the peer is genuinely down.
+pub(crate) async fn promote_reachable_deferred_retry_events(
+    runtime: &SiteReplicationRuntime,
+    actionable: &mut Vec<SiteReplicationRetryEvent>,
+    deferred: Vec<SiteReplicationRetryEvent>,
+) {
+    let due_peers: HashSet<String> = actionable.iter().map(|event| event.peer_deployment_id.clone()).collect();
+    let mut deferred_by_peer: BTreeMap<String, Vec<SiteReplicationRetryEvent>> = BTreeMap::new();
+    for event in deferred {
+        if due_peers.contains(&event.peer_deployment_id) {
+            // The peer is being dialed this tick anyway; its other events keep
+            // their own backoff.
+            continue;
+        }
+        deferred_by_peer
+            .entry(event.peer_deployment_id.clone())
+            .or_default()
+            .push(event);
+    }
+    for (deployment_id, events) in deferred_by_peer {
+        let Some(peer) = runtime.state.peers.get(&deployment_id) else {
+            continue;
+        };
+        if deployment_id == runtime.local_peer.deployment_id
+            || same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint)
+        {
+            continue;
+        }
+        if probe_site_replication_peer_reachable(runtime, peer).await {
+            info!(
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                peer = %peer.endpoint,
+                deployment_id = %deployment_id,
+                promoted = events.len(),
+                result = "retry_backoff_probe_promoted",
+                "peer reachable again; replaying its backed-off retry events this tick"
+            );
+            actionable.extend(events);
+        }
+    }
+}
+
+/// Operator-visible per-tick alert for retry entries that no longer converge
+/// on their own: `failed` deliveries deep in backoff and snapshot-escalated
+/// markers awaiting a repair. Healthy pending entries stay silent.
+pub(crate) fn log_site_replication_retry_liabilities(state: &SiteReplicationState) {
+    let escalated = state
+        .retry_queue
+        .iter()
+        .filter(|event| event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER)
+        .count();
+    let failed = state
+        .retry_queue
+        .iter()
+        .filter(|event| event.failed && event.last_error != SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER)
+        .count();
+    if failed == 0 && escalated == 0 {
+        return;
+    }
+    let pending = state.retry_queue.len().saturating_sub(failed + escalated);
+    let oldest_updated_at = state
+        .retry_queue
+        .iter()
+        .filter(|event| event.failed || event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER)
+        .filter_map(|event| event.updated_at)
+        .min();
+    warn!(
+        component = LOG_COMPONENT_ADMIN,
+        subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+        event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+        failed,
+        escalated,
+        pending,
+        oldest_updated_at = ?oldest_updated_at,
+        recorded_deletions = state.iam_deletion_replays.len(),
+        result = "retry_liabilities_outstanding",
+        "site replication retry queue holds failed or escalated deliveries; peer convergence is degraded"
+    );
+}
+
 /// Background consumer for the retry queue, run from the reconcile tick.
 ///
 /// Scope: this settles "delivered once and failed" entries whose replay is
-/// faithful (bucket ops, peer edits). Collapsed iam-item / bucket-meta
+/// faithful (bucket ops, peer edits). Collapsed iam-item entries replay the
-/// entries are snapshot-resent and then *escalated*, not cleared — a failed
+/// recorded deletion bodies and then the snapshot, which together cover every
-/// deletion leaves no task in the snapshot, so remote absence stays unproven
+/// failure the hook recorded, so a fully-recorded entry settles; an entry
-/// until a later delivery or a manual repair. A hook that never fired (crash
+/// with an unrecorded deletion (legacy rows, record overflow) is *escalated*
-/// between the local commit and the send) leaves no entry at all, so the
+/// instead — remote absence stays unproven until a later delivery or a
-/// drain is not a full cross-site diff-heal; manual repair remains the
+/// manual repair. Collapsed bucket-meta entries keep the escalate-only
-/// authoritative catch-all.
+/// semantics. A hook that never fired (crash between the local commit and
+/// the send) leaves no entry at all, so the drain is not a full cross-site
+/// diff-heal; manual repair remains the authoritative catch-all.
 pub(crate) async fn drain_site_replication_retry_queue() {
     if let Err(err) = drain_site_replication_retry_queue_inner().await {
         warn!(
@@ -655,8 +1146,13 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
     let Some(runtime) = runtime_site_replication_targets().await? else {
         return Ok(());
     };
-    let actionable = actionable_site_replication_retry_events(&runtime.state, OffsetDateTime::now_utc());
+    // The alert must fire even when nothing is drainable this tick —
-    if actionable.is_empty() {
+    // escalated markers are exactly the entries the drain skips.
+    log_site_replication_retry_liabilities(&runtime.state);
+    let now = OffsetDateTime::now_utc();
+    let mut actionable = actionable_site_replication_retry_events(&runtime.state, now);
+    let deferred = deferred_site_replication_retry_events(&runtime.state, now);
+    if actionable.is_empty() && deferred.is_empty() {
         return Ok(());
     }
     let Some(store) = current_object_store_handle() else {
@@ -671,6 +1167,12 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
         // guard) may have started since. Re-check on the fresh state.
         return Ok(());
     }
+    // Probe before taking the repair lock: probes are read-only peer traffic
+    // and a dead peer's connect timeout must not hold the lock.
+    promote_reachable_deferred_retry_events(&runtime, &mut actionable, deferred).await;
+    if actionable.is_empty() {
+        return Ok(());
+    }
     // Serialize against operator repair execution. This does NOT close the
     // dry-run -> execute window (dry-run takes no lock): a drain settling a
     // replayable bucket-op entry in that window changes the preflight token
@@ -778,6 +1280,28 @@ pub(crate) async fn drain_one_site_replication_retry_event(
             let Some(plan) = plan else {
                 return Ok(false);
             };
+            // Replay recorded IAM deletion bodies BEFORE the snapshot: an
+            // entity deleted and later recreated locally is restored by the
+            // snapshot that follows, so the replay can never end below the
+            // current local state (backlog#2071).
+            let is_iam = matches!(action, RetryDrainAction::IamSnapshot);
+            let mut replayed_record_ids = Vec::new();
+            if is_iam {
+                for record in iam_deletion_replays_for_peer(&runtime.state, peer) {
+                    let Ok(item) = serde_json::from_value::<SRIAMItem>(record.item.clone()) else {
+                        drop_corrupt_iam_deletion_replay(peer, &record.id).await;
+                        continue;
+                    };
+                    if let Err(err) = SiteReplicationRepairTask::Iam(&item)
+                        .send(transport, access_key, secret_key)
+                        .await
+                    {
+                        enqueue_site_replication_retry_event(peer, &event.path, &err).await;
+                        return Err(err);
+                    }
+                    replayed_record_ids.push(record.id.clone());
+                }
+            }
             let mut current_snapshot = RetrySnapshot::from_plan(&action, plan).expect("snapshot action has a snapshot");
             let mut replay = current_snapshot.clone();
             for _ in 0..SITE_REPLICATION_RETRY_SNAPSHOT_STABILITY_ATTEMPTS {
@@ -790,7 +1314,17 @@ pub(crate) async fn drain_one_site_replication_retry_event(
                 let fresh_plan = site_replication_bootstrap_plan(&fresh_info)?;
                 let fresh_snapshot = RetrySnapshot::from_plan(&action, &fresh_plan).expect("snapshot action has a snapshot");
                 if fresh_snapshot.fingerprint()? == current_fingerprint {
-                    escalate_site_replication_retry_event_up_to(peer, &event.path, event.updated_at).await;
+                    if is_iam {
+                        settle_replayed_site_replication_iam_retry_event(
+                            peer,
+                            &event.path,
+                            event.updated_at,
+                            replayed_record_ids,
+                        )
+                        .await;
+                    } else {
+                        escalate_site_replication_retry_event_up_to(peer, &event.path, event.updated_at).await;
+                    }
                     return Ok(true);
                 }
                 replay = RetrySnapshot::replay_after_change(&current_snapshot, &fresh_snapshot, OffsetDateTime::now_utc());

rustfs/src/site_replication/state.rs

@@ -41,6 +41,14 @@ pub(crate) struct SiteReplicationState {
     pub(crate) pending_endpoint_refresh: Option<PendingEndpointRefresh>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub(crate) retry_queue: Vec<SiteReplicationRetryEvent>,
+    /// Explicitly recorded IAM deletion events whose delivery to a peer
+    /// failed. The bootstrap snapshot cannot express "this entity no longer
+    /// exists", so these bodies are what makes a failed deletion replayable
+    /// by the retry drain instead of a permanent escalated marker
+    /// (backlog#2071). Only deletions recorded here are ever replayed — the
+    /// drain never derives deletions from a cross-site diff.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub(crate) iam_deletion_replays: Vec<SiteReplicationIamDeletionReplay>,
     #[serde(default)]
     pub(crate) sync_state_initialized: bool,
     /// Fencing token for peer-edit delivery, allocated inside the state
@@ -121,6 +129,14 @@ pub(crate) fn parse_site_replication_state(data: &[u8]) -> S3Result<SiteReplicat
     state
         .applied_edit_generations
         .retain(|origin, _| state.peers.contains_key(origin));
+    // Deletion replay records for a departed peer can never be replayed
+    // (remove_sites prunes them too; this covers any other membership
+    // change) — dropping them on load keeps the list bounded. Matching by
+    // id OR endpoint mirrors the retry queue, so identity re-keying does
+    // not orphan a live peer's records.
+    state
+        .iam_deletion_replays
+        .retain(|record| state.peers.values().any(|peer| iam_deletion_replay_matches(record, peer)));
     if !state.sync_state_initialized {
         if state.enabled() {
             mark_unknown_peer_sync_enabled(&mut state.peers);

rustfs/src/site_replication/tests.rs

@@ -419,9 +419,281 @@ fn drain_event(peer: &str, path: &str, retry_count: u32, updated_at: Option<Offs
         last_error: "remote-operation-failed".to_string(),
         updated_at,
         edit_generation: None,
+        deletions_recorded: false,
     }
 }
 
+fn user_delete_item(access_key: &str) -> SRIAMItem {
+    SRIAMItem {
+        r#type: "iam-user".to_string(),
+        iam_user: Some(rustfs_madmin::SRIAMUser {
+            access_key: access_key.to_string(),
+            is_delete_req: true,
+            user_req: None,
+            api_version: Some(SITE_REPL_API_VERSION.to_string()),
+        }),
+        updated_at: Some(OffsetDateTime::now_utc()),
+        ..Default::default()
+    }
+}
+
+fn policy_delete_item(name: &str) -> SRIAMItem {
+    SRIAMItem {
+        r#type: "policy".to_string(),
+        name: name.to_string(),
+        policy: None,
+        updated_at: Some(OffsetDateTime::now_utc()),
+        ..Default::default()
+    }
+}
+
+fn deletion_replay_state(peer: &PeerInfo) -> SiteReplicationState {
+    let mut state = SiteReplicationState::default();
+    state.peers.insert(peer.deployment_id.clone(), peer.clone());
+    state
+}
+
+/// Deletion-shaped IAM items get an entity key (and hence a replay record);
+/// creations and updates are covered by the snapshot resend and get none.
+#[test]
+fn test_iam_item_deletion_entity_shapes() {
+    assert_eq!(iam_item_deletion_entity(&user_delete_item("alice")).as_deref(), Some("iam-user:alice"));
+    assert_eq!(
+        iam_item_deletion_entity(&policy_delete_item("readonly")).as_deref(),
+        Some("policy:readonly")
+    );
+
+    let group_remove = SRIAMItem {
+        r#type: "group-info".to_string(),
+        group_info: Some(SRGroupInfo {
+            update_req: GroupAddRemove {
+                group: "devs".to_string(),
+                members: vec!["bob".to_string(), "alice".to_string()],
+                status: GroupStatus::Enabled,
+                is_remove: true,
+            },
+            api_version: None,
+        }),
+        ..Default::default()
+    };
+    assert_eq!(
+        iam_item_deletion_entity(&group_remove).as_deref(),
+        Some("group-remove:devs:alice,bob"),
+        "member set must be part of the key so distinct removals do not collapse"
+    );
+
+    let mapping_clear = SRIAMItem {
+        r#type: "policy-mapping".to_string(),
+        policy_mapping: Some(SRPolicyMapping {
+            user_or_group: "alice".to_string(),
+            user_type: 0,
+            is_group: false,
+            policy: String::new(),
+            ..Default::default()
+        }),
+        ..Default::default()
+    };
+    assert_eq!(iam_item_deletion_entity(&mapping_clear).as_deref(), Some("policy-mapping:alice:0:false"));
+
+    let svc_acc_delete = SRIAMItem {
+        r#type: "service-account".to_string(),
+        svc_acc_change: Some(rustfs_madmin::SRSvcAccChange {
+            delete: Some(rustfs_madmin::SRSvcAccDelete {
+                access_key: "svc-1".to_string(),
+                api_version: None,
+            }),
+            ..Default::default()
+        }),
+        ..Default::default()
+    };
+    assert_eq!(iam_item_deletion_entity(&svc_acc_delete).as_deref(), Some("svc-acc:svc-1"));
+
+    // Creations/updates carry no deletion entity.
+    let mut user_create = user_delete_item("alice");
+    user_create.iam_user.as_mut().expect("iam user").is_delete_req = false;
+    assert!(iam_item_deletion_entity(&user_create).is_none());
+    let mut policy_set = policy_delete_item("readonly");
+    policy_set.policy = Some(serde_json::json!({"Version": "2012-10-17"}));
+    assert!(iam_item_deletion_entity(&policy_set).is_none());
+}
+
+/// A failed deletion delivery persists a replay record next to the collapsed
+/// retry entry; a fresh entry is stamped `deletions_recorded` so a later
+/// replay can settle it, and a repeated deletion of the same entity keeps the
+/// newest body instead of growing the list.
+#[test]
+fn test_record_failed_iam_delivery_records_deletions_and_flags_entry() {
+    let target = PeerInfo {
+        deployment_id: "remote-dep".to_string(),
+        ..peer("remote", "https://remote.example.com")
+    };
+    let mut state = deletion_replay_state(&target);
+
+    // Non-deletion failure: entry flagged, no record.
+    let mut user_update = user_delete_item("alice");
+    user_update.iam_user.as_mut().expect("iam user").is_delete_req = false;
+    record_failed_iam_delivery(&mut state, &target, &user_update, "peer offline");
+    assert_eq!(state.retry_queue.len(), 1);
+    assert_eq!(state.retry_queue[0].path, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH);
+    assert!(state.retry_queue[0].deletions_recorded);
+    assert!(state.iam_deletion_replays.is_empty());
+
+    // Deletion failure: recorded for replay, entry stays flagged.
+    record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline");
+    assert_eq!(state.iam_deletion_replays.len(), 1);
+    assert_eq!(state.iam_deletion_replays[0].entity, "iam-user:alice");
+    assert!(state.retry_queue[0].deletions_recorded);
+    assert_eq!(state.retry_queue.len(), 1, "IAM failures stay collapsed per peer");
+
+    // Same entity again: newest body replaces the record.
+    record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline");
+    assert_eq!(state.iam_deletion_replays.len(), 1);
+
+    // Different entity: second record.
+    record_failed_iam_delivery(&mut state, &target, &policy_delete_item("readonly"), "peer offline");
+    assert_eq!(state.iam_deletion_replays.len(), 2);
+
+    // A legacy entry (created without recording) is never stamped.
+    let legacy = PeerInfo {
+        deployment_id: "legacy-dep".to_string(),
+        ..peer("legacy", "https://legacy.example.com")
+    };
+    state.peers.insert(legacy.deployment_id.clone(), legacy.clone());
+    upsert_site_replication_retry_event(
+        &mut state.retry_queue,
+        &legacy,
+        SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH,
+        "peer offline",
+        None,
+    );
+    record_failed_iam_delivery(&mut state, &legacy, &user_delete_item("bob"), "peer offline");
+    let legacy_event = state
+        .retry_queue
+        .iter()
+        .find(|event| event.peer_deployment_id == legacy.deployment_id)
+        .expect("legacy entry");
+    assert!(
+        !legacy_event.deletions_recorded,
+        "an entry that predates recording may hide an unrecorded deletion"
+    );
+}
+
+/// Overflowing the per-peer record cap degrades the entry back to the
+/// escalation semantics: the record set is no longer complete, so a replay
+/// can no longer prove the peer converged.
+#[test]
+fn test_record_failed_iam_delivery_overflow_degrades_to_escalation() {
+    let target = PeerInfo {
+        deployment_id: "remote-dep".to_string(),
+        ..peer("remote", "https://remote.example.com")
+    };
+    let mut state = deletion_replay_state(&target);
+    for index in 0..SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER {
+        record_failed_iam_delivery(&mut state, &target, &policy_delete_item(&format!("p{index}")), "peer offline");
+    }
+    assert!(state.retry_queue[0].deletions_recorded);
+    assert_eq!(state.iam_deletion_replays.len(), SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER);
+
+    record_failed_iam_delivery(&mut state, &target, &policy_delete_item("one-too-many"), "peer offline");
+    assert_eq!(
+        state.iam_deletion_replays.len(),
+        SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER,
+        "the list stays bounded"
+    );
+    assert!(
+        !state.retry_queue[0].deletions_recorded,
+        "an overflowed record set can no longer settle the entry"
+    );
+}
+
+/// After a successful deletion replay plus a stable snapshot resend, a
+/// fully-recorded entry settles (entry and replayed records removed); an
+/// unrecorded entry escalates as before, and a failure stamped after the
+/// snapshot keeps the entry drain-eligible.
+#[test]
+fn test_settle_replayed_iam_retry_events_settles_or_escalates() {
+    let target = PeerInfo {
+        deployment_id: "remote-dep".to_string(),
+        ..peer("remote", "https://remote.example.com")
+    };
+    let snapshot_at = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+
+    // Fully recorded: settles.
+    let mut state = deletion_replay_state(&target);
+    record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline");
+    state.retry_queue[0].updated_at = Some(snapshot_at);
+    let replayed: Vec<String> = state.iam_deletion_replays.iter().map(|record| record.id.clone()).collect();
+    assert!(settle_replayed_iam_retry_events(
+        &mut state,
+        &target,
+        SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH,
+        Some(snapshot_at),
+        &replayed,
+    ));
+    assert!(state.retry_queue.is_empty());
+    assert!(state.iam_deletion_replays.is_empty());
+
+    // Not fully recorded: replayed records are still removed, but the entry
+    // escalates instead of settling.
+    let mut state = deletion_replay_state(&target);
+    record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline");
+    state.retry_queue[0].updated_at = Some(snapshot_at);
+    state.retry_queue[0].deletions_recorded = false;
+    let replayed: Vec<String> = state.iam_deletion_replays.iter().map(|record| record.id.clone()).collect();
+    assert!(!settle_replayed_iam_retry_events(
+        &mut state,
+        &target,
+        SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH,
+        Some(snapshot_at),
+        &replayed,
+    ));
+    assert!(state.iam_deletion_replays.is_empty());
+    assert_eq!(state.retry_queue.len(), 1);
+    assert_eq!(state.retry_queue[0].last_error, SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER);
+
+    // Newer failure since the snapshot: entry untouched and drain-eligible,
+    // residual (unreplayed) record kept for the next pass.
+    let mut state = deletion_replay_state(&target);
+    record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline");
+    let replayed: Vec<String> = state.iam_deletion_replays.iter().map(|record| record.id.clone()).collect();
+    record_failed_iam_delivery(&mut state, &target, &user_delete_item("bob"), "peer offline");
+    state.retry_queue[0].updated_at = Some(snapshot_at + time::Duration::seconds(5));
+    assert!(!settle_replayed_iam_retry_events(
+        &mut state,
+        &target,
+        SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH,
+        Some(snapshot_at),
+        &replayed,
+    ));
+    assert_eq!(state.retry_queue.len(), 1);
+    assert_ne!(state.retry_queue[0].last_error, SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER);
+    assert!(
+        classify_site_replication_retry_event(&state.retry_queue[0]).is_some(),
+        "the newer failure must stay drain-eligible"
+    );
+    assert_eq!(state.iam_deletion_replays.len(), 1);
+    assert_eq!(state.iam_deletion_replays[0].entity, "iam-user:bob");
+}
+
+/// Merging legacy wire-path rows into the collapsed entry must not launder an
+/// unrecorded deletion into a settleable entry.
+#[test]
+fn test_normalize_collapsed_paths_taints_merged_deletion_recording() {
+    let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+    let mut recorded = drain_event("remote-dep", SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, 1, Some(now));
+    recorded.deletions_recorded = true;
+    let legacy = drain_event(
+        "remote-dep",
+        "/rustfs/admin/v3/site-replication/peer/iam-item",
+        2,
+        Some(now + time::Duration::seconds(5)),
+    );
+    let mut queue = vec![recorded, legacy];
+    assert!(normalize_collapsed_retry_queue_paths(&mut queue));
+    assert_eq!(queue.len(), 1);
+    assert!(!queue[0].deletions_recorded, "a merged legacy row may hide an unrecorded deletion");
+}
+
 /// P1-3 red-light: the drain must only ever act on deliveries it can
 /// replay faithfully. IAM / bucket-meta entries collapse per (peer, path)
 /// with no body persisted — only a snapshot resend is truthful; bucket
@@ -590,6 +862,59 @@ fn test_actionable_site_replication_retry_events_filters() {
     assert_eq!(actionable[0].path, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH);
 }
 
+/// The deferred subset is the probe's territory: replayable events held back
+/// only by backoff, at least one base interval after their last failure. A
+/// recovered peer's bucket-op stuck behind a 2400s+ backoff (the round-four
+/// R1.6 shape: three outage-window failures, then a 900s test window) must
+/// appear here so the probe can promote it at the first tick.
+#[test]
+fn test_deferred_site_replication_retry_events_partition() {
+    let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+    let mut state = SiteReplicationState::default();
+    state
+        .peers
+        .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
+
+    let bucket_make = "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning";
+    state.retry_queue = vec![
+        // retry_count 3 => 2400s backoff; failed 700s ago: deferred.
+        drain_event("remote", bucket_make, 3, Some(now - time::Duration::seconds(700))),
+        // Failed less than one base interval ago: neither due nor probed.
+        drain_event(
+            "remote",
+            SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH,
+            3,
+            Some(now - time::Duration::seconds(300)),
+        ),
+        // Past its own backoff: actionable, not deferred.
+        drain_event(
+            "remote",
+            "/rustfs/admin/v3/site-replication/peer/bucket-meta",
+            1,
+            Some(now - time::Duration::seconds(700)),
+        ),
+        // Unknown peer: never probed.
+        drain_event("gone", bucket_make, 3, Some(now - time::Duration::seconds(700))),
+    ];
+    // Escalated marker: not replayable, never probed.
+    let mut escalated = drain_event(
+        "remote",
+        SITE_REPLICATION_RETRY_BUCKET_METADATA_SNAPSHOT_PATH,
+        3,
+        Some(now - time::Duration::seconds(700)),
+    );
+    escalated.last_error = SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER.to_string();
+    state.retry_queue.push(escalated);
+
+    let deferred = deferred_site_replication_retry_events(&state, now);
+    assert_eq!(deferred.len(), 1, "only the backed-off, replayable, known-peer event defers");
+    assert_eq!(deferred[0].path, bucket_make);
+
+    let actionable = actionable_site_replication_retry_events(&state, now);
+    assert_eq!(actionable.len(), 1);
+    assert_eq!(actionable[0].path, "/rustfs/admin/v3/site-replication/peer/bucket-meta");
+}
+
 /// The drain settles a peer-edit success under a freshly allocated
 /// generation; legacy queue entries carry `edit_generation: None` and
 /// must be cleared by that generation-scoped settlement (`(Some, None)`