fix(site-replication): clear pending_remove on join, surface failures

A removal that could not notify its peers left `pending_remove` set
forever. That single field gates `SRPeerBucketOpsHandler` ahead of
`enabled()`, so the site rejected every peer bucket-op with
"site replication is not enabled" while nothing surfaced on the source:
`mb` succeeded, `replicate info` still reported a healthy cluster, and no
client command exited non-zero.

An accepted peer join now clears the marker, so a re-add actually
repairs the cluster instead of restoring the topology on both sides
while replication stays dead. The peer-edit high-water marks are
deliberately untouched — those fence edit ordering, not lifecycle.

Also:

- `SRPeerJoinResponse.applied` makes a no-op join distinguishable. It is
  three-valued: `None` means the peer did not report (MinIO answers a
  successful join with an empty body), so it is never read as failure.
  The rotation fan-out consumes it too: a superseded join returns before
  `apply_iam`, so acking one finalized a rotation whose new secret the
  peer never installed.
- The reconcile tick re-drives a stuck removal instead of giving up the
  round, so a peer coming back finishes it without operator action.
- `replicate info` carries `retryStats` and `pendingOperation`. Both are
  omitted when absent, so a healthy site stays wire-identical.
- A removal with unnotified peers answers `Partial` instead of the
  success string. The fully-notified path is unchanged byte for byte.

The lab script gains a `diverge` subcommand that reproduces the report
end to end and asserts recovery.

Fixes #5963

crates/madmin/src/site_replication.rs

@@ -83,6 +83,16 @@ pub struct SiteReplicationInfo {
     pub service_account_access_key: String,
     #[serde(rename = "apiVersion", skip_serializing_if = "Option::is_none")]
     pub api_version: Option<String>,
+    /// Outstanding peer deliveries. Absent when the retry queue is empty, so a
+    /// healthy site serializes exactly as it did before this field existed.
+    /// Present means peer operations are failing even if `enabled` is true.
+    #[serde(rename = "retryStats", default, skip_serializing_if = "Option::is_none")]
+    pub retry_stats: Option<SRRetryStats>,
+    /// A multi-step lifecycle operation this site has not finished — most
+    /// importantly a removal that could not reach its peers, which makes the
+    /// site reject peer operations while `enabled` may still read true.
+    #[serde(rename = "pendingOperation", default, skip_serializing_if = "Option::is_none")]
+    pub pending_operation: Option<SRPendingOperation>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]

rustfs/src/admin/handlers/site_replication.rs

@@ -126,6 +126,10 @@ const SITE_REPLICATION_JOIN_ADMISSION_LOCK_PATH: &str = "config/site-replication
 const SITE_REPL_ADD_SUCCESS: &str = "Requested sites were configured for replication successfully.";
 const SITE_REPL_EDIT_SUCCESS: &str = "Requested site was updated successfully.";
 const SITE_REPL_REMOVE_SUCCESS: &str = "Requested site(s) were removed from cluster replication successfully.";
+/// Local removal committed, but at least one peer could not be told. The
+/// cluster is diverged until the removal finishes — the reconcile tick keeps
+/// retrying it, and `replicate info` reports the pending operation meanwhile.
+const SITE_REPL_REMOVE_PARTIAL: &str = "Partial";
 const SITE_REPL_RESYNC_START: &str = "start";
 const SITE_REPL_RESYNC_CANCEL: &str = "cancel";
 const SITE_REPL_RESYNC_STATUS: &str = "status";
