refactor(data-usage): own SizeSummary once, with the scanner's semantics (#6237)

`SizeSummary` and `ReplTargetSizeSummary` existed in both `rustfs-data-usage` and `rustfs-scanner`, and the two copies had drifted three ways: four size fields were `usize` in one and `i64` in the other, only the scanner's carried `tier_stats`, and — the difference that matters — the scanner's `add` saturated while the data-usage copy used plain `+=`, which panics on overflow in a debug build and wraps in a release one.

The data-usage copy is now the only definition and takes the scanner's shape and semantics, since that is the side a test already pinned (`MAX + 1 == MAX`). An equivalent saturation test now guards it in its new home. The scanner re-exports both types alongside the ones it already re-exported.

`DataUsageEntry::add_sizes` and `BucketUsageInfo::add_size_summary` are removed. Both took a `SizeSummary` and had no callers anywhere — they were the duplicate fold paths, and `apply_scanner_size_summary` is now the only one.

`actions_accounting` stays in the scanner as the `ScannerSizeSummaryExt` extension trait: it needs `ObjectInfo`, which sits above `rustfs-data-usage`, and an inherent impl on a foreign type is not allowed. The three call sites are unchanged.

Refs backlog#1828

.github/workflows/package.yml

@@ -522,10 +522,16 @@ jobs:
             for f in "$DEB_FILE" "$RPM_FILE"; do
               if [[ -n "$f" && -f "$f" ]]; then
                 base="$(basename "$f")"
-                # Remove any stale entry, then append the fresh digest
+                # GitHub stores release asset names with '~' normalized to '.'
+                # (e.g. rustfs_1.0.0~rc.2_amd64.deb is stored as
+                # rustfs_1.0.0.rc.2_amd64.deb), so checksum entries must
+                # reference the name as stored on the release.
+                github_base="${base//\~/.}"
+                # Remove any stale entry (both naming variants), then append
                 grep -Fv -- "$base" "$checksum_file" > "${checksum_file}.tmp" || true
-                mv "${checksum_file}.tmp" "$checksum_file"
-                (cd "$(dirname "$f")" && "$checksum_cmd" -- "$base") >> "$checksum_file"
+                grep -Fv -- "$github_base" "${checksum_file}.tmp" > "${checksum_file}.tmp2" || true
+                mv "${checksum_file}.tmp2" "$checksum_file"
+                (cd "$(dirname "$f")" && "$checksum_cmd" -- "$github_base") >> "$checksum_file"
               fi
             done
 

crates/data-usage/src/data_usage.rs

