feat(heal): aggregate replacement recovery status (#5916)

Add a replacement recovery peer RPC so Admin v4 can distinguish definitive cluster proofs from unsupported, unavailable, or conflicting peer state without extending the existing background heal v3/v1 status protocol.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/e2e_test/src/reliant/grpc_lock_server.rs

@@ -848,6 +848,13 @@ impl NodeService for MinimalLockNodeService {
         Err(Status::unimplemented("lock-only test server"))
     }
 
+    async fn replacement_recovery_status(
+        &self,
+        _request: Request<rustfs_protos::proto_gen::node_service::ReplacementRecoveryStatusRequest>,
+    ) -> Result<Response<rustfs_protos::proto_gen::node_service::ReplacementRecoveryStatusResponse>, Status> {
+        Err(Status::unimplemented("lock-only test server"))
+    }
+
     async fn get_metacache_listing(
         &self,
         _request: Request<rustfs_protos::proto_gen::node_service::GetMetacacheListingRequest>,

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -44,9 +44,9 @@ use rustfs_protos::proto_gen::node_service::{
     GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, GetSysConfigRequest, GetSysErrorsRequest,
     HealControlRequest, LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest,
     LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest,
-    LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ScannerActivityRequest,
-    ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest,
-    StartProfilingRequest, StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest,
+    LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ReplacementRecoveryStatusRequest,
+    ScannerActivityRequest, ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest, SignalServiceResponse,
+    StartDecommissionRequest, StartProfilingRequest, StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest,
     TierMutationControlResponse, TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient,
     tier_mutation_control_service_client::TierMutationControlServiceClient,
 };
@@ -78,6 +78,7 @@ pub const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = 2;
 /// reload signal transport.
 pub const KMS_SIGNAL_SUBSYSTEM: &str = "kms";
 const BACKGROUND_HEAL_STATUS_MAX_MESSAGE_SIZE: usize = 64 * 1024;
+const REPLACEMENT_RECOVERY_STATUS_MAX_MESSAGE_SIZE: usize = 64 * 1024;
 const HEAL_CONTROL_FINGERPRINT_MAX_SIZE: usize = 256;
 const HEAL_CONTROL_PAYLOAD_MAX_SIZE: usize = 64 * 1024;
 const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
@@ -1083,6 +1084,38 @@ impl PeerRestClient {
         .await
     }
 
+    pub async fn replacement_recovery_status(&self) -> Result<Option<Vec<u8>>> {
+        self.finalize_result(
+            async {
+                let mut client = self
+                    .get_client()
+                    .await?
+                    .max_decoding_message_size(REPLACEMENT_RECOVERY_STATUS_MAX_MESSAGE_SIZE);
+                let response = match client
+                    .replacement_recovery_status(Request::new(ReplacementRecoveryStatusRequest::default()))
+                    .await
+                {
+                    Ok(response) => response.into_inner(),
+                    Err(status) if status.code() == tonic::Code::Unimplemented => {
+                        // RUSTFS_COMPAT_TODO(replacement-recovery-status-v1): old peers cannot prove replacement completion during rolling upgrades. Remove after the minimum supported RustFS peer version implements ReplacementRecoveryStatus.
+                        return Ok(None);
+                    }
+                    Err(status) => return Err(status.into()),
+                };
+                if !response.success {
+                    return Err(Error::other(
+                        response
+                            .error_info
+                            .unwrap_or_else(|| "peer replacement recovery status failed without an error".to_string()),
+                    ));
+                }
+                Ok(Some(response.recovery_status.to_vec()))
+            }
+            .await,
+        )
+        .await
+    }
+
     pub async fn prepare_tier_mutation(&self, mutation_id: Uuid, canonical_payload: Bytes) -> Result<PeerTierMutationOutcome> {
         self.tier_mutation_control(TierMutationRpcPhase::Prepare, mutation_id, canonical_payload)
             .await

crates/protos/src/generated/proto_gen/node_service.rs

@@ -1221,6 +1221,17 @@ pub struct BackgroundHealStatusResponse {
     #[prost(string, optional, tag = "3")]
     pub error_info: ::core::option::Option<::prost::alloc::string::String>,
 }
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ReplacementRecoveryStatusRequest {}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ReplacementRecoveryStatusResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub recovery_status: ::prost::bytes::Bytes,
+    #[prost(string, optional, tag = "3")]
+    pub error_info: ::core::option::Option<::prost::alloc::string::String>,
+}
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct HealControlRequest {
     #[prost(uint32, tag = "1")]
@@ -2692,6 +2703,21 @@ pub mod node_service_client {
                 .insert(GrpcMethod::new("node_service.NodeService", "BackgroundHealStatus"));
             self.inner.unary(req, path, codec).await
         }
+        pub async fn replacement_recovery_status(
+            &mut self,
+            request: impl tonic::IntoRequest<super::ReplacementRecoveryStatusRequest>,
+        ) -> std::result::Result<tonic::Response<super::ReplacementRecoveryStatusResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ReplacementRecoveryStatus");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "ReplacementRecoveryStatus"));
+            self.inner.unary(req, path, codec).await
+        }
         pub async fn get_metacache_listing(
             &mut self,
             request: impl tonic::IntoRequest<super::GetMetacacheListingRequest>,
@@ -3179,6 +3205,10 @@ pub mod node_service_server {
             &self,
             request: tonic::Request<super::BackgroundHealStatusRequest>,
         ) -> std::result::Result<tonic::Response<super::BackgroundHealStatusResponse>, tonic::Status>;
+        async fn replacement_recovery_status(
+            &self,
+            request: tonic::Request<super::ReplacementRecoveryStatusRequest>,
+        ) -> std::result::Result<tonic::Response<super::ReplacementRecoveryStatusResponse>, tonic::Status>;
         async fn get_metacache_listing(
             &self,
             request: tonic::Request<super::GetMetacacheListingRequest>,
@@ -5479,6 +5509,34 @@ pub mod node_service_server {
                     };
                     Box::pin(fut)
                 }
+                "/node_service.NodeService/ReplacementRecoveryStatus" => {
+                    #[allow(non_camel_case_types)]
+                    struct ReplacementRecoveryStatusSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::ReplacementRecoveryStatusRequest> for ReplacementRecoveryStatusSvc<T> {
+                        type Response = super::ReplacementRecoveryStatusResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::ReplacementRecoveryStatusRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::replacement_recovery_status(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = ReplacementRecoveryStatusSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 "/node_service.NodeService/GetMetacacheListing" => {
                     #[allow(non_camel_case_types)]
                     struct GetMetacacheListingSvc<T: NodeService>(pub Arc<T>);

crates/protos/src/node.proto

@@ -851,6 +851,14 @@ message BackgroundHealStatusResponse {
   optional string error_info = 3;
 }
 
+message ReplacementRecoveryStatusRequest {}
+
+message ReplacementRecoveryStatusResponse {
+  bool success = 1;
+  bytes recovery_status = 2;
+  optional string error_info = 3;
+}
+
 message HealControlRequest {
   uint32 version = 1;
   string topology_fingerprint = 2;
@@ -1084,6 +1092,7 @@ service NodeService {
   rpc SignalService(SignalServiceRequest) returns (SignalServiceResponse) {}; // auth-policy: body-bound
   rpc ScannerActivity(ScannerActivityRequest) returns (ScannerActivityResponse) {}; // auth-policy: body-bound
   rpc BackgroundHealStatus(BackgroundHealStatusRequest) returns (BackgroundHealStatusResponse) {}; // auth-policy: read-only
+  rpc ReplacementRecoveryStatus(ReplacementRecoveryStatusRequest) returns (ReplacementRecoveryStatusResponse) {}; // auth-policy: read-only
   rpc GetMetacacheListing(GetMetacacheListingRequest) returns (GetMetacacheListingResponse) {}; // auth-policy: unimplemented
   rpc UpdateMetacacheListing(UpdateMetacacheListingRequest) returns (UpdateMetacacheListingResponse) {}; // auth-policy: unimplemented
   rpc ReloadPoolMeta(ReloadPoolMetaRequest) returns (ReloadPoolMetaResponse) {}; // auth-policy: body-bound

docs/architecture/compat-cleanup-register.md

@@ -25,6 +25,7 @@ for later deletion.
 - `heal-rpc-auth-v2` internode gRPC authentication: servers temporarily accept legacy prefix signatures so old peers remain available during rolling upgrades. Remove the legacy fallback after the minimum supported RustFS peer version sends v2 authentication on every internode gRPC request.
 - `disk-mutation-body-digest` internode mutating disk RPCs: servers temporarily accept mutating disk RPCs (RenameData, DeleteVersion, DeleteVersions, WriteMetadata, UpdateMetadata, WriteAll, Delete, DeletePaths, RenameFile, RenamePart, DeleteVolume, MakeVolume, MakeVolumes) that carry no signature-bound canonical body digest, so peers from releases that predate body-digest signing remain available during rolling upgrades. Accepted digestless mutations increment the internode body-digest fallback counter; that counter must read zero fleet-wide across a release window before RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT is enabled. Because body-bound requests now consume replay-cache nonces on the receiver, deploy the raised RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY default fleet-wide before enabling strict mode, and watch the internode replay-cache overflow counter for undersized capacity during the rollout. Remove the digestless fallback after the minimum supported RustFS peer version body-binds every mutating disk RPC.
 - `heal-status-rpc-v1` node heal status capability: new peers treat an unimplemented BackgroundHealStatus RPC as an explicitly incomplete rolling-upgrade response. Remove the fallback after the minimum supported RustFS peer version implements BackgroundHealStatus.
+- `replacement-recovery-status-v1` replacement recovery status capability: new peers treat an unimplemented ReplacementRecoveryStatus RPC as an explicitly non-definitive rolling-upgrade response, so Admin v4 cannot claim distributed replacement completion from old peers. Remove the fallback after the minimum supported RustFS peer version implements ReplacementRecoveryStatus.
 - `backlog-1316` legacy encrypted multipart range seek: the feature remains opt-in until every server that can initiate, write, or complete multipart uploads supports the candidate-to-final marker protocol and uploadId commit lock, and pre-upgrade multipart uploads have drained. Remove the RUSTFS_ENCRYPTED_RANGE_SEEK switch after the minimum supported release does so; keep the quorum marker and malformed-layout full-read guards permanently.
 - `tonic-013-status-render` peer RPC failure classification: internode failures that reach a node only as text (a peer's error_info payload, a status flattened through format!) are classified by matching the rendering of an Unavailable gRPC status. Releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered that status as "status: Unavailable, message: ..."; tonic 0.14 renders it as "code: 'The service is currently unavailable', message: ...". Both forms are matched so an older peer's relayed text still marks an unreachable peer offline. Remove the tonic 0.13 form after the minimum supported RustFS peer version ships tonic 0.14 or later.
 - `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection.

docs/operations/scanner-runtime-controls.md

@@ -221,7 +221,7 @@ Treat replacement recovery as verified only after the repair task has completed
 
 The v3 route and its peer status protocol preserve their existing fields for mixed-version clusters. A new node must not infer replacement completion from an old or unavailable peer; regard that information as unknown or degraded until every required peer can report the same replacement instance and verified completion. Do not automate destructive replacement actions from an `idle` observation alone.
 
-`GET /rustfs/admin/v4/heal/replacement-recovery` reports the local node's durable automatic replacement records from survivor disks. Its `local.records[]` entries distinguish `waiting_for_replacement`, `running`, `incomplete`, `unrecoverable`, `cleanup_pending`, `completed`, and `unknown`; `local.definitive=false` or any `unknown` record means the node could not prove a local replacement state. The `cluster.definitive` field is intentionally `false` until a peer capability RPC can prove that every required node reports the same replacement generation and target instance, so operators must not treat this route alone as distributed completion proof.
+`GET /rustfs/admin/v4/heal/replacement-recovery` reports durable automatic replacement records from survivor disks. Its `local.records[]` entries distinguish `waiting_for_replacement`, `running`, `incomplete`, `unrecoverable`, `cleanup_pending`, `completed`, and `unknown`; `local.definitive=false` or any `unknown` record means the node could not prove a local replacement state. Its `cluster` section queries the replacement-recovery peer RPC and sets `cluster.definitive=true` only when the expected peer topology is complete, every peer supports the RPC, every peer snapshot is locally definitive, and all peers report the same replacement records. Old peers, unavailable peers, malformed peer payloads, topology gaps, and generation disagreements are reported as degraded or unknown rather than complete.
 
 Replacement resume and checkpoint files use an independent on-disk schema. A newer reader rejects a future schema rather than continuing with data it cannot interpret, while an older binary cannot safely enforce the new generation fence because it may ignore fields it does not know. Do not roll a cluster back after a replacement generation has started. Complete that recovery with the current-or-newer release; if it cannot complete, keep that version for diagnosis rather than deleting its durable records or continuing with an older binary.
 

rustfs/src/admin/handlers/heal.rs

@@ -22,7 +22,7 @@ use crate::server::ADMIN_PREFIX;
 use crate::server::RemoteAddr;
 use crate::storage::rpc::node_service::heal::{
     HealControlCoordinator, NodeHealProgress, NodeHealStatusSnapshot, capture_node_heal_status, decode_node_heal_status,
-    heal_control_coordinator, heal_topology_fingerprint,
+    decode_node_replacement_recovery_status, heal_control_coordinator, heal_topology_fingerprint,
 };
 use bytes::Bytes;
 use futures_util::future::join_all;
@@ -40,7 +40,7 @@ use rustfs_utils::path::path_join;
 use s3s::header::{CONTENT_LENGTH, CONTENT_TYPE};
 use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
 use serde::{Deserialize, Serialize};
-use std::collections::HashSet;
+use std::collections::{BTreeMap, HashSet};
 use std::future::Future;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -55,7 +55,7 @@ const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed";
 const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted";
 const PEER_HEAL_STATUS_TIMEOUT: Duration = Duration::from_secs(5);
 pub(crate) const REPLACEMENT_RECOVERY_STATUS_ROUTE_SUFFIX: &str = "/v4/heal/replacement-recovery";
-const REPLACEMENT_RECOVERY_STATUS_CONTRACT_VERSION: u32 = 1;
+const REPLACEMENT_RECOVERY_STATUS_CONTRACT_VERSION: u32 = 2;
 
 #[derive(Debug, Default, Serialize, Deserialize)]
 struct HealInitParams {
@@ -529,6 +529,204 @@ async fn read_cluster_heal_status(
     merge_peer_heal_statuses(snapshots, peer_statuses, expected_nodes, topology_complete)
 }
 
+async fn query_peer_replacement_recovery_status<E>(
+    host: &str,
+    request: impl std::future::Future<Output = Result<Option<Vec<u8>>, E>>,
+    request_timeout: Duration,
+) -> Result<Option<rustfs_heal::ReplacementRecoverySnapshot>, String>
+where
+    E: std::fmt::Display,
+{
+    match timeout(request_timeout, request).await {
+        Ok(Ok(Some(status))) => decode_node_replacement_recovery_status(&status)
+            .map(|snapshot| Some(snapshot.snapshot))
+            .map_err(|err| format!("peer {host}: {err}")),
+        Ok(Ok(None)) => Ok(None),
+        Ok(Err(err)) => Err(format!("peer {host}: {err}")),
+        Err(_) => Err(format!("peer {host}: replacement recovery status timed out")),
+    }
+}
+
+fn canonical_replacement_records(
+    records: &[rustfs_heal::ReplacementRecoveryRecord],
+) -> Vec<rustfs_heal::ReplacementRecoveryRecord> {
+    let mut records = records.to_vec();
+    for record in &mut records {
+        record.target_slots.sort();
+    }
+    records.sort_by(|left, right| {
+        (
+            &left.task_id,
+            &left.generation,
+            &left.set_disk_id,
+            left.state,
+            &left.target_slots,
+            &left.verified_at,
+            &left.reason,
+        )
+            .cmp(&(
+                &right.task_id,
+                &right.generation,
+                &right.set_disk_id,
+                right.state,
+                &right.target_slots,
+                &right.verified_at,
+                &right.reason,
+            ))
+    });
+    records
+}
+
+fn merge_replacement_recovery_records(
+    snapshots: &[rustfs_heal::ReplacementRecoverySnapshot],
+) -> Vec<rustfs_heal::ReplacementRecoveryRecord> {
+    let mut records = BTreeMap::<String, rustfs_heal::ReplacementRecoveryRecord>::new();
+    for snapshot in snapshots {
+        for record in canonical_replacement_records(&snapshot.records) {
+            match records.entry(record.task_id.clone()) {
+                std::collections::btree_map::Entry::Vacant(entry) => {
+                    entry.insert(record);
+                }
+                std::collections::btree_map::Entry::Occupied(entry) if entry.get() == &record => {}
+                std::collections::btree_map::Entry::Occupied(mut entry) => {
+                    let task_id = entry.key().clone();
+                    entry.insert(rustfs_heal::ReplacementRecoveryRecord {
+                        task_id,
+                        state: rustfs_heal::ReplacementRecoveryState::Unknown,
+                        generation: None,
+                        set_disk_id: None,
+                        target_slots: Vec::new(),
+                        reason: Some("peer replacement recovery records disagree".to_string()),
+                        verified_at: None,
+                    });
+                }
+            }
+        }
+    }
+    records.into_values().collect()
+}
+
+fn aggregate_replacement_recovery_cluster_status(
+    mut snapshots: Vec<rustfs_heal::ReplacementRecoverySnapshot>,
+    peer_statuses: Vec<Result<Option<rustfs_heal::ReplacementRecoverySnapshot>, String>>,
+    expected_nodes: usize,
+    topology_complete: bool,
+) -> ReplacementRecoveryClusterStatus {
+    let mut reason = None::<&'static str>;
+    if !topology_complete {
+        reason = Some("peer_topology_incomplete");
+    }
+
+    for status in peer_statuses {
+        match status {
+            Ok(Some(snapshot)) => snapshots.push(snapshot),
+            Ok(None) => {
+                reason.get_or_insert("peer_replacement_recovery_status_unsupported");
+            }
+            Err(_) => {
+                reason.get_or_insert("peer_replacement_recovery_status_unavailable");
+            }
+        }
+    }
+
+    if snapshots.len() != expected_nodes {
+        reason.get_or_insert("peer_replacement_recovery_status_incomplete");
+    }
+    if snapshots.iter().any(|snapshot| !snapshot.definitive) {
+        reason.get_or_insert("peer_replacement_recovery_status_not_definitive");
+    }
+
+    let first_records = snapshots
+        .first()
+        .map(|snapshot| canonical_replacement_records(&snapshot.records));
+    if let Some(first_records) = &first_records
+        && snapshots
+            .iter()
+            .skip(1)
+            .any(|snapshot| canonical_replacement_records(&snapshot.records) != *first_records)
+    {
+        reason.get_or_insert("peer_replacement_recovery_status_conflict");
+    }
+
+    let records = merge_replacement_recovery_records(&snapshots);
+    if records
+        .iter()
+        .any(|record| matches!(record.state, rustfs_heal::ReplacementRecoveryState::Unknown))
+    {
+        reason.get_or_insert("peer_replacement_recovery_status_conflict");
+    }
+
+    ReplacementRecoveryClusterStatus {
+        definitive: reason.is_none(),
+        reason: reason.map(str::to_string),
+        records,
+    }
+}
+
+async fn read_cluster_replacement_recovery_status(
+    local: rustfs_heal::ReplacementRecoverySnapshot,
+    notification_system: Option<&crate::admin::storage_api::runtime_sources::NotificationSys>,
+    expected_nodes: usize,
+) -> S3Result<ReplacementRecoveryClusterStatus> {
+    if expected_nodes == 1 {
+        return Ok(aggregate_replacement_recovery_cluster_status(
+            vec![local],
+            Vec::new(),
+            expected_nodes,
+            true,
+        ));
+    }
+    let Some(notification_system) = notification_system else {
+        return Ok(ReplacementRecoveryClusterStatus {
+            definitive: false,
+            reason: Some("notification_system_unavailable".to_string()),
+            records: local.records,
+        });
+    };
+    let topology_complete = peer_topology_complete(
+        expected_nodes,
+        notification_system.peer_clients.len(),
+        notification_system
+            .peer_clients
+            .iter()
+            .filter(|client| client.is_none())
+            .count(),
+        notification_system.all_peer_clients.len(),
+        notification_system
+            .all_peer_clients
+            .iter()
+            .filter(|client| client.is_some())
+            .count(),
+    );
+    if !topology_complete {
+        warn!(
+            event = EVENT_ADMIN_REQUEST_FAILED,
+            component = LOG_COMPONENT_ADMIN_API,
+            subsystem = LOG_SUBSYSTEM_HEAL_ADMIN,
+            operation = "replacement_recovery_status",
+            result = "degraded",
+            reason = "peer_topology_incomplete",
+            "cluster replacement recovery status will be partial"
+        );
+    }
+
+    let peer_statuses = join_all(notification_system.peer_clients.iter().map(|client| async move {
+        let Some(client) = client else {
+            return Err("configured peer is unavailable".to_string());
+        };
+        let host = client.host.to_string();
+        query_peer_replacement_recovery_status(&host, client.replacement_recovery_status(), PEER_HEAL_STATUS_TIMEOUT).await
+    }))
+    .await;
+
+    Ok(aggregate_replacement_recovery_cluster_status(
+        vec![local],
+        peer_statuses,
+        expected_nodes,
+        topology_complete,
+    ))
+}
+
 fn cluster_heal_control_unavailable(reason: &str) -> s3s::S3Error {
     warn!(
         event = EVENT_ADMIN_REQUEST_FAILED,
@@ -1036,21 +1234,20 @@ pub(crate) struct ReplacementRecoveryStatusResponse {
 #[serde(rename_all = "camelCase")]
 pub(crate) struct ReplacementRecoveryClusterStatus {
     pub definitive: bool,
-    pub reason: &'static str,
+    pub reason: Option<String>,
+    pub records: Vec<rustfs_heal::ReplacementRecoveryRecord>,
 }
 
 fn build_replacement_recovery_status_response(
     local: rustfs_heal::ReplacementRecoverySnapshot,
+    cluster: ReplacementRecoveryClusterStatus,
 ) -> ReplacementRecoveryStatusResponse {
     ReplacementRecoveryStatusResponse {
         contract_version: REPLACEMENT_RECOVERY_STATUS_CONTRACT_VERSION,
         path: format!("{}{}", ADMIN_PREFIX, REPLACEMENT_RECOVERY_STATUS_ROUTE_SUFFIX),
-        scope: "local_survivor_disks",
+        scope: "cluster_survivor_disks",
         local,
-        cluster: ReplacementRecoveryClusterStatus {
-            definitive: false,
-            reason: "distributed replacement recovery status requires a peer capability RPC",
-        },
+        cluster,
     }
 }
 
@@ -1347,8 +1544,21 @@ impl Operation for ReplacementRecoveryStatusHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         validate_heal_admin_request(&req).await?;
 
+        let Some(context) = app_context_from_req(&req) else {
+            return Err(cluster_heal_status_unavailable("server_not_initialized"));
+        };
+        let Some(endpoints) = context.endpoints().handle() else {
+            return Err(cluster_heal_status_unavailable("endpoint_topology_unavailable"));
+        };
+        let expected_nodes = endpoints.get_nodes().len();
+        if expected_nodes == 0 {
+            return Err(cluster_heal_status_unavailable("endpoint_topology_empty"));
+        }
+        let notification_system = context.notification_system().handle();
         let local = rustfs_heal::current_replacement_recovery_snapshot().await;
-        let response = build_replacement_recovery_status_response(local);
+        let cluster =
+            read_cluster_replacement_recovery_status(local.clone(), notification_system.as_deref(), expected_nodes).await?;
+        let response = build_replacement_recovery_status_response(local, cluster);
         let body = encode_replacement_recovery_status(&response)?;
         info!(
             event = EVENT_ADMIN_RESPONSE_EMITTED,
@@ -1368,14 +1578,17 @@ mod tests {
     use super::extract_heal_init_params;
     use super::{
         BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState, aggregate_cluster_heal_status,
-        background_heal_runtime_state, build_heal_channel_request, build_replacement_recovery_status_response,
-        encode_background_heal_status, encode_heal_start_success, encode_heal_task_status, execute_after_heal_control_capability,
-        heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary, json_response,
-        map_heal_response, map_root_heal_status, merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status,
+        aggregate_replacement_recovery_cluster_status, background_heal_runtime_state, build_heal_channel_request,
+        build_replacement_recovery_status_response, encode_background_heal_status, encode_heal_start_success,
+        encode_heal_task_status, execute_after_heal_control_capability, heal_channel_response_items,
+        heal_channel_response_progress, heal_channel_response_summary, json_response, map_heal_response, map_root_heal_status,
+        merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status, query_peer_replacement_recovery_status,
         reject_heal_admission, should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target,
     };
     use crate::admin::storage_api::error::StorageError;
-    use crate::storage::rpc::node_service::heal::{NodeHealProgress, NodeHealStatusSnapshot};
+    use crate::storage::rpc::node_service::heal::{
+        NodeHealProgress, NodeHealStatusSnapshot, NodeReplacementRecoveryStatusSnapshot, encode_node_replacement_recovery_status,
+    };
     use bytes::Bytes;
     use http::StatusCode;
     use http::Uri;
@@ -1394,6 +1607,26 @@ mod tests {
     use tokio::sync::mpsc;
     use tokio::time::Duration;
 
+    fn replacement_record(task_id: &str) -> rustfs_heal::ReplacementRecoveryRecord {
+        rustfs_heal::ReplacementRecoveryRecord {
+            task_id: task_id.to_string(),
+            state: rustfs_heal::ReplacementRecoveryState::Completed,
+            generation: Some(task_id.to_string()),
+            set_disk_id: Some("pool_0_set_0".to_string()),
+            target_slots: vec!["http://node-a:9000/mnt/disk1".to_string()],
+            reason: None,
+            verified_at: Some(42),
+        }
+    }
+
+    fn replacement_snapshot(task_id: &str) -> rustfs_heal::ReplacementRecoverySnapshot {
+        rustfs_heal::ReplacementRecoverySnapshot {
+            records: vec![replacement_record(task_id)],
+            definitive: true,
+            reason: None,
+        }
+    }
+
     #[tokio::test]
     async fn cluster_capability_gate_runs_before_execution() {
         let executed = AtomicBool::new(false);
@@ -1421,32 +1654,94 @@ mod tests {
     }
 
     #[test]
-    fn replacement_recovery_status_response_never_claims_cluster_completion() {
-        let response = build_replacement_recovery_status_response(rustfs_heal::ReplacementRecoverySnapshot {
-            records: vec![rustfs_heal::ReplacementRecoveryRecord {
-                task_id: "11111111-1111-4111-8111-111111111111".to_string(),
-                state: rustfs_heal::ReplacementRecoveryState::Completed,
-                generation: Some("11111111-1111-4111-8111-111111111111".to_string()),
-                set_disk_id: Some("pool_0_set_0".to_string()),
-                target_slots: vec!["http://node-a:9000/mnt/disk1".to_string()],
-                reason: None,
-                verified_at: Some(42),
-            }],
-            definitive: true,
-            reason: None,
-        });
+    fn replacement_recovery_status_response_reports_cluster_proof() {
+        let local = replacement_snapshot("11111111-1111-4111-8111-111111111111");
+        let cluster = aggregate_replacement_recovery_cluster_status(vec![local.clone()], Vec::new(), 1, true);
+        let response = build_replacement_recovery_status_response(local, cluster);
         let value = serde_json::to_value(response).expect("replacement status response should serialize");
 
-        assert_eq!(value["contractVersion"], 1);
+        assert_eq!(value["contractVersion"], 2);
         assert_eq!(value["path"], "/rustfs/admin/v4/heal/replacement-recovery");
-        assert_eq!(value["scope"], "local_survivor_disks");
+        assert_eq!(value["scope"], "cluster_survivor_disks");
         assert_eq!(value["local"]["definitive"], true);
         assert_eq!(value["local"]["records"][0]["state"], "completed");
-        assert_eq!(value["cluster"]["definitive"], false);
-        assert_eq!(
-            value["cluster"]["reason"],
-            "distributed replacement recovery status requires a peer capability RPC"
-        );
+        assert_eq!(value["cluster"]["definitive"], true);
+        assert!(value["cluster"]["reason"].is_null());
+        assert_eq!(value["cluster"]["records"][0]["state"], "completed");
+    }
+
+    #[test]
+    fn replacement_recovery_cluster_requires_peer_capability() {
+        let local = replacement_snapshot("11111111-1111-4111-8111-111111111111");
+        let cluster = aggregate_replacement_recovery_cluster_status(vec![local], vec![Ok(None)], 2, true);
+
+        assert!(!cluster.definitive);
+        assert_eq!(cluster.reason.as_deref(), Some("peer_replacement_recovery_status_unsupported"));
+        assert_eq!(cluster.records[0].state, rustfs_heal::ReplacementRecoveryState::Completed);
+    }
+
+    #[test]
+    fn replacement_recovery_cluster_requires_matching_peer_records() {
+        let local = replacement_snapshot("11111111-1111-4111-8111-111111111111");
+        let peer = replacement_snapshot("22222222-2222-4222-8222-222222222222");
+        let cluster = aggregate_replacement_recovery_cluster_status(vec![local], vec![Ok(Some(peer))], 2, true);
+
+        assert!(!cluster.definitive);
+        assert_eq!(cluster.reason.as_deref(), Some("peer_replacement_recovery_status_conflict"));
+        assert_eq!(cluster.records.len(), 2);
+    }
+
+    #[test]
+    fn replacement_recovery_cluster_rejects_conflicting_records_inside_a_snapshot() {
+        let mut local = replacement_snapshot("11111111-1111-4111-8111-111111111111");
+        let mut conflicting = replacement_record("11111111-1111-4111-8111-111111111111");
+        conflicting.set_disk_id = Some("pool_0_set_1".to_string());
+        local.records.push(conflicting);
+        let cluster = aggregate_replacement_recovery_cluster_status(vec![local], Vec::new(), 1, true);
+
+        assert!(!cluster.definitive);
+        assert_eq!(cluster.reason.as_deref(), Some("peer_replacement_recovery_status_conflict"));
+        assert_eq!(cluster.records[0].state, rustfs_heal::ReplacementRecoveryState::Unknown);
+    }
+
+    #[test]
+    fn replacement_recovery_cluster_is_definitive_when_peers_match() {
+        let local = replacement_snapshot("11111111-1111-4111-8111-111111111111");
+        let peer = replacement_snapshot("11111111-1111-4111-8111-111111111111");
+        let cluster = aggregate_replacement_recovery_cluster_status(vec![local], vec![Ok(Some(peer))], 2, true);
+
+        assert!(cluster.definitive);
+        assert!(cluster.reason.is_none());
+        assert_eq!(cluster.records.len(), 1);
+        assert_eq!(cluster.records[0].state, rustfs_heal::ReplacementRecoveryState::Completed);
+    }
+
+    #[tokio::test]
+    async fn peer_replacement_recovery_status_decodes_supported_peers() {
+        let payload = encode_node_replacement_recovery_status(&NodeReplacementRecoveryStatusSnapshot::new(replacement_snapshot(
+            "11111111-1111-4111-8111-111111111111",
+        )))
+        .expect("peer status should encode");
+        let decoded =
+            query_peer_replacement_recovery_status("node-a", async { Ok::<_, String>(Some(payload)) }, Duration::from_secs(1))
+                .await
+                .expect("peer query should succeed")
+                .expect("peer should support status");
+
+        assert_eq!(decoded.records[0].state, rustfs_heal::ReplacementRecoveryState::Completed);
+    }
+
+    #[tokio::test]
+    async fn peer_replacement_recovery_status_treats_missing_rpc_as_unknown() {
+        let decoded = query_peer_replacement_recovery_status(
+            "node-a",
+            async { Ok::<Option<Vec<u8>>, String>(None) },
+            Duration::from_secs(1),
+        )
+        .await
+        .expect("missing rpc should not be a transport failure");
+
+        assert!(decoded.is_none());
     }
 
     #[test]

rustfs/src/storage/rpc/node_service.rs

@@ -1934,6 +1934,32 @@ impl Node for NodeService {
         }
     }
 
+    async fn replacement_recovery_status(
+        &self,
+        _request: Request<ReplacementRecoveryStatusRequest>,
+    ) -> Result<Response<ReplacementRecoveryStatusResponse>, Status> {
+        if self.resolve_object_store().is_none() {
+            return Ok(Response::new(ReplacementRecoveryStatusResponse {
+                success: false,
+                recovery_status: Bytes::new(),
+                error_info: Some("storage layer not initialized".to_string()),
+            }));
+        }
+        let snapshot = heal::capture_node_replacement_recovery_status().await;
+        match heal::encode_node_replacement_recovery_status(&snapshot) {
+            Ok(recovery_status) => Ok(Response::new(ReplacementRecoveryStatusResponse {
+                success: true,
+                recovery_status: recovery_status.into(),
+                error_info: None,
+            })),
+            Err(err) => Ok(Response::new(ReplacementRecoveryStatusResponse {
+                success: false,
+                recovery_status: Bytes::new(),
+                error_info: Some(err),
+            })),
+        }
+    }
+
     async fn get_metacache_listing(
         &self,
         _request: Request<GetMetacacheListingRequest>,

rustfs/src/storage/rpc/node_service/heal.rs

@@ -27,6 +27,8 @@ use super::super::encode_msgpack_map;
 
 const NODE_HEAL_STATUS_VERSION: u8 = 1;
 const NODE_HEAL_STATUS_MAX_SIZE: usize = 64 * 1024;
+const NODE_REPLACEMENT_RECOVERY_STATUS_VERSION: u8 = 1;
+const NODE_REPLACEMENT_RECOVERY_STATUS_MAX_SIZE: usize = 64 * 1024;
 const HEAL_TOPOLOGY_FINGERPRINT_DOMAIN: &[u8] = b"rustfs-heal-topology-v1\0";
 
 fn chrono_to_jiff_timestamp(timestamp: chrono::DateTime<chrono::Utc>) -> Timestamp {
@@ -286,18 +288,115 @@ pub(crate) fn decode_node_heal_status(data: &[u8]) -> Result<NodeHealStatusSnaps
     Ok(snapshot)
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub(crate) struct NodeReplacementRecoveryStatusSnapshot {
+    version: u8,
+    pub snapshot: rustfs_heal::ReplacementRecoverySnapshot,
+}
+
+impl NodeReplacementRecoveryStatusSnapshot {
+    pub(crate) fn new(snapshot: rustfs_heal::ReplacementRecoverySnapshot) -> Self {
+        Self {
+            version: NODE_REPLACEMENT_RECOVERY_STATUS_VERSION,
+            snapshot,
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct NodeReplacementRecoveryStatusSnapshotWire {
+    version: u8,
+    snapshot: ReplacementRecoverySnapshotWire,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct ReplacementRecoverySnapshotWire {
+    records: Vec<ReplacementRecoveryRecordWire>,
+    definitive: bool,
+    reason: Option<String>,
+}
+
+impl From<ReplacementRecoverySnapshotWire> for rustfs_heal::ReplacementRecoverySnapshot {
+    fn from(snapshot: ReplacementRecoverySnapshotWire) -> Self {
+        Self {
+            records: snapshot.records.into_iter().map(Into::into).collect(),
+            definitive: snapshot.definitive,
+            reason: snapshot.reason,
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct ReplacementRecoveryRecordWire {
+    task_id: String,
+    state: rustfs_heal::ReplacementRecoveryState,
+    generation: Option<String>,
+    set_disk_id: Option<String>,
+    target_slots: Vec<String>,
+    reason: Option<String>,
+    verified_at: Option<u64>,
+}
+
+impl From<ReplacementRecoveryRecordWire> for rustfs_heal::ReplacementRecoveryRecord {
+    fn from(record: ReplacementRecoveryRecordWire) -> Self {
+        Self {
+            task_id: record.task_id,
+            state: record.state,
+            generation: record.generation,
+            set_disk_id: record.set_disk_id,
+            target_slots: record.target_slots,
+            reason: record.reason,
+            verified_at: record.verified_at,
+        }
+    }
+}
+
+pub(crate) async fn capture_node_replacement_recovery_status() -> NodeReplacementRecoveryStatusSnapshot {
+    NodeReplacementRecoveryStatusSnapshot::new(rustfs_heal::current_replacement_recovery_snapshot().await)
+}
+
+pub(crate) fn encode_node_replacement_recovery_status(
+    snapshot: &NodeReplacementRecoveryStatusSnapshot,
+) -> Result<Vec<u8>, String> {
+    encode_msgpack_map(snapshot).map_err(|err| format!("failed to encode replacement recovery status: {err}"))
+}
+
+pub(crate) fn decode_node_replacement_recovery_status(data: &[u8]) -> Result<NodeReplacementRecoveryStatusSnapshot, String> {
+    if data.len() > NODE_REPLACEMENT_RECOVERY_STATUS_MAX_SIZE {
+        return Err("replacement recovery status exceeds size limit".to_string());
+    }
+    let mut deserializer = Deserializer::new(Cursor::new(data));
+    let wire = NodeReplacementRecoveryStatusSnapshotWire::deserialize(&mut deserializer)
+        .map_err(|err| format!("failed to decode replacement recovery status: {err}"))?;
+    if usize::try_from(deserializer.get_ref().position()).ok() != Some(data.len()) {
+        return Err("replacement recovery status contains trailing data".to_string());
+    }
+    if wire.version != NODE_REPLACEMENT_RECOVERY_STATUS_VERSION {
+        return Err(format!("unsupported replacement recovery status version: {}", wire.version));
+    }
+    Ok(NodeReplacementRecoveryStatusSnapshot {
+        version: wire.version,
+        snapshot: wire.snapshot.into(),
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::{
-        NODE_HEAL_STATUS_MAX_SIZE, NODE_HEAL_STATUS_VERSION, NodeHealProgress, NodeHealStatusSnapshot, decode_node_heal_status,
-        encode_node_heal_status, heal_control_coordinator, heal_topology_fingerprint,
+        NODE_HEAL_STATUS_MAX_SIZE, NODE_HEAL_STATUS_VERSION, NodeHealProgress, NodeHealStatusSnapshot,
+        NodeReplacementRecoveryStatusSnapshot, decode_node_heal_status, decode_node_replacement_recovery_status,
+        encode_node_heal_status, encode_node_replacement_recovery_status, heal_control_coordinator, heal_topology_fingerprint,
     };
     use crate::storage::storage_api::{
         Endpoint,
         ecstore_layout::{EndpointServerPools, Endpoints, PoolEndpoints},
     };
     use chrono::SecondsFormat;
-    use rustfs_heal::HealOperationsSnapshot;
+    use rustfs_heal::{HealOperationsSnapshot, ReplacementRecoveryRecord, ReplacementRecoverySnapshot, ReplacementRecoveryState};
     use rustfs_scanner::scanner::BackgroundHealInfo;
 
     fn topology_endpoints(last_host: &str) -> EndpointServerPools {
@@ -566,4 +665,72 @@ mod tests {
                 .contains("size limit")
         );
     }
+
+    #[test]
+    fn node_replacement_recovery_status_round_trip_is_versioned() {
+        let snapshot = NodeReplacementRecoveryStatusSnapshot::new(ReplacementRecoverySnapshot {
+            records: vec![ReplacementRecoveryRecord {
+                task_id: "11111111-1111-4111-8111-111111111111".to_string(),
+                state: ReplacementRecoveryState::Completed,
+                generation: Some("11111111-1111-4111-8111-111111111111".to_string()),
+                set_disk_id: Some("pool_0_set_0".to_string()),
+                target_slots: vec!["http://node-a:9000/mnt/disk1".to_string()],
+                reason: None,
+                verified_at: Some(42),
+            }],
+            definitive: true,
+            reason: None,
+        });
+
+        let encoded = encode_node_replacement_recovery_status(&snapshot).expect("snapshot should encode");
+        let decoded = decode_node_replacement_recovery_status(&encoded).expect("snapshot should decode");
+
+        assert_eq!(decoded, snapshot);
+    }
+
+    #[test]
+    fn node_replacement_recovery_status_rejects_unknown_trailing_and_oversized_data() {
+        let mut snapshot = NodeReplacementRecoveryStatusSnapshot::new(ReplacementRecoverySnapshot {
+            records: Vec::new(),
+            definitive: true,
+            reason: None,
+        });
+        snapshot.version += 1;
+        let encoded = encode_node_replacement_recovery_status(&snapshot).expect("snapshot should encode");
+        assert!(
+            decode_node_replacement_recovery_status(&encoded)
+                .expect_err("unknown version should fail closed")
+                .contains("unsupported replacement recovery status version")
+        );
+
+        snapshot.version = super::NODE_REPLACEMENT_RECOVERY_STATUS_VERSION;
+        let mut encoded = encode_node_replacement_recovery_status(&snapshot).expect("snapshot should encode");
+        encoded.push(0);
+        assert!(
+            decode_node_replacement_recovery_status(&encoded)
+                .expect_err("trailing data must fail")
+                .contains("trailing")
+        );
+        assert!(
+            decode_node_replacement_recovery_status(&vec![0; super::NODE_REPLACEMENT_RECOVERY_STATUS_MAX_SIZE + 1])
+                .expect_err("oversized status must fail")
+                .contains("size limit")
+        );
+
+        let fixture = serde_json::json!({
+            "version": super::NODE_REPLACEMENT_RECOVERY_STATUS_VERSION,
+            "snapshot": {
+                "records": [],
+                "definitive": true,
+                "reason": null,
+                "futureField": true,
+            }
+        });
+        let encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
+        assert!(
+            decode_node_replacement_recovery_status(&encoded)
+                .expect_err("nested unknown field must fail")
+                .contains("futureField")
+        );
+    }
 }