fix: recover heal after peer boot epoch changes (#7889)

* test(heal): keep outage objects in the replacement erasure set

* fix(rpc): retry authenticated requests after peer epoch changes

* test(rpc): assert retryable rejection of stale boot epochs

* ci: bound concurrent heal clusters in the full E2E gate

.config/nextest.toml

@@ -638,8 +638,8 @@ path = "junit.xml"
 # Each e2e test spawns its own single-node rustfs server on a random port with
 # an isolated temp dir (crates/e2e_test/src/common.rs), so the set is
 # parallel-safe — the same property e2e-smoke relies on. The exceptions are the
-# 4-disk reliability / degraded-read fault-injection tests and the fixed-port
-# Vault tests, both serialized below.
+# 4-disk reliability / degraded-read fault-injection tests, multi-node heal
+# clusters, and the fixed-port Vault tests, all serialized below.
 [profile.e2e-full]
 default-filter = """
     package(e2e_test)
@@ -668,6 +668,12 @@ test-group = 'e2e-reliability'
 filter = 'package(e2e_test) & test(/^inline_fast_path_cluster_test::/)'
 test-group = 'e2e-inline-boundaries'
 
+# Heal interruption cases start whole multi-node clusters. Bound their overlap
+# just as the nightly lane does, without changing membership or retry policy.
+[[profile.e2e-full.overrides]]
+filter = 'package(e2e_test) & test(/^heal_erasure_disk_rebuild_test::/)'
+test-group = 'e2e-cluster-nightly'
+
 [[profile.e2e-full.overrides]]
 filter = 'package(e2e_test) & (test(/^kms::kms_vault_test::/) | test(/^kms::kms_rekey_sweep_test::/) | test(/^kms::configured_roundtrip_test::test_configured_vault_kms_admin_and_versioned_cleanup$/))'
 test-group = 'e2e-vault'

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -1418,6 +1418,21 @@ mod tests {
             "replacement target must retain only its preformatted topology identity"
         );
 
+        // A multi-set pool can route a successful outage PUT away from the
+        // replacement drive. Identify its set through a witnessed baseline.
+        let target_set_peer_drives = cluster
+            .nodes
+            .iter()
+            .enumerate()
+            .filter(|(node_index, _)| *node_index != 1)
+            .flat_map(|(_, node)| node.data_dirs.iter().map(PathBuf::from))
+            .filter(|drive| object_metadata_exists_on_disk(drive, bucket, &expected_manifests[0].key))
+            .collect::<Vec<_>>();
+        assert!(
+            !outage_target_manifest_required || !target_set_peer_drives.is_empty(),
+            "the replacement erasure set must retain an online baseline shard"
+        );
+
         let outage_payload_seed = 0xf1;
         let max_outage_write_attempts = topology.total_drives().max(1);
         let mut outage_key = None;
@@ -1438,6 +1453,18 @@ mod tests {
             .await;
             match put_result {
                 Ok(Ok(_)) => {
+                    if outage_target_manifest_required
+                        && !target_set_peer_drives
+                            .iter()
+                            .any(|drive| object_metadata_exists_on_disk(drive, bucket, &candidate_key))
+                    {
+                        timeout(
+                            Duration::from_secs(30),
+                            clients[2].delete_object().bucket(bucket).key(&candidate_key).send(),
+                        )
+                        .await??;
+                        continue;
+                    }
                     outage_key = Some(candidate_key);
                     break;
                 }
@@ -1465,6 +1492,14 @@ mod tests {
             }
         };
 
+        assert!(
+            !outage_target_manifest_required
+                || target_set_peer_drives
+                    .iter()
+                    .any(|drive| object_metadata_exists_on_disk(drive, bucket, &outage_key)),
+            "outage object {outage_key} belongs to a different erasure set than the replacement drive"
+        );
+
         let mut outage_peer_erasure_indices = HashSet::new();
         if !outage_write_deferred_until_rejoin {
             for (node_index, node) in cluster.nodes.iter().enumerate() {

crates/e2e_test/src/internode_rpc_signature_e2e_test.rs

@@ -58,10 +58,11 @@
 //!
 //! Every covered handler checks the digest before touching storage, and
 //! `MakeVolume` resolves its disk *after* that check. Aiming at a disk that
-//! cannot exist gives three cleanly separable outcomes with zero side effects
+//! cannot exist gives four cleanly separable outcomes with zero side effects
 //! on the server's real data:
 //!
 //! - `Err(Unauthenticated)` — rejected by `check_auth` (signature layer).
+//! - `Err(Unavailable)` — a valid signature with a stale boot epoch, rejected before execution.
 //! - `Err(PermissionDenied)` — rejected by the handler's body-digest gate.
 //! - `Ok(success: false)` — **authentication passed**; the request reached
 //!   handler logic and only then failed on the bogus disk.
@@ -468,8 +469,8 @@ async fn replay_scope_rejects_replay_path_transplant_and_stale_epoch_e2e() -> Te
     rustfs_protos::evict_failed_connection(&url).await;
     assert_rejected(
         call_make_volume(&url, request.clone(), stale_epoch).await,
-        Code::Unauthenticated,
-        None,
+        Code::Unavailable,
+        Some("RPC boot epoch changed"),
         "a replay-scoped signature captured before the receiving process restart",
     );
 

rustfs/src/server/http.rs

@@ -2446,7 +2446,13 @@ fn check_auth(req: Request<()>) -> std::result::Result<Request<()>, Status> {
             error = %e,
             "RPC signature verification failed"
         );
-        Status::unauthenticated("No valid auth token")
+        if failure_reason == "stale_boot_epoch" {
+            // The signature is valid, but the peer restarted before this request. Reject it
+            // before execution and let the client retry after its authenticated epoch refresh.
+            Status::unavailable("RPC boot epoch changed")
+        } else {
+            Status::unauthenticated("No valid auth token")
+        }
     })?;
 
     let parent_context =
@@ -3675,6 +3681,62 @@ mod tests {
         rustfs_common::set_global_local_node_name(&previous_node_name).await;
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn rpc_auth_stale_boot_epoch_is_retryable_after_signature_verification() {
+        let _ = rustfs_credentials::set_global_rpc_secret("rpc-http-test-secret".to_string());
+        let previous_node_name = rustfs_common::get_global_local_node_name().await;
+        let audience = "127.0.0.1:9000";
+        let path = "/node_service.NodeService/ReadVersion";
+        rustfs_common::set_global_local_node_name(audience).await;
+        let challenge = uuid::Uuid::new_v4();
+        let proof = storage::tonic_boot_epoch_response_headers(audience, challenge).expect("signed epoch proof");
+        let epoch = storage::verify_tonic_boot_epoch_response(audience, challenge, &proof).expect("authenticated epoch");
+        let stale_epoch = uuid::Uuid::from_u128(epoch.as_u128() ^ 1);
+        let signed_headers = |audience: &str, epoch| {
+            let mut headers = storage::gen_tonic_signature_headers(audience, "node_service.NodeService", "ReadVersion", None)
+                .expect("method-bound signature");
+            let replay = storage::gen_tonic_replay_scope_headers(
+                audience,
+                path,
+                headers["x-rustfs-timestamp"].to_str().expect("timestamp"),
+                headers["x-rustfs-content-sha256"].to_str().expect("body digest"),
+                epoch,
+            )
+            .expect("replay-scoped signature");
+            headers.extend(replay);
+            headers
+        };
+        let request = |headers: HeaderMap| {
+            let mut request = Request::new(());
+            request.metadata_mut().as_mut().extend(headers);
+            request.extensions_mut().insert(RpcRequestTarget {
+                uri: format!("http://{audience}{path}").parse().expect("RPC URI"),
+                method: Method::POST,
+            });
+            request
+        };
+
+        let stale = signed_headers(audience, stale_epoch);
+        let error = check_auth(request(stale.clone())).expect_err("a stale epoch must still reject the request");
+        assert_eq!(error.code(), tonic::Code::Unavailable);
+        assert_eq!(error.message(), "RPC boot epoch changed");
+
+        let mut forged = stale;
+        forged.insert("x-rustfs-rpc-signature-v3", HeaderValue::from_static("00"));
+        let error = check_auth(request(forged)).expect_err("a forged stale-epoch signature must remain terminal");
+        assert_eq!(error.code(), tonic::Code::Unauthenticated);
+        let error = check_auth(request(signed_headers("127.0.0.1:9001", stale_epoch)))
+            .expect_err("a stale epoch must not hide an audience mismatch");
+        assert_eq!(error.code(), tonic::Code::Unauthenticated);
+
+        let current = signed_headers(audience, epoch);
+        assert!(check_auth(request(current.clone())).is_ok(), "a fresh authenticated scope must succeed");
+        let error = check_auth(request(current)).expect_err("a same-epoch nonce replay must remain terminal");
+        assert_eq!(error.code(), tonic::Code::Unauthenticated);
+        rustfs_common::set_global_local_node_name(&previous_node_name).await;
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn rpc_auth_rejection_records_failure_reason_metric() {

rustfs/src/storage/storage_api.rs

@@ -558,8 +558,9 @@ pub(crate) mod ecstore_rpc {
     };
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::rpc::{
-        ScannerScopedDirtyUsageAckEntry, build_put_file_auth_trailer, gen_signature_headers, gen_tonic_signature_headers,
-        set_tonic_canonical_body_digest, verify_put_file_capability, verify_tonic_rpc_response_proof,
+        ScannerScopedDirtyUsageAckEntry, build_put_file_auth_trailer, gen_signature_headers, gen_tonic_replay_scope_headers,
+        gen_tonic_signature_headers, set_tonic_canonical_body_digest, verify_put_file_capability,
+        verify_tonic_boot_epoch_response, verify_tonic_rpc_response_proof,
     };
 }
 
@@ -655,11 +656,11 @@ pub(crate) fn try_current_local_node_name() -> Option<String> {
 
 #[cfg(test)]
 pub(crate) use ecstore_rpc::gen_signature_headers;
-#[cfg(test)]
-pub(crate) use ecstore_rpc::gen_tonic_signature_headers;
 pub(crate) use ecstore_rpc::sign_tonic_rpc_response_proof;
 #[cfg(test)]
 pub(crate) use ecstore_rpc::verify_tonic_rpc_response_proof;
+#[cfg(test)]
+pub(crate) use ecstore_rpc::{gen_tonic_replay_scope_headers, gen_tonic_signature_headers, verify_tonic_boot_epoch_response};
 
 pub(crate) const STORAGE_CLASS_SUB_SYS: &str = ecstore_config::com::STORAGE_CLASS_SUB_SYS;
 

rustfs/src/storage_api.rs

@@ -162,7 +162,7 @@ pub(crate) mod server {
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::{
             Endpoint, EndpointServerPools, Endpoints, PeerRestClient, PoolEndpoints, ScannerScopedDirtyUsageAckEntry,
-            gen_signature_headers, gen_tonic_signature_headers,
+            gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, verify_tonic_boot_epoch_response,
         };
 
         pub(crate) mod ecfs {