@@ -317,15 +317,15 @@ pub struct SizeSummary {
     /// Number of delete markers
     pub delete_markers: usize,
     /// Replicated size
-    pub replicated_size: usize,
+    pub replicated_size: i64,
     /// Replicated count
     pub replicated_count: usize,
     /// Pending size
-    pub pending_size: usize,
+    pub pending_size: i64,
     /// Failed size
-    pub failed_size: usize,
+    pub failed_size: i64,
     /// Replica size
-    pub replica_size: usize,
+    pub replica_size: i64,
     /// Replica count
     pub replica_count: usize,
     /// Pending count
@@ -334,19 +334,21 @@ pub struct SizeSummary {
     pub failed_count: usize,
     /// Replication target stats
     pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
+    /// Per-tier accounting, keyed by storage class or remote tier name
+    pub tier_stats: HashMap<String, TierStats>,
 }
 
 /// Replication target size summary
 #[derive(Debug, Default, Clone)]
 pub struct ReplTargetSizeSummary {
     /// Replicated size
-    pub replicated_size: usize,
+    pub replicated_size: i64,
     /// Replicated count
     pub replicated_count: usize,
     /// Pending size
-    pub pending_size: usize,
+    pub pending_size: i64,
     /// Failed size
-    pub failed_size: usize,
+    pub failed_size: i64,
     /// Pending count
     pub pending_count: usize,
     /// Failed count
@@ -710,28 +712,6 @@ impl DataUsageEntry {
         self.children.insert(hash.key());
     }
 
-    pub fn add_sizes(&mut self, summary: &SizeSummary) {
-        self.size += summary.total_size;
-        self.versions += summary.versions;
-        self.delete_markers += summary.delete_markers;
-        self.obj_sizes.add(summary.total_size as u64);
-        self.obj_versions.add(summary.versions as u64);
-
-        let replication_stats = self.replication_stats.get_or_insert_with(ReplicationAllStats::default);
-        replication_stats.replica_size += summary.replica_size as u64;
-        replication_stats.replica_count += summary.replica_count as u64;
-
-        for (arn, st) in &summary.repl_target_stats {
-            let tgt_stat = replication_stats.targets.entry(arn.to_string()).or_default();
-            tgt_stat.pending_size += st.pending_size as u64;
-            tgt_stat.failed_size += st.failed_size as u64;
-            tgt_stat.replicated_size += st.replicated_size as u64;
-            tgt_stat.replicated_count += st.replicated_count as u64;
-            tgt_stat.failed_count += st.failed_count as u64;
-            tgt_stat.pending_count += st.pending_count as u64;
-        }
-    }
-
     pub fn merge(&mut self, other: &DataUsageEntry) {
         self.objects += other.objects;
         self.versions += other.versions;
@@ -1722,14 +1702,6 @@ impl BucketUsageInfo {
     }
 
     /// Add size summary to this bucket usage
-    pub fn add_size_summary(&mut self, summary: &SizeSummary) {
-        self.size += summary.total_size as u64;
-        self.versions_count += summary.versions as u64;
-        self.delete_markers_count += summary.delete_markers as u64;
-        self.replica_size += summary.replica_size as u64;
-        self.replica_count += summary.replica_count as u64;
-    }
-
     /// Merge another BucketUsageInfo into this one
     pub fn merge(&mut self, other: &BucketUsageInfo) {
         self.size += other.size;
@@ -1775,29 +1747,32 @@ impl SizeSummary {
         Self::default()
     }
 
-    /// Add another SizeSummary to this one
+    /// Add another SizeSummary to this one.
+    ///
+    /// Saturating throughout: a scan that overflows a counter should report the
+    /// ceiling rather than panic in a debug build or wrap in a release one.
     pub fn add(&mut self, other: &SizeSummary) {
-        self.total_size += other.total_size;
-        self.versions += other.versions;
-        self.delete_markers += other.delete_markers;
-        self.replicated_size += other.replicated_size;
-        self.replicated_count += other.replicated_count;
-        self.pending_size += other.pending_size;
-        self.failed_size += other.failed_size;
-        self.replica_size += other.replica_size;
-        self.replica_count += other.replica_count;
-        self.pending_count += other.pending_count;
-        self.failed_count += other.failed_count;
+        self.total_size = self.total_size.saturating_add(other.total_size);
+        self.versions = self.versions.saturating_add(other.versions);
+        self.delete_markers = self.delete_markers.saturating_add(other.delete_markers);
+        self.replicated_size = self.replicated_size.saturating_add(other.replicated_size);
+        self.replicated_count = self.replicated_count.saturating_add(other.replicated_count);
+        self.pending_size = self.pending_size.saturating_add(other.pending_size);
+        self.failed_size = self.failed_size.saturating_add(other.failed_size);
+        self.replica_size = self.replica_size.saturating_add(other.replica_size);
+        self.replica_count = self.replica_count.saturating_add(other.replica_count);
+        self.pending_count = self.pending_count.saturating_add(other.pending_count);
+        self.failed_count = self.failed_count.saturating_add(other.failed_count);
 
         // Merge replication target stats
         for (target, stats) in &other.repl_target_stats {
             let entry = self.repl_target_stats.entry(target.clone()).or_default();
-            entry.replicated_size += stats.replicated_size;
-            entry.replicated_count += stats.replicated_count;
-            entry.pending_size += stats.pending_size;
-            entry.failed_size += stats.failed_size;
-            entry.pending_count += stats.pending_count;
-            entry.failed_count += stats.failed_count;
+            entry.replicated_size = entry.replicated_size.saturating_add(stats.replicated_size);
+            entry.replicated_count = entry.replicated_count.saturating_add(stats.replicated_count);
+            entry.pending_size = entry.pending_size.saturating_add(stats.pending_size);
+            entry.failed_size = entry.failed_size.saturating_add(stats.failed_size);
+            entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
+            entry.failed_count = entry.failed_count.saturating_add(stats.failed_count);
         }
     }
 }
@@ -2343,6 +2318,64 @@ mod tests {
         assert_eq!(usage1.versions_count, 15);
     }
 
+    #[test]
+    fn size_summary_add_saturates_instead_of_overflowing() {
+        // The scanner folds one summary per object into a per-prefix total, so a
+        // counter at its ceiling must stay there rather than panic in a debug
+        // build or wrap in a release one (backlog#1828).
+        let mut summary = SizeSummary {
+            total_size: usize::MAX,
+            versions: usize::MAX,
+            replicated_size: i64::MAX,
+            pending_size: i64::MAX,
+            failed_size: i64::MAX,
+            replica_size: i64::MAX,
+            ..Default::default()
+        };
+        summary.repl_target_stats.insert(
+            "arn".to_string(),
+            ReplTargetSizeSummary {
+                replicated_size: i64::MAX,
+                pending_size: i64::MAX,
+                failed_size: i64::MAX,
+                ..Default::default()
+            },
+        );
+
+        let mut increment = SizeSummary {
+            total_size: 1,
+            versions: 1,
+            replicated_size: 1,
+            pending_size: 1,
+            failed_size: 1,
+            replica_size: 1,
+            ..Default::default()
+        };
+        increment.repl_target_stats.insert(
+            "arn".to_string(),
+            ReplTargetSizeSummary {
+                replicated_size: 1,
+                pending_size: 1,
+                failed_size: 1,
+                ..Default::default()
+            },
+        );
+
+        summary.add(&increment);
+
+        assert_eq!(summary.total_size, usize::MAX);
+        assert_eq!(summary.versions, usize::MAX);
+        assert_eq!(summary.replicated_size, i64::MAX);
+        assert_eq!(summary.pending_size, i64::MAX);
+        assert_eq!(summary.failed_size, i64::MAX);
+        assert_eq!(summary.replica_size, i64::MAX);
+
+        let target = summary.repl_target_stats.get("arn").expect("target survives the merge");
+        assert_eq!(target.replicated_size, i64::MAX);
+        assert_eq!(target.pending_size, i64::MAX);
+        assert_eq!(target.failed_size, i64::MAX);
+    }
+
     #[test]
     fn test_size_summary_add() {
         let mut summary1 = SizeSummary::new();

crates/ecstore/src/data_usage/mod.rs

@@ -2026,8 +2026,17 @@ enum DataUsageCacheRead {
 
 /// True when the error means the cache object does not exist, as opposed to a
 /// transient failure that is worth another attempt.
+///
+/// `SetDisks::get_object_reader` runs its failures through `to_object_err`,
+/// which rewrites `FileNotFound` to `ObjectNotFound` and `VolumeNotFound` to
+/// `BucketNotFound`, so those are the variants that actually arrive here. The
+/// raw pair is matched too because callers reading through a different layer
+/// can still surface it.
 fn is_data_usage_cache_absent(err: &Error) -> bool {
-    matches!(err, Error::FileNotFound | Error::VolumeNotFound)
+    matches!(
+        err,
+        Error::FileNotFound | Error::VolumeNotFound | Error::ObjectNotFound(..) | Error::BucketNotFound(..)
+    )
 }
 
 async fn read_data_usage_cache_object<S>(store: &S, key: &str) -> crate::error::Result<DataUsageCacheRead>
@@ -2511,7 +2520,10 @@ mod tests {
                 *remaining -= 1;
                 return Err(Error::other("transient read failure"));
             }
-            Err(Error::FileNotFound)
+            // `SetDisks::get_object_reader` reports a missing object through
+            // `to_object_err`, so the absence that reaches the caller is
+            // `ObjectNotFound`, not the raw `FileNotFound`.
+            Err(Error::ObjectNotFound(RUSTFS_META_BUCKET.to_string(), object.to_string()))
         }
 
         async fn put_object(
@@ -2533,6 +2545,23 @@ mod tests {
             .to_string()
     }
 
+    #[test]
+    fn data_usage_cache_absence_covers_the_variants_that_actually_arrive() {
+        // `to_object_err` rewrites the raw storage variants before they reach
+        // `load_data_usage_cache`; classifying only the raw pair would treat a
+        // missing cache as a transient failure and retry it.
+        assert!(is_data_usage_cache_absent(&Error::ObjectNotFound(
+            "bucket".to_string(),
+            "object".to_string()
+        )));
+        assert!(is_data_usage_cache_absent(&Error::BucketNotFound("bucket".to_string())));
+        assert!(is_data_usage_cache_absent(&Error::FileNotFound));
+        assert!(is_data_usage_cache_absent(&Error::VolumeNotFound));
+
+        assert!(!is_data_usage_cache_absent(&Error::other("transient read failure")));
+        assert!(!is_data_usage_cache_absent(&Error::DiskNotFound));
+    }
+
     #[tokio::test]
     async fn load_data_usage_cache_treats_absence_as_an_empty_cache_without_retrying() {
         let name = "usage-cache";

crates/io-metrics/src/internode_metrics.rs

@@ -916,7 +916,7 @@ fn cluster_peer_health_keys() -> Vec<String> {
 mod tests {
     use super::*;
     use metrics::with_local_recorder;
-    use metrics_util::debugging::DebuggingRecorder;
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
     use std::collections::{HashMap, HashSet};
 
     #[test]
@@ -1308,11 +1308,47 @@ mod tests {
     }
 
     #[test]
-    fn msgpack_json_fallback_counter_records_without_panicking() {
-        // Smoke test: the counter accepts both directions and a static message label.
+    fn msgpack_json_fallback_counter_separates_the_two_directions() {
+        // Previously a smoke test that asserted nothing; the counter carries no
+        // in-struct total, so the emission itself is what has to be checked
+        // (rustfs/backlog#1836).
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
         let metrics = InternodeMetrics::default();
-        metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_REQUEST, "FileInfo");
-        metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo");
+
+        metrics::with_local_recorder(&recorder, || {
+            metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_REQUEST, "FileInfo");
+            metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo");
+        });
+
+        let observed: Vec<(String, String, u64)> = snapshotter
+            .snapshot()
+            .into_vec()
+            .into_iter()
+            .filter(|(composite, _, _, _)| composite.key().name() == INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL)
+            .map(|(composite, _, _, value)| {
+                let labels: HashMap<String, String> = composite
+                    .key()
+                    .labels()
+                    .map(|label| (label.key().to_string(), label.value().to_string()))
+                    .collect();
+                let count = match value {
+                    DebugValue::Counter(count) => count,
+                    other => panic!("fallback total must be a counter, got {other:?}"),
+                };
+                (
+                    labels.get(DIRECTION_LABEL).cloned().unwrap_or_default(),
+                    labels.get(MESSAGE_LABEL).cloned().unwrap_or_default(),
+                    count,
+                )
+            })
+            .collect();
+
+        // Each direction/message pair is its own series, so a regression that
+        // dropped a label would collapse these into one row.
+        assert_eq!(observed.len(), 2, "each direction must land in its own series: {observed:?}");
+        assert!(observed.contains(&(INTERNODE_MSGPACK_DIRECTION_REQUEST.to_string(), "FileInfo".to_string(), 1)));
+        assert!(observed.contains(&(INTERNODE_MSGPACK_DIRECTION_RESPONSE.to_string(), "RawFileInfo".to_string(), 1)));
     }
 
     #[test]

crates/io-metrics/src/list_objects_metrics.rs

@@ -403,73 +403,132 @@ pub fn record_list_objects_local_read_dir(observation: ListObjectsLocalReadDirOb
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::set_get_stage_metrics_enabled;
+    use crate::tests::{METRICS_FLAG_LOCK, counter_total, emitted_names, histogram_samples};
+    use metrics_util::debugging::DebuggingRecorder;
 
-    #[test]
-    fn record_gather_observation_handles_empty_page() {
-        init_list_objects_metrics();
-        record_list_objects_gather(ListObjectsGatherObservation {
-            source: LIST_OBJECTS_SOURCE_WALKER,
-            outcome: LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED,
-            limit: 1001,
-            scanned_entries: 42,
-            returned_entries: 0,
-            duration_ms: 3.5,
-            has_prefix: true,
-            has_delimiter: false,
-            has_marker: true,
+    /// Run `body` against a local recorder with stage metrics on, and return the
+    /// snapshot rows.
+    ///
+    /// Enabling the flag is the point: every `record_*` here returns early when
+    /// `get_stage_metrics_enabled()` is false, which defaults to false. The
+    /// previous versions of these tests never set it, so they exercised nothing
+    /// but the early return (rustfs/backlog#1836).
+    fn recorded(body: impl FnOnce()) -> Vec<crate::tests::MetricRow> {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        metrics::with_local_recorder(&recorder, || {
+            init_list_objects_metrics();
+            set_get_stage_metrics_enabled(true);
+            body();
+            set_get_stage_metrics_enabled(false);
         });
+        snapshotter.snapshot().into_vec()
     }
 
     #[test]
-    fn record_merge_observation_accepts_zero_quorum() {
-        init_list_objects_metrics();
-        record_list_objects_merge(LIST_OBJECTS_SOURCE_WALKER, 4, 0);
+    fn gather_scan_amplification_falls_back_to_the_scan_count_on_an_empty_page() {
+        let rows = recorded(|| {
+            record_list_objects_gather(ListObjectsGatherObservation {
+                source: LIST_OBJECTS_SOURCE_WALKER,
+                outcome: LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED,
+                limit: 1001,
+                scanned_entries: 42,
+                returned_entries: 0,
+                duration_ms: 3.5,
+                has_prefix: true,
+                has_delimiter: false,
+                has_marker: true,
+            })
+        });
+
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_GATHER_TOTAL), Some(1));
+        // Zero returned entries must not divide: the amplification reports the
+        // scan count itself rather than an infinity or a NaN.
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_GATHER_SCAN_AMPLIFICATION), vec![42.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_GATHER_FILTERED_ENTRIES), vec![42.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_GATHER_RETURNED_ENTRIES), vec![0.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_GATHER_DURATION_MS), vec![3.5]);
     }
 
     #[test]
-    fn record_index_fallback_observation_accepts_reason() {
-        init_list_objects_metrics();
-        record_list_objects_index_fallback("index_key_only", "unsupported_request");
+    fn merge_records_a_zero_read_quorum_rather_than_skipping_it() {
+        let rows = recorded(|| record_list_objects_merge(LIST_OBJECTS_SOURCE_WALKER, 4, 0));
+
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_MERGE_FAN_IN), vec![4.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_MERGE_READ_QUORUM), vec![0.0]);
     }
 
     #[test]
-    fn record_index_attempt_and_served_observations_accept_counts() {
-        init_list_objects_metrics();
-        record_list_objects_index_attempt("index_key_only", "walker_key_only", true, true, false);
-        record_list_objects_index_served(ListObjectsIndexPageObservation {
-            source: "index_key_only",
-            provider: "walker_key_only",
-            candidate_keys: 1000,
-            live_verify_attempts: 700,
-            live_verify_hits: 650,
-            live_verify_misses: 50,
-            returned_objects: 600,
-            returned_prefixes: 10,
-            is_truncated: true,
+    fn index_fallback_counts_once_per_reason() {
+        let rows = recorded(|| {
+            record_list_objects_index_fallback("index_key_only", "unsupported_request");
+            record_list_objects_index_fallback("index_key_only", "unsupported_request");
         });
-        record_list_objects_index_live_verify_failure("index_key_only", "read_error");
+
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_INDEX_FALLBACK_TOTAL), Some(2));
     }
 
     #[test]
-    fn record_local_read_dir_observation_accepts_whole_directory_counts() {
-        init_list_objects_metrics();
-        record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation {
-            outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_OK,
-            requested_count: -1,
-            returned_entries: 4096,
-            duration_ms: 12.5,
-            is_root: true,
-            has_filter_prefix: false,
-            has_forward: false,
+    fn index_serving_reports_verification_amplification_against_returned_objects() {
+        let rows = recorded(|| {
+            record_list_objects_index_attempt("index_key_only", "walker_key_only", true, true, false);
+            record_list_objects_index_served(ListObjectsIndexPageObservation {
+                source: "index_key_only",
+                provider: "walker_key_only",
+                candidate_keys: 1000,
+                live_verify_attempts: 700,
+                live_verify_hits: 650,
+                live_verify_misses: 50,
+                returned_objects: 600,
+                returned_prefixes: 10,
+                is_truncated: true,
+            });
+            record_list_objects_index_live_verify_failure("index_key_only", "read_error");
         });
-        record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation {
-            outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_ERROR,
-            requested_count: -1,
-            returned_entries: 0,
-            duration_ms: 5000.0,
-            is_root: true,
-            has_filter_prefix: false,
-            has_forward: true,
+
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_INDEX_ATTEMPT_TOTAL), Some(1));
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_INDEX_SERVED_TOTAL), Some(1));
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_INDEX_LIVE_VERIFY_FAILURE_TOTAL), Some(1));
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_INDEX_CANDIDATE_KEYS), vec![1000.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_INDEX_LIVE_VERIFY_HITS), vec![650.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_INDEX_LIVE_VERIFY_MISSES), vec![50.0]);
+        assert_eq!(
+            histogram_samples(&rows, LIST_OBJECTS_INDEX_VERIFICATION_IO_AMPLIFICATION),
+            vec![700.0 / 600.0]
+        );
+    }
+
+    #[test]
+    fn local_read_dir_passes_the_whole_directory_sentinel_through_as_the_limit() {
+        let rows = recorded(|| {
+            record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation {
+                outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_OK,
+                requested_count: -1,
+                returned_entries: 4096,
+                duration_ms: 12.5,
+                is_root: true,
+                has_filter_prefix: false,
+                has_forward: false,
+            });
+            record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation {
+                outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_ERROR,
+                requested_count: -1,
+                returned_entries: 0,
+                duration_ms: 5000.0,
+                is_root: true,
+                has_filter_prefix: false,
+                has_forward: true,
+            });
         });
+
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_LOCAL_READ_DIR_TOTAL), Some(2));
+        // `-1` is the "read the whole directory" sentinel and must reach the
+        // limit histogram unchanged rather than being clamped to zero.
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_LOCAL_READ_DIR_LIMIT), vec![-1.0, -1.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_LOCAL_READ_DIR_ENTRIES), vec![0.0, 4096.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_LOCAL_READ_DIR_DURATION_MS), vec![12.5, 5000.0]);
+        assert!(emitted_names(&rows).contains(LIST_OBJECTS_LOCAL_READ_DIR_TOTAL));
     }
 }