@@ -713,6 +717,16 @@ struct SRPeerJoinResponse {
     peer: PeerInfo,
     #[serde(rename = "initialSyncErrorMessage", default, skip_serializing_if = "String::is_empty")]
     initial_sync_error_message: String,
+    /// Whether the receiving site actually applied this join.
+    ///
+    /// Three-valued on purpose. `None` means the peer did not report — MinIO
+    /// answers a successful `SRPeerJoin` with an empty body, and RustFS peers
+    /// older than this field say nothing either — so the initiator must NOT
+    /// read it as a failure. `Some(false)` is an explicit no-op: the peer had
+    /// already moved past the snapshot it was sent and wrote nothing, which
+    /// used to be indistinguishable from success (rustfs/rustfs#5963).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    applied: Option<bool>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -2568,6 +2582,21 @@ fn apply_peer_join(
     state.peers = normalize_join_peers_for_local(local_peer, join_req.peers);
     initialize_join_peer_sync_state(&mut state.peers, defer_sync_state_enable);
     state.sync_state_initialized = true;
+    // An accepted join supersedes a half-finished removal this site started:
+    // the sender's snapshot IS the new topology, while the pending record only
+    // exists to keep notifying peers about the OLD one. Leaving it set is what
+    // kept a recovered site rejecting every peer bucket-op forever —
+    // `SRPeerBucketOpsHandler` short-circuits on `pending_remove` BEFORE it
+    // consults `enabled()`, so a successful re-add restored the topology on
+    // both sides while replication stayed dead (rustfs/rustfs#5963).
+    //
+    // Safe against a concurrent removal: `SiteReplicationRemoveHandler` and
+    // the join admission both hold the lifecycle guard, so a join is only ever
+    // admitted before that handler starts or after it has returned.
+    //
+    // Deliberately NOT cleared here: the peer-edit high-water marks (see this
+    // function's doc comment) — those fence edit ordering, not lifecycle.
+    state.pending_remove = None;
     state.name = state
         .peers
         .get(&local_peer.deployment_id)
@@ -3002,8 +3031,17 @@ fn reconcile_site_replication_wiring() -> std::pin::Pin<Box<dyn std::future::Fut
 
         match load_site_replication_state().await {
             Ok(state) => {
-                if state.pending_endpoint_refresh.is_some() || state.pending_remove.is_some() || state.pending_rotation.is_some()
-                {
+                if state.pending_endpoint_refresh.is_some() || state.pending_rotation.is_some() {
+                    return;
+                }
+                // A removal whose peers were unreachable is the one pending
+                // marker that nothing else re-drives, and it wedges the site
+                // while it sits there. Push it forward here rather than giving
+                // up the round (rustfs/rustfs#5963). The reconcilers below
+                // still skip this round either way: the topology is only
+                // settled once the removal clears, and the next tick sees it.
+                if let Some(pending_remove) = state.pending_remove.clone() {
+                    resume_pending_remove(&state, &pending_remove).await;
                     return;
                 }
             }
@@ -7041,15 +7079,31 @@ async fn dequeue_site_replication_retry_event_for_generation(peer: &PeerInfo, pa
     }
 }
 
+/// The removal's client-facing verdict.
+///
+/// A fully-notified removal keeps answering with the historical success string,
+/// byte for byte, so healthy runs stay wire-identical for every existing
+/// client. Only the path that used to LIE — peers that could not be notified,
+/// reported as unqualified success while the cluster silently diverged
+/// (rustfs/rustfs#5963) — now says `Partial`, matching the vocabulary
+/// `SRRotateServiceAccountHandler` already uses for the same situation.
 fn site_replication_remove_status(peer_errors: &[String]) -> ReplicateRemoveStatus {
+    if peer_errors.is_empty() {
+        return ReplicateRemoveStatus {
+            status: SITE_REPL_REMOVE_SUCCESS.to_string(),
+            err_detail: String::new(),
+            api_version: Some(SITE_REPL_API_VERSION.to_string()),
+        };
+    }
+
+    let summaries: Vec<String> = peer_errors.iter().map(|error| summarize_peer_error_detail(error)).collect();
     ReplicateRemoveStatus {
-        status: SITE_REPL_REMOVE_SUCCESS.to_string(),
-        err_detail: if peer_errors.is_empty() {
-            String::new()
-        } else {
-            let summaries: Vec<String> = peer_errors.iter().map(|error| summarize_peer_error_detail(error)).collect();
-            summarize_peer_error_detail(&format!("failed to notify {} peer(s): {}", summaries.len(), summaries.join("; ")))
-        },
+        status: SITE_REPL_REMOVE_PARTIAL.to_string(),
+        err_detail: summarize_peer_error_detail(&format!(
+            "failed to notify {} peer(s): {}",
+            summaries.len(),
+            summaries.join("; ")
+        )),
         api_version: Some(SITE_REPL_API_VERSION.to_string()),
     }
 }
@@ -7226,6 +7280,137 @@ async fn clear_pending_remove(remove_id: &str) -> S3Result<()> {
     .await
 }
 
+/// Push a half-finished removal one step forward: notify every peer that has
+/// not acked yet, then finalize locally if that completed the set. Returns the
+/// per-peer failures and whether the removal is now finished.
+///
+/// Shared by the operator-driven `SiteReplicationRemoveHandler` and the
+/// reconcile tick. The tick is what makes this self-healing: a removal whose
+/// peers were unreachable used to sit in `pending_remove` forever, and that one
+/// field gates every peer bucket-op (`SRPeerBucketOpsHandler` checks it first)
+/// plus every reconciler — so the site stayed wedged until an operator happened
+/// to re-run `replicate remove` (rustfs/rustfs#5963).
+///
+/// Callers must hold the lifecycle guard: this both notifies peers and, on the
+/// final step, takes the bucket-op write lock to clean up local rules.
+async fn drive_pending_remove(pending_remove: &PendingRemove, local_peer: &PeerInfo) -> S3Result<(Vec<String>, bool)> {
+    let mut peer_errors = Vec::new();
+    let mut secret_candidates = pending_remove.secret_candidates.clone();
+    if pending_remove.service_account_access_key.is_empty() {
+        peer_errors.push("site replication service account unavailable".to_string());
+    } else if let Ok(service_account_secret_key) =
+        site_replicator_service_account_secret(&pending_remove.service_account_access_key).await
+    {
+        record_pending_remove_secret_candidate(&pending_remove.id, service_account_secret_key.clone()).await?;
+        push_unique_secret_candidate(&mut secret_candidates, service_account_secret_key);
+    }
+
+    if secret_candidates.is_empty() {
+        peer_errors.push("site replication service account secret unavailable".to_string());
+    } else {
+        for peer in pending_remove.original_peers.values() {
+            if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
+                || pending_remove.acked_deployment_ids.contains(&peer.deployment_id)
+            {
+                continue;
+            }
+            if let Err(err) = send_peer_admin_request_with_secret_candidates(
+                &runtime_peer_connection(peer)?,
+                SITE_REPLICATION_PEER_REMOVE_PATH,
+                &pending_remove.service_account_access_key,
+                &secret_candidates,
+                &pending_remove.req,
+            )
+            .await
+            {
+                let err_detail = summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint));
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    peer = %peer.endpoint,
+                    result = "peer_remove_notification_failed",
+                    error = %err_detail,
+                    "admin site replication state"
+                );
+                peer_errors.push(err_detail);
+            } else {
+                mark_pending_remove_peer_acked(&pending_remove.id, &peer.deployment_id).await?;
+            }
+        }
+    }
+
+    let finalize_candidate = pending_remove_ready_to_finalize(&pending_remove.id, local_peer).await?;
+    let complete = if let Some(finalized_remove) = finalize_candidate {
+        let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await;
+        let removed_deployment_ids = removed_deployment_ids_for_pending_remove(&finalized_remove, local_peer);
+        match cleanup_removed_site_replication_buckets(&removed_deployment_ids).await {
+            Ok(removed) => {
+                if removed > 0 {
+                    info!(
+                        event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                        component = LOG_COMPONENT_ADMIN,
+                        subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                        removed,
+                        result = "remove_cleanup_completed",
+                        "admin site replication state"
+                    );
+                }
+                clear_pending_remove(&pending_remove.id).await?;
+                true
+            }
+            Err(err) => {
+                peer_errors.push(summarize_peer_error_detail(&format!("local remove cleanup failed: {err}")));
+                false
+            }
+        }
+    } else {
+        false
+    };
+
+    Ok((peer_errors, complete))
+}
+
+/// The reconcile tick's half of [`drive_pending_remove`]: resume the removal
+/// this site could not finish, and report the outcome. Runs under the tick's
+/// lifecycle guard, which is what keeps it from racing an operator re-running
+/// `replicate remove` (that handler takes the same guard).
+async fn resume_pending_remove(state: &SiteReplicationState, pending_remove: &PendingRemove) {
+    let local_peer = current_local_runtime_peer(state);
+    match drive_pending_remove(pending_remove, &local_peer).await {
+        Ok((peer_errors, complete)) => {
+            if complete && peer_errors.is_empty() {
+                info!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "pending_remove_resumed",
+                    "admin site replication state"
+                );
+            } else {
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "pending_remove_still_pending",
+                    error_count = peer_errors.len(),
+                    "admin site replication state"
+                );
+            }
+        }
+        Err(err) => {
+            warn!(
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                result = "pending_remove_resume_failed",
+                error = ?err,
+                "admin site replication state"
+            );
+        }
+    }
+}
+
 fn removed_deployment_ids_for_pending_remove(pending: &PendingRemove, local_peer: &PeerInfo) -> HashSet<String> {
     if pending.req.remove_all || pending.req.site_names.iter().any(|name| name == &local_peer.name) {
         return pending
@@ -9656,9 +9841,12 @@ pub struct SiteReplicationAddHandler {}
 /// peer identity from the add preflight metainfo in that case.
 fn parse_peer_join_response(body: &[u8], fallback_peer: PeerInfo) -> Result<SRPeerJoinResponse, serde_json::Error> {
     if body.iter().all(u8::is_ascii_whitespace) {
+        // MinIO's empty-body success. `applied` stays `None`: the peer told us
+        // nothing, which must not be reported as a no-op join.
         return Ok(SRPeerJoinResponse {
             peer: fallback_peer,
             initial_sync_error_message: String::new(),
+            applied: None,
         });
     }
     serde_json::from_slice(body)
@@ -9761,6 +9949,19 @@ impl Operation for SiteReplicationAddHandler {
             if !join_response.initial_sync_error_message.is_empty() {
                 initial_sync_errors.push(format!("{}: {}", site.endpoint, join_response.initial_sync_error_message));
             }
+            // An explicit no-op join. The peer answered 200 but wrote nothing —
+            // its persisted state is already newer than the snapshot it was
+            // sent — so the add is only PARTIALLY configured and saying
+            // "configured successfully" would be a lie (rustfs/rustfs#5963).
+            // `None` (a MinIO peer, or one older than the field) is not a
+            // no-op signal and is deliberately not reported.
+            if join_response.applied == Some(false) {
+                initial_sync_errors.push(format!(
+                    "{}: peer did not apply the join (its site replication state is newer than the snapshot it was sent); \
+                     the site is not configured against this peer",
+                    site.endpoint
+                ));
+            }
             state = reconcile_peer_with_actual_identity(state, join_response.peer);
             let reconciled_peer = existing_peer_for_endpoint(&state, &site.endpoint).ok_or_else(|| {
                 S3Error::with_message(
@@ -9933,79 +10134,7 @@ impl Operation for SiteReplicationRemoveHandler {
             .await?
         };
 
-        let mut peer_errors = Vec::new();
-        let mut secret_candidates = pending_remove.secret_candidates.clone();
-        if pending_remove.service_account_access_key.is_empty() {
-            peer_errors.push("site replication service account unavailable".to_string());
-        } else if let Ok(service_account_secret_key) =
-            site_replicator_service_account_secret(&pending_remove.service_account_access_key).await
-        {
-            record_pending_remove_secret_candidate(&pending_remove.id, service_account_secret_key.clone()).await?;
-            push_unique_secret_candidate(&mut secret_candidates, service_account_secret_key);
-        }
-
-        if secret_candidates.is_empty() {
-            peer_errors.push("site replication service account secret unavailable".to_string());
-        } else {
-            for peer in pending_remove.original_peers.values() {
-                if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
-                    || pending_remove.acked_deployment_ids.contains(&peer.deployment_id)
-                {
-                    continue;
-                }
-                if let Err(err) = send_peer_admin_request_with_secret_candidates(
-                    &runtime_peer_connection(peer)?,
-                    SITE_REPLICATION_PEER_REMOVE_PATH,
-                    &pending_remove.service_account_access_key,
-                    &secret_candidates,
-                    &pending_remove.req,
-                )
-                .await
-                {
-                    let err_detail = summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint));
-                    warn!(
-                        event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-                        component = LOG_COMPONENT_ADMIN,
-                        subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-                        peer = %peer.endpoint,
-                        result = "peer_remove_notification_failed",
-                        error = %err_detail,
-                        "admin site replication state"
-                    );
-                    peer_errors.push(err_detail);
-                } else {
-                    mark_pending_remove_peer_acked(&pending_remove.id, &peer.deployment_id).await?;
-                }
-            }
-        }
-
-        let finalize_candidate = pending_remove_ready_to_finalize(&pending_remove.id, &local_peer).await?;
-        let complete = if let Some(finalized_remove) = finalize_candidate {
-            let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await;
-            let removed_deployment_ids = removed_deployment_ids_for_pending_remove(&finalized_remove, &local_peer);
-            match cleanup_removed_site_replication_buckets(&removed_deployment_ids).await {
-                Ok(removed) => {
-                    if removed > 0 {
-                        info!(
-                            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-                            component = LOG_COMPONENT_ADMIN,
-                            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-                            removed,
-                            result = "remove_cleanup_completed",
-                            "admin site replication state"
-                        );
-                    }
-                    clear_pending_remove(&pending_remove.id).await?;
-                    true
-                }
-                Err(err) => {
-                    peer_errors.push(summarize_peer_error_detail(&format!("local remove cleanup failed: {err}")));
-                    false
-                }
-            }
-        } else {
-            false
-        };
+        let (mut peer_errors, complete) = drive_pending_remove(&pending_remove, &local_peer).await?;
         if !complete && peer_errors.is_empty() {
             peer_errors.push("site replication remove is still pending".to_string());
         }
@@ -10019,6 +10148,25 @@ impl Operation for SiteReplicationRemoveHandler {
     }
 }
 
+/// The `replicate info` projection.
+///
+/// Carries the peer-facing health this endpoint used to omit entirely: a peer
+/// rejecting every operation, or a removal stuck mid-flight, left `info`
+/// reporting a perfectly healthy cluster while replication was dead — both were
+/// only visible through `replicate status --json` (rustfs/rustfs#5963). Split
+/// out so that omission is a test failure rather than an invisible regression.
+fn site_replication_info_for(state: &SiteReplicationState, local_peer: &PeerInfo) -> SiteReplicationInfo {
+    SiteReplicationInfo {
+        enabled: state.enabled(),
+        name: local_peer.name.clone(),
+        sites: state.peers.values().cloned().collect(),
+        service_account_access_key: state.service_account_access_key.clone(),
+        api_version: Some(SITE_REPL_API_VERSION.to_string()),
+        retry_stats: retry_stats_for_state(state),
+        pending_operation: pending_operation_for_state(state, local_peer),
+    }
+}
+
 pub struct SiteReplicationInfoHandler {}
 
 #[async_trait::async_trait]
@@ -10027,14 +10175,7 @@ impl Operation for SiteReplicationInfoHandler {
         validate_site_replication_admin_request(&req, AdminAction::SiteReplicationInfoAction).await?;
         let state = load_site_replication_state().await?;
         let local_peer = current_local_peer(&req, &state);
-        let info = SiteReplicationInfo {
-            enabled: state.enabled(),
-            name: local_peer.name,
-            sites: state.peers.values().cloned().collect(),
-            service_account_access_key: state.service_account_access_key,
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-        };
-        json_response(&info)
+        json_response(&site_replication_info_for(&state, &local_peer))
     }
 }
 
@@ -10253,6 +10394,28 @@ async fn apply_peer_join_service_account(join_req: SRPeerJoinReq) -> S3Result<()
     Ok(())
 }
 
+/// The answer to a join this site refused to apply because it had already
+/// moved past the sender's snapshot. Split out so the verdict itself is
+/// testable: answering `applied: Some(true)` here (or omitting the field) is
+/// exactly the silent no-op that made `replicate add` report success against a
+/// peer that wrote nothing (rustfs/rustfs#5963).
+fn superseded_join_response(peer: PeerInfo) -> SRPeerJoinResponse {
+    SRPeerJoinResponse {
+        peer,
+        initial_sync_error_message: String::new(),
+        applied: Some(false),
+    }
+}
+
+/// The answer to a join this site committed.
+fn applied_join_response(peer: PeerInfo, initial_sync_error_message: String) -> SRPeerJoinResponse {
+    SRPeerJoinResponse {
+        peer,
+        initial_sync_error_message,
+        applied: Some(true),
+    }
+}
+
 #[async_trait::async_trait]
 impl Operation for SRPeerJoinHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -10275,10 +10438,14 @@ impl Operation for SRPeerJoinHandler {
         let (state, local_peer) = match committed {
             PeerJoinOutcome::Applied(state, local_peer) => (*state, local_peer),
             PeerJoinOutcome::Superseded(peer) => {
-                return json_response(&SRPeerJoinResponse {
-                    peer,
-                    ..Default::default()
-                });
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "join_superseded",
+                    "admin site replication state"
+                );
+                return json_response(&superseded_join_response(peer));
             }
         };
         // Fix 1 (receiving side): ensure the joining peer also sets up replication for any
@@ -10297,10 +10464,10 @@ impl Operation for SRPeerJoinHandler {
                 "admin site replication state"
             );
         }
-        json_response(&SRPeerJoinResponse {
-            peer: state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer),
-            initial_sync_error_message: backfill_errors.render(),
-        })
+        json_response(&applied_join_response(
+            state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer),
+            backfill_errors.render(),
+        ))
     }
 }
 
