fix(rpc): recover tier reloads after peer capability revocation (#7920)

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

@@ -1237,6 +1237,195 @@ mod tests {
         clear_peer_capability(audience);
     }
 
+    #[tokio::test]
+    async fn tier_reload_recovers_only_with_authenticated_replay_capability() {
+        use super::super::peer_rest_client::{PeerRestClient, TierConfigReloadOutcome};
+        use bytes::Bytes;
+        use http_body_util::StreamBody;
+        use hyper::{body::Frame, service::service_fn};
+        use hyper_util::rt::{TokioExecutor, TokioIo};
+        use rustfs_protos::proto_gen::node_service::{LoadTransitionTierConfigResponse, PingResponse};
+        use std::sync::{
+            Arc,
+            atomic::{AtomicUsize, Ordering},
+        };
+        use tonic_prost::prost::Message;
+
+        ensure_test_rpc_secret();
+        let server_epoch = crate::cluster::rpc::http_auth::tonic_rpc_boot_epoch();
+        for (proof, restarted) in [
+            ("valid", false),
+            ("valid", true),
+            ("missing", false),
+            ("missing", true),
+            ("invalid", false),
+            ("invalid", true),
+            ("rejected", false),
+            ("rejected", true),
+        ] {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind test peer");
+            let address = listener.local_addr().expect("test peer address");
+            let audience = address.to_string();
+            let client = PeerRestClient::new(
+                rustfs_utils::XHost {
+                    name: "127.0.0.1".to_string(),
+                    port: address.port(),
+                    is_port_set: true,
+                },
+                format!("http://{address}"),
+            );
+            let old_epoch = if restarted { Uuid::new_v4() } else { server_epoch };
+            let pinned = PeerReplayState {
+                boot_epoch: Some(old_epoch),
+                cache_capability: Some(PeerReplayCapability::Capable { boot_epoch: old_epoch }),
+            };
+            set_peer_capability(&audience, pinned);
+            apply_peer_replay_response(audience.clone(), pinned, Err(std::io::Error::other("startup response omitted proof")));
+            let pings = Arc::new(AtomicUsize::new(0));
+            let reloads = Arc::new(AtomicUsize::new(0));
+            let server_audience = audience.clone();
+            let server_pings = pings.clone();
+            let server_reloads = reloads.clone();
+            let server = tokio::spawn(async move {
+                let (stream, _) = listener.accept().await.expect("accept test client");
+                let service = service_fn(move |request: HttpRequest<hyper::body::Incoming>| {
+                    let audience = server_audience.clone();
+                    let pings = server_pings.clone();
+                    let reloads = server_reloads.clone();
+                    async move {
+                        let authentication = crate::cluster::rpc::verify_tonic_rpc_signature_with_bootstrap(
+                            &audience,
+                            request.uri().path(),
+                            request.headers(),
+                            request.uri().path() == "/node_service.NodeService/Ping",
+                        );
+                        let stale_epoch = authentication
+                            .as_ref()
+                            .is_err_and(|error| error.to_string() == "RPC boot epoch is stale");
+                        assert!(
+                            authentication.is_ok() || stale_epoch,
+                            "requests must retain valid signatures: {authentication:?}"
+                        );
+                        let challenge = tonic_boot_epoch_challenge(request.headers())
+                            .expect("valid challenge")
+                            .expect("challenge present");
+                        let payload = match request.uri().path() {
+                            "/node_service.NodeService/Ping" => {
+                                pings.fetch_add(1, Ordering::SeqCst);
+                                PingResponse {
+                                    version: 1,
+                                    body: Bytes::new(),
+                                }
+                                .encode_to_vec()
+                            }
+                            "/node_service.NodeService/LoadTransitionTierConfig" => {
+                                reloads.fetch_add(1, Ordering::SeqCst);
+                                assert!(
+                                    request.headers().contains_key(RPC_CONTENT_SHA256_HEADER),
+                                    "recovery must not downgrade body binding"
+                                );
+                                LoadTransitionTierConfigResponse {
+                                    success: true,
+                                    error_info: None,
+                                    error_code: None,
+                                }
+                                .encode_to_vec()
+                            }
+                            path => panic!("unexpected test RPC: {path}"),
+                        };
+                        let mut data = vec![0];
+                        data.extend_from_slice(&u32::try_from(payload.len()).expect("small test protobuf").to_be_bytes());
+                        data.extend_from_slice(&payload);
+                        let mut trailers = http::HeaderMap::new();
+                        trailers.insert(
+                            "grpc-status",
+                            if proof == "rejected" {
+                                "16"
+                            } else if stale_epoch {
+                                "14"
+                            } else {
+                                "0"
+                            }
+                            .parse()
+                            .expect("status header"),
+                        );
+                        if proof == "rejected" {
+                            trailers.insert("grpc-message", "invalid credentials".parse().expect("message header"));
+                        }
+                        let frames: Vec<Result<Frame<Bytes>, std::convert::Infallible>> =
+                            vec![Ok(Frame::data(Bytes::from(data))), Ok(Frame::trailers(trailers))];
+                        let mut response = HttpResponse::new(StreamBody::new(futures::stream::iter(frames)));
+                        response
+                            .headers_mut()
+                            .insert("content-type", "application/grpc".parse().expect("content type"));
+                        if matches!(proof, "valid" | "invalid") {
+                            let mut headers =
+                                tonic_boot_epoch_response_headers(&audience, challenge).expect("sign response proof");
+                            if proof == "invalid" {
+                                headers.insert(
+                                    RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER,
+                                    "forged".parse().expect("forged proof header"),
+                                );
+                            }
+                            response.headers_mut().extend(headers);
+                        }
+                        Ok::<_, std::convert::Infallible>(response)
+                    }
+                });
+                hyper::server::conn::http2::Builder::new(TokioExecutor::new())
+                    .serve_connection(TokioIo::new(stream), service)
+                    .await
+                    .expect("serve test peer");
+            });
+
+            let first =
+                tokio::time::timeout(std::time::Duration::from_secs(10), client.load_transition_tier_config_once_outcome())
+                    .await
+                    .expect("recovery attempt must finish");
+            assert_eq!(reloads.load(Ordering::SeqCst), 0, "revoked mutation must never reach the peer");
+            assert_eq!(
+                pings.load(Ordering::SeqCst),
+                1,
+                "revoked capability must trigger a signed read-only probe"
+            );
+            if proof == "rejected" {
+                assert!(
+                    matches!(first, TierConfigReloadOutcome::Terminal(_)),
+                    "invalid credentials must stay terminal"
+                );
+            } else {
+                assert!(
+                    matches!(
+                        first,
+                        TierConfigReloadOutcome::TransientReconnect(_) | TierConfigReloadOutcome::TransientRetrySameChannel(_)
+                    ),
+                    "a probe response must schedule another guarded reload"
+                );
+                let second =
+                    tokio::time::timeout(std::time::Duration::from_secs(10), client.load_transition_tier_config_once_outcome())
+                        .await
+                        .expect("second reload attempt must finish");
+                if proof == "valid" {
+                    assert!(matches!(second, TierConfigReloadOutcome::Success));
+                    assert_eq!(reloads.load(Ordering::SeqCst), 1);
+                    assert_eq!(pings.load(Ordering::SeqCst), 1);
+                    assert_eq!(peer_replay_state(&audience).boot_epoch, Some(server_epoch));
+                } else {
+                    assert!(matches!(
+                        second,
+                        TierConfigReloadOutcome::TransientReconnect(_) | TierConfigReloadOutcome::TransientRetrySameChannel(_)
+                    ));
+                    assert_eq!(reloads.load(Ordering::SeqCst), 0, "unverified proof must not admit a mutation");
+                    assert_eq!(pings.load(Ordering::SeqCst), 2);
+                    assert_eq!(peer_replay_state(&audience).cache_capability, Some(PeerReplayCapability::Revoked));
+                }
+            }
+            server.abort();
+            let _ = server.await;
+            clear_peer_capability(&audience);
+        }
+    }
+
     #[test]
     fn test_signature_interceptor_binds_audience_from_peer_uri() {
         let interceptor = TonicInterceptor::Signature(gen_tonic_signature_interceptor())

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

@@ -48,14 +48,14 @@ 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, ReplacementRecoveryStatusRequest,
-    ScannerActivityRequest, ScannerActivityResponse, ScannerDirtyUsageSnapshotRequest, ScannerDirtyUsageSnapshotResponse,
-    ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, ScannerPublicationLeaseResponse,
-    ScannerScopedDirtyUsageAckRequest, ScannerScopedDirtyUsageAckResponse, ScannerScopedDirtyUsageEntry, ServerInfoRequest,
-    SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest, StartProfilingRequest, StopRebalanceRequest,
-    TierDailyStatsRequest, TierMutationAbortRequest, TierMutationCommitRequest, TierMutationControlResponse,
-    TierMutationFailureClass, TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient,
-    tier_mutation_control_service_client::TierMutationControlServiceClient,
+    LocalStorageInfoRequest, Mss, PingRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest,
+    ReplacementRecoveryStatusRequest, ScannerActivityRequest, ScannerActivityResponse, ScannerDirtyUsageSnapshotRequest,
+    ScannerDirtyUsageSnapshotResponse, ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest,
+    ScannerPublicationLeaseResponse, ScannerScopedDirtyUsageAckRequest, ScannerScopedDirtyUsageAckResponse,
+    ScannerScopedDirtyUsageEntry, ServerInfoRequest, SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest,
+    StartProfilingRequest, StopRebalanceRequest, TierDailyStatsRequest, TierMutationAbortRequest, TierMutationCommitRequest,
+    TierMutationControlResponse, TierMutationFailureClass, TierMutationPeerState, TierMutationPrepareRequest,
+    node_service_client::NodeServiceClient, tier_mutation_control_service_client::TierMutationControlServiceClient,
 };
 pub use rustfs_protos::{PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS};
 use rustfs_protos::{TierMutationRpcPhase, evict_failed_connection};
@@ -2760,6 +2760,21 @@ impl PeerRestClient {
 
         let response = match client.load_transition_tier_config(request).await {
             Ok(response) => response.into_inner(),
+            Err(status)
+                if status.code() == tonic::Code::Unauthenticated && status.message() == "RPC peer replay capability changed" =>
+            {
+                // A restart can revoke the pinned capability. A signed read-only probe
+                // lets the peer prove its new epoch without bypassing the mutation guard.
+                let mut probe = Request::new(PingRequest {
+                    version: 1,
+                    body: Bytes::new(),
+                });
+                probe.set_timeout(rustfs_protos::heal_control_execution_timeout());
+                return match client.ping(probe).await {
+                    Ok(_) => TierConfigReloadOutcome::TransientRetrySameChannel(status.into()),
+                    Err(probe_status) => tier_config_reload_status_outcome(probe_status),
+                };
+            }
             Err(status) => return tier_config_reload_status_outcome(status),
         };
         if !response.success {

crates/ecstore/src/set_disk/read.rs

@@ -625,6 +625,7 @@ impl SetDisks {
         }
     }
 
+    #[hotpath::measure(impl_type = "SetDisks")]
     pub(super) async fn get_object_info_fileinfo_and_quorum(
         &self,
         bucket: &str,

crates/ecstore/tests/hotpath_cpu_attribution_contract_test.rs

@@ -73,7 +73,7 @@ fn inherent_hotpath_measurements_have_cpu_attribution_types() {
     for function in [
         "read_version_optimized",
         "get_object_fileinfo",
-        "get_object_info_and_quorum",
+        "get_object_info_fileinfo_and_quorum",
         "get_object_with_fileinfo",
         "get_object_decode_reader_with_fileinfo",
         "build_codec_streaming_part_reader",

crates/heal/tests/mrf_partial_write_test.rs

@@ -30,7 +30,7 @@ use tokio::io::AsyncReadExt;
 mod storage_api;
 use storage_api::endpoint_index::{EndpointServerPools, Endpoints, init_local_disks};
 use storage_api::integration::{
-    DiskAPI, DiskStore, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader, RUSTFS_META_BUCKET, ReadOptions,
+    DiskAPI, DiskError, DiskStore, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader, RUSTFS_META_BUCKET, ReadOptions,
 };
 
 const SNAPSHOT_LIMIT: usize = 64 * 1024 * 1024;
@@ -348,15 +348,12 @@ async fn partial_write_ec12_4_ack_rejoin_repairs_versions_and_delete_marker_inne
             );
             assert!(snapshot_contains("new.bin").await);
             assert!(snapshot_contains("versioned.bin").await);
-            manager.stop().await.expect("test manager should stop");
             let before_purge = inspect_local_committed_snapshot(SNAPSHOT_LIMIT)
                 .await
                 .expect("legacy checkpoint must validate")
-                .expect("unverified legacy responsibility must remain")
-                .payload()
-                .to_vec();
+                .expect("unverified legacy responsibility must remain");
             // Deleting an existing marker by VersionId removes a version; it
-            // must not create a new partial-write repair responsibility.
+            // must persist a purge so the returning members cannot resurrect it.
             for path in &env.disk_paths[12..] {
                 tokio::fs::rename(path, path.with_extension("offline"))
                     .await
@@ -380,14 +377,14 @@ async fn partial_write_ec12_4_ack_rejoin_repairs_versions_and_delete_marker_inne
                 )
                 .await
                 .expect("explicit marker purge should retain quorum");
-            assert_eq!(
+            assert!(
                 inspect_local_committed_snapshot(SNAPSHOT_LIMIT)
                     .await
                     .expect("purge checkpoint must validate")
-                    .expect("existing legacy checkpoint must remain")
-                    .payload(),
-                before_purge,
-                "a physical marker purge must not add a marker creation repair"
+                    .expect("purge responsibility must be committed")
+                    .sequence()
+                    > before_purge.sequence(),
+                "a degraded marker purge must commit a successor checkpoint"
             );
             for (path, disk) in env.disk_paths[12..].iter().zip(&all[12..]) {
                 tokio::fs::remove_file(path).await.expect("remove purge outage sentinel");
@@ -397,6 +394,23 @@ async fn partial_write_ec12_4_ack_rejoin_repairs_versions_and_delete_marker_inne
                 disk.reset_health_for_store_init_retry();
             }
             *set.disks.write().await = all.iter().cloned().map(Some).collect();
+            assert!(
+                wait_until(|| async {
+                    let results = futures::future::join_all(all.iter().map(|disk| async {
+                        disk.read_version("", "partial-versions", "versioned.bin", &marker, &ReadOptions::default())
+                            .await
+                    }))
+                    .await;
+                    results
+                        .iter()
+                        .all(|result| matches!(result, Err(DiskError::FileVersionNotFound)))
+                })
+                .await,
+                "MRF must complete the marker purge on the returning members"
+            );
+            assert_payload(&env, "partial-versions", "versioned.bin", Some(&first), &payload1).await;
+            assert_payload(&env, "partial-versions", "versioned.bin", Some(&second), &payload2).await;
+            manager.stop().await.expect("test manager should stop");
         },
     )
     .await;

rustfs/tests/connect_perf_network.rs

@@ -252,17 +252,18 @@ async fn controlled_peer_reports_exact_bytes_duration_latency_and_attributed_fai
     assert!(aggregate.get("peers").is_none(), "frozen aggregate schema has no per-peer field");
 }
 
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn slow_peer_is_attributed_and_stops_at_the_wall_clock_limit() {
     let _guard = TEST_HARNESS_LOCK.lock().await;
     let mut request = request(1);
     request.duration = Duration::from_millis(20);
     request.traffic_bytes_per_peer = 1_000;
-    let started = Instant::now();
+    let started = tokio::time::Instant::now();
     let measurement = measure_network_with_harness(&request, &BlockingHarness, &CancellationToken::new())
         .await
         .expect("typed timeout result");
     assert!(started.elapsed() < Duration::from_millis(100));
+    assert!(started.elapsed() >= request.duration);
     assert_eq!(measurement.result.outcome(), NetworkOutcome::Failed);
     assert_eq!(measurement.result.reason_code(), NetworkReasonCode::CollectionFailed);
     assert!(measurement.result.data().is_none());