crates/io-metrics/src/sampler/process.rs

@@ -188,18 +188,40 @@ mod tests {
     }
 
     #[test]
-    fn process_snapshots_are_collectable() {
-        let _ = snapshot_process_resource();
-        let _ = snapshot_process_system();
-        let _ = snapshot_process_resource_and_system();
+    fn combined_snapshot_agrees_with_the_individual_ones_on_per_process_facts() {
+        // Previously three discarded calls that asserted nothing. The values that
+        // move (cpu, memory) cannot be compared across calls, but the facts that
+        // identify the process must not differ by which entry point produced them
+        // (rustfs/backlog#1836).
+        let system = snapshot_process_system();
+        let (_, combined_system) = snapshot_process_resource_and_system();
+
+        assert_eq!(
+            system.start_time_seconds, combined_system.start_time_seconds,
+            "both entry points describe this process, so its start time cannot differ"
+        );
+        assert_eq!(
+            system.file_descriptor_limit_total, combined_system.file_descriptor_limit_total,
+            "the descriptor limit is a property of the process, not of the call"
+        );
+        assert_eq!(
+            system.status_value, combined_system.status_value,
+            "the status enum and its numeric projection must stay in step"
+        );
+        assert_eq!(combined_system.status_value, combined_system.status as i64);
     }
 
     #[test]
-    fn independent_samplers_are_collectable() {
+    fn independent_samplers_observe_the_same_process() {
         let mut sampler_a = ProcessSampler::new();
         let mut sampler_b = ProcessSampler::new();
 
-        let _ = snapshot_process_resource_and_system_with(&mut sampler_a);
-        let _ = snapshot_process_resource_and_system_with(&mut sampler_b);
+        let (_, system_a) = snapshot_process_resource_and_system_with(&mut sampler_a);
+        let (_, system_b) = snapshot_process_resource_and_system_with(&mut sampler_b);
+
+        // Two samplers hold separate sysinfo state; they must still agree on the
+        // process they are both looking at rather than each inventing a value.
+        assert_eq!(system_a.start_time_seconds, system_b.start_time_seconds);
+        assert_eq!(system_a.file_descriptor_limit_total, system_b.file_descriptor_limit_total);
     }
 }

crates/madmin/src/site_replication.rs

@@ -83,16 +83,6 @@ 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)]