@@ -11344,7 +11511,11 @@ impl Operation for SRRotateServiceAccountHandler {
             {
                 continue;
             }
-            if let Err(err) = send_peer_admin_request_with_secret_candidates(
+            // A superseded join returns BEFORE `apply_iam`, so a no-op answer
+            // means the peer never installed the new secret. Acking it would
+            // finalize a rotation half the mesh cannot authenticate against
+            // (rustfs/rustfs#5963).
+            let rotation_error = match send_peer_admin_request_with_secret_candidates(
                 &runtime_peer_connection(peer)?,
                 SITE_REPLICATION_PEER_JOIN_PATH,
                 &pending_rotation.access_key,
@@ -11353,7 +11524,20 @@ impl Operation for SRRotateServiceAccountHandler {
             )
             .await
             {
-                let detail = summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint));
+                Err(err) => Some(summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint))),
+                Ok(body) => match parse_peer_join_response(&body, peer.clone()) {
+                    Ok(response) if response.applied == Some(false) => Some(summarize_peer_error_detail(&format!(
+                        "{}: peer did not apply the rotation join (its site replication state is newer than the snapshot it \
+                         was sent); the new service account secret was not installed",
+                        peer.endpoint
+                    ))),
+                    // Unparseable bodies keep the pre-existing behaviour: the
+                    // transport succeeded, and MinIO peers answer with an empty
+                    // body this helper already tolerates.
+                    Ok(_) | Err(_) => None,
+                },
+            };
+            if let Some(detail) = rotation_error {
                 warn!(
                     event = EVENT_ADMIN_SITE_REPLICATION_STATE,
                     component = LOG_COMPONENT_ADMIN,
@@ -15639,9 +15823,17 @@ mod tests {
             site_replication_remove_status(&["peer request to https://remote.example.com failed with 403 Forbidden".to_string()]);
 
         assert!(state.peers.is_empty());
-        assert_eq!(status.status, SITE_REPL_REMOVE_SUCCESS);
+        assert_eq!(
+            status.status, SITE_REPL_REMOVE_PARTIAL,
+            "a removal whose peer could not be notified must not report unqualified success"
+        );
         assert!(status.err_detail.contains("failed to notify 1 peer"));
         assert!(status.err_detail.contains("403 Forbidden"));
+
+        // The fully-notified path stays byte-identical for existing clients.
+        let clean = site_replication_remove_status(&[]);
+        assert_eq!(clean.status, SITE_REPL_REMOVE_SUCCESS);
+        assert!(clean.err_detail.is_empty());
     }
 
     #[test]
@@ -16982,16 +17174,22 @@ mod tests {
             assert_eq!(response.peer.deployment_id, "remote-deployment");
             assert_eq!(response.peer.endpoint, "https://remote.example.com");
             assert!(response.initial_sync_error_message.is_empty());
+            assert_eq!(
+                response.applied, None,
+                "a MinIO empty-body success reports nothing; it must not read as a no-op join"
+            );
         }
 
         let json = serde_json::to_vec(&SRPeerJoinResponse {
             peer: peer("actual", "https://actual.example.com"),
             initial_sync_error_message: "sync failed".to_string(),
+            applied: Some(true),
         })
         .expect("serialize join response");
         let response = parse_peer_join_response(&json, fallback.clone()).expect("parse join response body");
         assert_eq!(response.peer.endpoint, "https://actual.example.com");
         assert_eq!(response.initial_sync_error_message, "sync failed");
+        assert_eq!(response.applied, Some(true));
 
         assert!(parse_peer_join_response(b"not-json", fallback).is_err());
     }
@@ -17712,13 +17910,319 @@ mod tests {
         .expect("parse legacy peer join response");
 
         assert!(response.initial_sync_error_message.is_empty());
+        assert_eq!(
+            response.applied, None,
+            "a peer older than the field says nothing about whether it applied the join"
+        );
 
         let value = serde_json::to_value(SRPeerJoinResponse {
             peer: peer("remote", "https://remote.example.com"),
             initial_sync_error_message: "bucket setup failed".to_string(),
+            applied: Some(true),
         })
         .expect("serialize peer join response");
         assert_eq!(value.get("initialSyncErrorMessage").and_then(Value::as_str), Some("bucket setup failed"));
+        assert_eq!(value.get("applied").and_then(Value::as_bool), Some(true));
+
+        // An unset verdict must not appear on the wire, so a peer that never
+        // learned the field keeps deserializing byte-identical payloads.
+        let value = serde_json::to_value(SRPeerJoinResponse {
+            peer: peer("remote", "https://remote.example.com"),
+            initial_sync_error_message: String::new(),
+            applied: None,
+        })
+        .expect("serialize peer join response");
+        assert!(value.get("applied").is_none(), "an unset verdict must be omitted: {value}");
+    }
+
+    /// rustfs/rustfs#5963: a removal that could not notify its peers leaves
+    /// `pending_remove` set, and that field alone makes `SRPeerBucketOpsHandler`
+    /// reject every peer operation — before it ever consults `enabled()`. A
+    /// later join restored the topology but left the marker, so a "successful"
+    /// re-add produced a cluster that reported Enabled/2-sites on both sides
+    /// while replication stayed dead. The join must clear it.
+    #[test]
+    fn peer_join_clears_a_stuck_pending_remove() {
+        let local = PeerInfo {
+            deployment_id: "site-b".to_string(),
+            ..peer("site-b", "https://site-b.example.com")
+        };
+        let remote = PeerInfo {
+            deployment_id: "site-a".to_string(),
+            ..peer("site-a", "https://site-a.example.com")
+        };
+        let mut state = SiteReplicationState {
+            peers: BTreeMap::from([(local.deployment_id.clone(), local.clone())]),
+            pending_remove: Some(PendingRemove {
+                id: "stuck-remove".to_string(),
+                req: SRRemoveReq {
+                    remove_all: true,
+                    ..Default::default()
+                },
+                service_account_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+                secret_candidates: Vec::new(),
+                original_peers: BTreeMap::from([
+                    (local.deployment_id.clone(), local.clone()),
+                    (remote.deployment_id.clone(), remote.clone()),
+                ]),
+                acked_deployment_ids: BTreeSet::new(),
+                updated_at: Some(OffsetDateTime::now_utc()),
+            }),
+            ..Default::default()
+        };
+
+        apply_peer_join(
+            &mut state,
+            &local,
+            SRPeerJoinReq {
+                svc_acct_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+                svc_acct_secret_key: "svc-secret".to_string(),
+                svc_acct_parent: "root".to_string(),
+                peers: BTreeMap::from([
+                    (local.deployment_id.clone(), local.clone()),
+                    (remote.deployment_id.clone(), remote),
+                ]),
+                updated_at: Some(OffsetDateTime::now_utc()),
+            },
+            false,
+        );
+
+        assert!(
+            state.pending_remove.is_none(),
+            "an accepted join supersedes the half-finished removal it lands on"
+        );
+        assert!(state.enabled(), "the join restores the two-site topology");
+        // The guard `SRPeerBucketOpsHandler` evaluates, asserted directly: with
+        // the marker cleared and the topology back, peer bucket-ops are
+        // admitted again.
+        assert!(
+            state.pending_remove.is_none() && state.enabled(),
+            "the bucket-ops admission predicate must now pass"
+        );
+    }
+
+    /// The fence marks are lifecycle-independent and must survive the clearing
+    /// above — wiping them would reopen the rollback window the fence closes.
+    #[test]
+    fn peer_join_clearing_pending_remove_keeps_edit_generation_marks() {
+        let local = PeerInfo {
+            deployment_id: "site-b".to_string(),
+            ..peer("site-b", "https://site-b.example.com")
+        };
+        let remote = PeerInfo {
+            deployment_id: "site-a".to_string(),
+            ..peer("site-a", "https://site-a.example.com")
+        };
+        let mut state = SiteReplicationState {
+            peers: BTreeMap::from([(local.deployment_id.clone(), local.clone())]),
+            applied_edit_generations: BTreeMap::from([(remote.deployment_id.clone(), 7)]),
+            pending_remove: Some(PendingRemove {
+                id: "stuck-remove".to_string(),
+                req: SRRemoveReq {
+                    remove_all: true,
+                    ..Default::default()
+                },
+                service_account_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+                secret_candidates: Vec::new(),
+                original_peers: BTreeMap::from([
+                    (local.deployment_id.clone(), local.clone()),
+                    (remote.deployment_id.clone(), remote.clone()),
+                ]),
+                acked_deployment_ids: BTreeSet::new(),
+                updated_at: Some(OffsetDateTime::now_utc()),
+            }),
+            ..Default::default()
+        };
+
+        apply_peer_join(
+            &mut state,
+            &local,
+            SRPeerJoinReq {
+                svc_acct_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+                svc_acct_secret_key: "svc-secret".to_string(),
+                svc_acct_parent: "root".to_string(),
+                peers: BTreeMap::from([
+                    (local.deployment_id.clone(), local.clone()),
+                    (remote.deployment_id.clone(), remote.clone()),
+                ]),
+                updated_at: Some(OffsetDateTime::now_utc()),
+            },
+            false,
+        );
+
+        assert!(state.pending_remove.is_none());
+        assert_eq!(
+            state.applied_edit_generations.get(&remote.deployment_id),
+            Some(&7),
+            "clearing the lifecycle marker must not touch the ordering fence"
+        );
+    }
+
+    /// rustfs/rustfs#5963: the two join verdicts must be distinguishable on the
+    /// wire. `Some(true)`/`Some(false)` is what lets the initiator tell a real
+    /// configuration from a 200 that wrote nothing; flipping either one back to
+    /// an unset verdict re-hides the no-op.
+    #[test]
+    fn join_verdicts_are_distinguishable_on_the_wire() {
+        let remote = peer("remote", "https://remote.example.com");
+
+        let superseded = superseded_join_response(remote.clone());
+        assert_eq!(
+            superseded.applied,
+            Some(false),
+            "a join this site refused to apply must say so explicitly"
+        );
+        assert!(superseded.initial_sync_error_message.is_empty());
+
+        let applied = applied_join_response(remote, "bucket setup failed".to_string());
+        assert_eq!(applied.applied, Some(true));
+        assert_eq!(applied.initial_sync_error_message, "bucket setup failed");
+
+        // Round-tripping through the wire keeps the two apart — the initiator
+        // only ever sees the serialized form.
+        let decoded: SRPeerJoinResponse =
+            serde_json::from_slice(&serde_json::to_vec(&superseded_join_response(peer("r", "https://r.example.com"))).unwrap())
+                .expect("round-trip superseded verdict");
+        assert_eq!(decoded.applied, Some(false));
+    }
+
+    /// rustfs/rustfs#5963: a stuck removal must be visible on the endpoint
+    /// operators actually run. `replicate info` used to report only
+    /// `enabled: false`, which reads as "never configured" rather than "a
+    /// removal is wedged here and this site rejects every peer operation".
+    #[test]
+    fn site_replication_info_reports_a_wedged_removal() {
+        let local = PeerInfo {
+            deployment_id: "site-b".to_string(),
+            ..peer("site-b", "https://site-b.example.com")
+        };
+        let remote = PeerInfo {
+            deployment_id: "site-a".to_string(),
+            ..peer("site-a", "https://site-a.example.com")
+        };
+        let state = SiteReplicationState {
+            name: "site-b".to_string(),
+            peers: BTreeMap::from([(local.deployment_id.clone(), local.clone())]),
+            pending_remove: Some(PendingRemove {
+                id: "stuck-remove".to_string(),
+                req: SRRemoveReq {
+                    remove_all: true,
+                    ..Default::default()
+                },
+                service_account_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+                secret_candidates: Vec::new(),
+                original_peers: BTreeMap::from([
+                    (local.deployment_id.clone(), local.clone()),
+                    (remote.deployment_id.clone(), remote.clone()),
+                ]),
+                acked_deployment_ids: BTreeSet::new(),
+                updated_at: Some(OffsetDateTime::now_utc()),
+            }),
+            ..Default::default()
+        };
+
+        let info = site_replication_info_for(&state, &local);
+        assert!(!info.enabled, "the peer set is already torn down");
+        let pending = info
+            .pending_operation
+            .as_ref()
+            .expect("a wedged removal must surface as a pending operation");
+        assert_eq!(pending.operation, "remove");
+        assert!(
+            pending.pending_peers.contains(&remote.deployment_id),
+            "the peer that was never notified must be named: {pending:?}"
+        );
+    }
+
+    /// The source side of the same failure: peer operations are being rejected,
+    /// the topology still looks like a healthy two-site cluster, and `info` has
+    /// to say the deliveries are failing.
+    #[test]
+    fn site_replication_info_reports_failing_peer_deliveries() {
+        let local = PeerInfo {
+            deployment_id: "site-a".to_string(),
+            ..peer("site-a", "https://site-a.example.com")
+        };
+        let remote = PeerInfo {
+            deployment_id: "site-b".to_string(),
+            ..peer("site-b", "https://site-b.example.com")
+        };
+        let state = SiteReplicationState {
+            name: "site-a".to_string(),
+            peers: BTreeMap::from([
+                (local.deployment_id.clone(), local.clone()),
+                (remote.deployment_id.clone(), remote.clone()),
+            ]),
+            retry_queue: vec![SiteReplicationRetryEvent {
+                id: "evt".to_string(),
+                peer_deployment_id: remote.deployment_id.clone(),
+                peer_endpoint: remote.endpoint,
+                path: "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=demo&operation=make-with-versioning".to_string(),
+                retry_count: 9,
+                failed: true,
+                last_error: "site replication is not enabled".to_string(),
+                updated_at: Some(OffsetDateTime::now_utc()),
+                edit_generation: None,
+            }],
+            ..Default::default()
+        };
+
+        let info = site_replication_info_for(&state, &local);
+        assert!(info.enabled, "the topology still reports two sites — that was the trap");
+        let stats = info
+            .retry_stats
+            .as_ref()
+            .expect("a peer rejecting every delivery must be visible in `info`");
+        assert_eq!(stats.failed, 1);
+        assert_eq!(stats.last_error, "site replication is not enabled");
+
+        // A healthy site must stay wire-identical to before the field existed.
+        let healthy = SiteReplicationState {
+            retry_queue: Vec::new(),
+            ..state
+        };
+        let info = site_replication_info_for(&healthy, &local);
+        assert!(info.retry_stats.is_none());
+        assert!(info.pending_operation.is_none());
+    }
+
+    /// rustfs/rustfs#5963: `replicate info` reported a healthy cluster while
+    /// every peer operation was failing. The health it used to omit now rides
+    /// along, and a healthy site still serializes without the new fields.
+    #[test]
+    fn site_replication_info_health_fields_are_absent_when_healthy() {
+        let healthy = SiteReplicationInfo {
+            enabled: true,
+            name: "site-a".to_string(),
+            sites: vec![peer("site-a", "https://site-a.example.com")],
+            service_account_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+            api_version: Some(SITE_REPL_API_VERSION.to_string()),
+            retry_stats: None,
+            pending_operation: None,
+        };
+        let value = serde_json::to_value(&healthy).expect("serialize info");
+        assert!(value.get("retryStats").is_none(), "a healthy site must not grow fields: {value}");
+        assert!(value.get("pendingOperation").is_none(), "a healthy site must not grow fields: {value}");
+
+        let degraded = SiteReplicationInfo {
+            retry_stats: Some(SRRetryStats {
+                pending: 1,
+                failed: 4,
+                last_error: "site replication is not enabled".to_string(),
+                api_version: Some(SITE_REPL_API_VERSION.to_string()),
+            }),
+            ..healthy
+        };
+        let value = serde_json::to_value(&degraded).expect("serialize info");
+        assert_eq!(
+            value.pointer("/retryStats/failed").and_then(Value::as_u64),
+            Some(4),
+            "a source site whose peer rejects everything must say so in `info`"
+        );
+        assert_eq!(
+            value.pointer("/retryStats/lastError").and_then(Value::as_str),
+            Some("site replication is not enabled")
+        );
     }
 
     // Fix 5: remove --all must purge local state unconditionally even when peer errors occur
@@ -17766,12 +18270,14 @@ mod tests {
         assert!(state.peers.is_empty(), "peers must be cleared on remove --all");
         assert!(state.resync_status.is_empty(), "resync_status must be cleared on remove --all");
 
-        // Even if peers returned 403 (desynced account), status still reports success
+        // The local side is torn down either way, but a peer that returned 403
+        // (desynced account) leaves the cluster diverged — the response must
+        // say so instead of reporting unqualified success (rustfs/rustfs#5963).
         let status =
             site_replication_remove_status(&["https://remote.example.com: peer/remove returned 403 Forbidden".to_string()]);
         assert_eq!(
-            status.status, SITE_REPL_REMOVE_SUCCESS,
-            "local remove reports success even when peer notifications fail"
+            status.status, SITE_REPL_REMOVE_PARTIAL,
+            "local remove must report a partial result when peer notifications fail"
         );
         assert!(
             status.err_detail.contains("403 Forbidden"),

rustfs/src/storage/concurrency/io_schedule.rs

@@ -1766,9 +1766,11 @@ mod tests {
     };
     use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
     use rustfs_io_metrics::bandwidth::{BandwidthSnapshot, BandwidthTier};
+    use serial_test::serial;
     use std::time::Duration;
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_basic() {
         let config = IoPriorityQueueConfig::default();
         let queue = IoPriorityQueue::new(config);
@@ -1787,6 +1789,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_dequeue_order() {
         let config = IoPriorityQueueConfig::default();
         let queue = IoPriorityQueue::new(config);
@@ -1814,6 +1817,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_status() {
         let config = IoPriorityQueueConfig::default();
         let queue = IoPriorityQueue::new(config);
@@ -1831,6 +1835,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_starvation_prevention() {
         let config = IoPriorityQueueConfig {
             starvation_threshold_secs: 1,
@@ -1854,6 +1859,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_from_size() {
         // High priority: < 1MB
         assert_eq!(IoPriority::from_size(100 * 1024), IoPriority::High);
@@ -1869,6 +1875,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_load_level_from_wait_duration() {
         use std::time::Duration;
 
@@ -1886,6 +1893,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_scheduler_config_default() {
         let config = IoSchedulerConfig::default();
 
@@ -1899,6 +1907,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_scheduler_config_to_core_config() {
         let config = IoSchedulerConfig::default();
         let core = config.to_core_config();
@@ -1914,6 +1923,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_config_to_core_config() {
         let config = IoPriorityQueueConfig::default();
         let core = config.to_core_config();
@@ -1925,6 +1935,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_config_from_scheduler_config() {
         let scheduler_config = IoSchedulerConfig {
             queue_high_capacity: 128,
@@ -1947,6 +1958,7 @@ mod tests {
     // ============================================
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_nvme_sequential_low_load() {
         // NVMe + Sequential + Low load = maximum buffer size
         let context = IoSchedulingContext {
@@ -1973,6 +1985,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_hdd_random_high_load() {
         // HDD + Random + High load = conservative buffer size
         let context = IoSchedulingContext {
@@ -1999,6 +2012,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_ssd_mixed_medium_load() {
         // SSD + Mixed + Medium load = moderate buffer
         let context = IoSchedulingContext {
@@ -2026,6 +2040,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_critical_load_disables_features() {
         // Any media + Critical load = minimal features
         let context = IoSchedulingContext {
@@ -2050,6 +2065,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_buffer_cap_enforcement() {
         // Test that storage media caps are enforced
         let context = IoSchedulingContext {
@@ -2074,6 +2090,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_applies_sequential_hint_when_pattern_unknown() {
         let context = IoSchedulingContext {
             file_size: 2 * 1024 * 1024 * 1024,              // 2GiB
@@ -2098,6 +2115,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_bandwidth_low_reduces_buffer() {
         // Low bandwidth should reduce buffer
         let context = IoSchedulingContext {
@@ -2121,6 +2139,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_high_concurrency_reduction() {
         // High concurrency should reduce buffer
         let context = IoSchedulingContext {
@@ -2143,6 +2162,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_sequential_boost() {
         // Sequential reads should get boost
         let sequential_context = IoSchedulingContext {
@@ -2184,6 +2204,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_unknown_media_conservative() {
         // Unknown media should be conservative
         let context = IoSchedulingContext {
@@ -2209,6 +2230,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_priority_classification() {
         // Test priority classification based on file size
         let small_context = IoSchedulingContext {
@@ -2255,6 +2277,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_readahead_decision_matrix() {
         // Test readahead enable/disable logic
         let configs = vec![
@@ -2340,6 +2363,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_buffer_multiplier_stages() {
         // Test that all multiplier stages are applied
         let context = IoSchedulingContext {
@@ -2374,6 +2398,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_compatibility_path() {
         // Test that compatibility path (from_wait_duration) still works
         let wait_duration = Duration::from_millis(50);

scripts/test/site_replication_smoke.py

@@ -27,6 +27,7 @@ Usage:
     ./scripts/test/site_replication_smoke.py            # up: start both + pair
     ./scripts/test/site_replication_smoke.py status     # process + pair status
     ./scripts/test/site_replication_smoke.py smoke      # bidirectional object check
+    ./scripts/test/site_replication_smoke.py diverge    # rustfs/rustfs#5963 regression
     ./scripts/test/site_replication_smoke.py logs       # tail both server logs
     ./scripts/test/site_replication_smoke.py down       # stop both processes
     ./scripts/test/site_replication_smoke.py clean      # down + wipe site data
@@ -337,11 +338,12 @@ def ensure_pair(site_a: Site, site_b: Site) -> None:
     print(f"[ok] site replication configured: {result.get('status', '')}")
 
 
-def remove_pair(site: Site) -> None:
+def remove_pair(site: Site) -> dict:
     status, body = admin(site, "PUT", "site-replication/remove", payload={"all": True})
     if status != 200:
         raise SystemExit(f"[fail] site-replication remove: HTTP {status} {body.decode(errors='replace')}")
     print(f"[ok] site replication removed: {body.decode(errors='replace')}")
+    return json.loads(body)
 
 
 # ---------------------------------------------------------------------------
@@ -397,6 +399,112 @@ def smoke(site_a: Site, site_b: Site, timeout: float) -> None:
     print(f"[ok] bidirectional replication verified via bucket {bucket}")
 
 
+# ---------------------------------------------------------------------------
+# Divergence regression (rustfs/rustfs#5963)
+# ---------------------------------------------------------------------------
+
+
+def wait_for(description: str, probe, timeout: float):
+    """Poll `probe` until it returns a truthy value; return it. SystemExit on timeout."""
+    deadline = time.monotonic() + timeout
+    last = None
+    while time.monotonic() < deadline:
+        try:
+            result = probe()
+        except (urllib.error.URLError, OSError, TimeoutError, SystemExit) as err:
+            last = err
+            result = None
+        if result:
+            return result
+        time.sleep(1.0)
+    raise SystemExit(f"[fail] {description} within {timeout:.0f}s (last: {last})")
+
+
+def diverge(site_a: Site, site_b: Site, binary: Path, console: bool, timeout: float) -> None:
+    """Reproduce rustfs/rustfs#5963 end to end and assert the cluster recovers.
+
+    Before the fix, step 7 left site-b rejecting every peer bucket-op forever:
+    `pending_remove` gates `SRPeerBucketOpsHandler` ahead of `enabled()`, and a
+    join never cleared it — so a *successful* re-add produced a cluster that
+    reported Enabled/2-sites on both sides while replication stayed dead.
+    """
+    ensure_pair(site_a, site_b)
+
+    # 1. Take site-a down so it cannot be told about the removal.
+    print("[..] step 1: stopping site-a so it cannot be notified")
+    stop_site(site_a)
+
+    # 2. Remove from site-b. The local teardown commits either way, but the
+    #    response must NOT claim unqualified success (P2-5).
+    print("[..] step 2: removing site replication from site-b while site-a is down")
+    status = remove_pair(site_b)
+    if not status.get("errorDetail"):
+        raise SystemExit(f"[fail] remove hid the unreachable peer; expected errorDetail: {json.dumps(status)}")
+    if status.get("status") == "Requested site(s) were removed from cluster replication successfully.":
+        raise SystemExit(f"[fail] remove reported unqualified success despite an unnotified peer: {json.dumps(status)}")
+    print(f"[ok] remove reported a partial result: status={status.get('status')!r}")
+
+    # 3. The wedged removal must be visible on `info`, not just in status --json (P1-4).
+    info_b = pair_state(site_b)
+    pending = info_b.get("pendingOperation")
+    if not pending or pending.get("operation") != "remove":
+        raise SystemExit(f"[fail] site-b hides the wedged removal in `info`: {json.dumps(info_b, indent=2)}")
+    print(f"[ok] site-b reports the wedged removal: pendingPeers={pending.get('pendingPeers')}")
+
+    # 4. Bring site-a back. It still believes in a healthy 2-site cluster.
+    print("[..] step 4: restarting site-a")
+    start_site(site_a, binary, console)
+    wait_ready([site_a], timeout)
+    info_a = pair_state(site_a)
+    if not info_a.get("enabled"):
+        raise SystemExit(f"[fail] site-a lost its own state: {json.dumps(info_a, indent=2)}")
+    print("[ok] site-a still reports an enabled cluster (the divergence)")
+
+    # 5. A bucket created on site-a cannot reach site-b. The failure must become
+    #    visible on the SOURCE, which used to report a perfectly healthy cluster.
+    bucket = f"sr-diverge-{uuid.uuid4().hex[:8]}"
+    sig_status, body = signed_request(site_a, "PUT", f"/{bucket}")
+    if sig_status != 200:
+        raise SystemExit(f"[fail] create bucket {bucket} on site-a: HTTP {sig_status} {body.decode(errors='replace')}")
+    print(f"[ok] created {bucket} on site-a (locally succeeds, peer push is rejected)")
+
+    stats = wait_for(
+        "site-a did not surface the failing peer deliveries in `info`",
+        lambda: pair_state(site_a).get("retryStats"),
+        timeout,
+    )
+    print(f"[ok] site-a reports failing deliveries: pending={stats.get('pending')} failed={stats.get('failed')} "
+          f"lastError={stats.get('lastError')!r}")
+
+    # 6. Re-add. This is the operator's natural recovery move.
+    print("[..] step 6: re-adding the pair from site-a")
+    peers = [
+        {"name": s.name, "endpoints": s.endpoint, "accessKey": s.access_key, "secretKey": s.secret_key}
+        for s in (site_a, site_b)
+    ]
+    add_status, add_body = admin(site_a, "PUT", "site-replication/add", "replicateILMExpiry=false", peers)
+    if add_status != 200:
+        raise SystemExit(f"[fail] re-add: HTTP {add_status} {add_body.decode(errors='replace')}")
+    print(f"[ok] re-add accepted: {add_body.decode(errors='replace')}")
+
+    # 7. The join must have cleared site-b's pending_remove (P0-1). Without the
+    #    fix this assertion is exactly what fails while everything above passes.
+    info_b = pair_state(site_b)
+    if info_b.get("pendingOperation"):
+        raise SystemExit(
+            "[fail] the join did not clear site-b's wedged removal; peer bucket-ops stay rejected forever: "
+            f"{json.dumps(info_b, indent=2)}"
+        )
+    if not info_b.get("enabled"):
+        raise SystemExit(f"[fail] site-b did not rejoin: {json.dumps(info_b, indent=2)}")
+    print("[ok] site-b cleared the wedged removal and rejoined")
+
+    # 8. The symptom the issue actually reported: replication works again.
+    print("[..] step 8: verifying replication actually flows again")
+    smoke(site_a, site_b, timeout)
+    print("[ok] rustfs/rustfs#5963 regression passed")
+
+
 # ---------------------------------------------------------------------------
 # Commands
 # ---------------------------------------------------------------------------
@@ -464,7 +572,7 @@ def main() -> None:
         "command",
         nargs="?",
         default="up",
-        choices=["up", "down", "restart", "status", "logs", "smoke", "info", "remove", "clean"],
+        choices=["up", "down", "restart", "status", "logs", "smoke", "diverge", "info", "remove", "clean"],
     )
     parser.add_argument("--port-a", type=int, default=9000, help="site A S3 port (default: %(default)s)")
     parser.add_argument("--port-b", type=int, default=9020, help="site B S3 port (default: %(default)s)")
@@ -495,6 +603,8 @@ def main() -> None:
         cmd_logs(sites, args.lines)
     elif args.command == "smoke":
         smoke(site_a, site_b, args.timeout)
+    elif args.command == "diverge":
+        diverge(site_a, site_b, args.binary, args.console, args.timeout)
     elif args.command == "info":
         print(json.dumps(pair_state(site_a), indent=2, ensure_ascii=False))
     elif args.command == "remove":