crates/scanner/src/data_usage_define.rs

@@ -29,7 +29,7 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
     AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
     DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
-    PrefixUsageQuery, PrefixUsageSummary, TierStats, hash_path, prefix_usage_in_cache,
+    PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -188,38 +188,18 @@ pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
 
 const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
 
-/// Size summary for a single object or group of objects
-#[derive(Debug, Default, Clone)]
-pub struct SizeSummary {
-    /// Total size
-    pub total_size: usize,
-    /// Number of versions
-    pub versions: usize,
-    /// Number of delete markers
-    pub delete_markers: usize,
-    /// Replicated size
-    pub replicated_size: i64,
-    /// Replicated count
-    pub replicated_count: usize,
-    /// Pending size
-    pub pending_size: i64,
-    /// Failed size
-    pub failed_size: i64,
-    /// Replica size
-    pub replica_size: i64,
-    /// Replica count
-    pub replica_count: usize,
-    /// Pending count
-    pub pending_count: usize,
-    /// Failed count
-    pub failed_count: usize,
-    /// Replication target stats
-    pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
-    pub tier_stats: HashMap<String, TierStats>,
+/// Scanner-side accounting on the shared [`SizeSummary`].
+///
+/// The type itself lives in `rustfs-data-usage`, which sits below the storage
+/// layer and cannot see `ObjectInfo`, so this stays an extension trait rather
+/// than an inherent method (backlog#1828).
+pub trait ScannerSizeSummaryExt {
+    /// Fold one object's contribution into the summary, including its tier.
+    fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64);
 }
 
-impl SizeSummary {
-    pub fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64) {
+impl ScannerSizeSummaryExt for SizeSummary {
+    fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64) {
         if oi.delete_marker {
             self.delete_markers = self.delete_markers.saturating_add(1);
             return;
@@ -251,23 +231,6 @@ impl SizeSummary {
     }
 }
 
-/// Replication target size summary
-#[derive(Debug, Default, Clone)]
-pub struct ReplTargetSizeSummary {
-    /// Replicated size
-    pub replicated_size: i64,
-    /// Replicated count
-    pub replicated_count: usize,
-    /// Pending size
-    pub pending_size: i64,
-    /// Failed size
-    pub failed_size: i64,
-    /// Pending count
-    pub pending_count: usize,
-    /// Failed count
-    pub failed_count: usize,
-}
-
 // ===== Cache-related data structures =====
 
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -1544,39 +1507,6 @@ pub trait DataUsageCacheStorage {
     async fn save(&self, name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
 }
 
-impl SizeSummary {
-    /// Create a new SizeSummary
-    pub fn new() -> Self {
-        Self::default()
-    }
-
-    /// Add another SizeSummary to this one
-    pub fn add(&mut self, other: &SizeSummary) {
-        self.total_size = self.total_size.saturating_add(other.total_size);
-        self.versions = self.versions.saturating_add(other.versions);
-        self.delete_markers = self.delete_markers.saturating_add(other.delete_markers);
-        self.replicated_size = self.replicated_size.saturating_add(other.replicated_size);
-        self.replicated_count = self.replicated_count.saturating_add(other.replicated_count);
-        self.pending_size = self.pending_size.saturating_add(other.pending_size);
-        self.failed_size = self.failed_size.saturating_add(other.failed_size);
-        self.replica_size = self.replica_size.saturating_add(other.replica_size);
-        self.replica_count = self.replica_count.saturating_add(other.replica_count);
-        self.pending_count = self.pending_count.saturating_add(other.pending_count);
-        self.failed_count = self.failed_count.saturating_add(other.failed_count);
-
-        // Merge replication target stats
-        for (target, stats) in &other.repl_target_stats {
-            let entry = self.repl_target_stats.entry(target.clone()).or_default();
-            entry.replicated_size = entry.replicated_size.saturating_add(stats.replicated_size);
-            entry.replicated_count = entry.replicated_count.saturating_add(stats.replicated_count);
-            entry.pending_size = entry.pending_size.saturating_add(stats.pending_size);
-            entry.failed_size = entry.failed_size.saturating_add(stats.failed_size);
-            entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
-            entry.failed_count = entry.failed_count.saturating_add(stats.failed_count);
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/scanner/src/scanner_folder.rs

@@ -21,7 +21,7 @@ use std::time::{Duration, Instant, SystemTime};
 use crate::ReplTargetSizeSummary;
 use crate::data_usage_define::{
     DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint,
-    DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, SizeSummary, hash_path,
+    DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeSummary, hash_path,
 };
 use crate::error::ScannerError;
 use crate::runtime_config::{

rustfs/src/admin/handlers/site_replication.rs

@@ -126,10 +126,6 @@ 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";
@@ -717,16 +713,6 @@ 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)]
@@ -2582,21 +2568,6 @@ 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)
@@ -3031,17 +3002,8 @@ 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_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;
+                if state.pending_endpoint_refresh.is_some() || state.pending_remove.is_some() || state.pending_rotation.is_some()
+                {
                     return;
                 }
             }
@@ -7079,31 +7041,15 @@ 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_PARTIAL.to_string(),
-        err_detail: summarize_peer_error_detail(&format!(
-            "failed to notify {} peer(s): {}",
-            summaries.len(),
-            summaries.join("; ")
-        )),
+        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("; ")))
+        },
         api_version: Some(SITE_REPL_API_VERSION.to_string()),
     }
 }
@@ -7280,137 +7226,6 @@ 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
@@ -9841,12 +9656,9 @@ 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)
@@ -9949,19 +9761,6 @@ 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(
@@ -10134,7 +9933,79 @@ impl Operation for SiteReplicationRemoveHandler {
             .await?
         };
 
-        let (mut peer_errors, complete) = drive_pending_remove(&pending_remove, &local_peer).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
+        };
         if !complete && peer_errors.is_empty() {
             peer_errors.push("site replication remove is still pending".to_string());
         }
@@ -10148,25 +10019,6 @@ 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]
@@ -10175,7 +10027,14 @@ 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);
-        json_response(&site_replication_info_for(&state, &local_peer))
+        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)
     }
 }
 
@@ -10394,28 +10253,6 @@ 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)>> {
@@ -10438,14 +10275,10 @@ impl Operation for SRPeerJoinHandler {
         let (state, local_peer) = match committed {
             PeerJoinOutcome::Applied(state, local_peer) => (*state, local_peer),
             PeerJoinOutcome::Superseded(peer) => {
-                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));
+                return json_response(&SRPeerJoinResponse {
+                    peer,
+                    ..Default::default()
+                });
             }
         };
         // Fix 1 (receiving side): ensure the joining peer also sets up replication for any
@@ -10464,10 +10297,10 @@ impl Operation for SRPeerJoinHandler {
                 "admin site replication state"
             );
         }
-        json_response(&applied_join_response(
-            state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer),
-            backfill_errors.render(),
-        ))
+        json_response(&SRPeerJoinResponse {
+            peer: state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer),
+            initial_sync_error_message: backfill_errors.render(),
+        })
     }
 }
 
@@ -11511,11 +11344,7 @@ impl Operation for SRRotateServiceAccountHandler {
             {
                 continue;
             }
-            // 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(
+            if let Err(err) = send_peer_admin_request_with_secret_candidates(
                 &runtime_peer_connection(peer)?,
                 SITE_REPLICATION_PEER_JOIN_PATH,
                 &pending_rotation.access_key,
@@ -11524,20 +11353,7 @@ impl Operation for SRRotateServiceAccountHandler {
             )
             .await
             {
-                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 {
+                let detail = summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint));
                 warn!(
                     event = EVENT_ADMIN_SITE_REPLICATION_STATE,
                     component = LOG_COMPONENT_ADMIN,
@@ -15823,17 +15639,9 @@ 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_PARTIAL,
-            "a removal whose peer could not be notified must not report unqualified success"
-        );
+        assert_eq!(status.status, SITE_REPL_REMOVE_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]
@@ -17174,22 +16982,16 @@ 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());
     }
@@ -17910,319 +17712,13 @@ 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
@@ -18270,14 +17766,12 @@ 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");
 
-        // 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).
+        // Even if peers returned 403 (desynced account), status still reports success
         let status =
             site_replication_remove_status(&["https://remote.example.com: peer/remove returned 403 Forbidden".to_string()]);
         assert_eq!(
-            status.status, SITE_REPL_REMOVE_PARTIAL,
-            "local remove must report a partial result when peer notifications fail"
+            status.status, SITE_REPL_REMOVE_SUCCESS,
+            "local remove reports success even when peer notifications fail"
         );
         assert!(
             status.err_detail.contains("403 Forbidden"),

scripts/test/site_replication_smoke.py

@@ -27,7 +27,6 @@ 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
@@ -338,12 +337,11 @@ def ensure_pair(site_a: Site, site_b: Site) -> None:
     print(f"[ok] site replication configured: {result.get('status', '')}")
 
 
-def remove_pair(site: Site) -> dict:
+def remove_pair(site: Site) -> None:
     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)
 
 
 # ---------------------------------------------------------------------------
@@ -399,112 +397,6 @@ 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
 # ---------------------------------------------------------------------------
@@ -572,7 +464,7 @@ def main() -> None:
         "command",
         nargs="?",
         default="up",
-        choices=["up", "down", "restart", "status", "logs", "smoke", "diverge", "info", "remove", "clean"],
+        choices=["up", "down", "restart", "status", "logs", "smoke", "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)")
@@ -603,8 +495,6 @@ 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":