ci(release): use sm-standard-4 for Linux builds (#7925)

.github/workflows/build.yml

@@ -207,10 +207,10 @@ jobs:
           fi
 
           all='{"include":[
-            {"target_id":"linux-x86_64-musl","os":"sm-standard-2","target":"x86_64-unknown-linux-musl","cross":false,"platform":"linux","rustflags":"","features":""},
-            {"target_id":"linux-aarch64-musl","os":"sm-standard-2","target":"aarch64-unknown-linux-musl","cross":true,"platform":"linux","rustflags":"","features":""},
-            {"target_id":"linux-x86_64-gnu","os":"sm-standard-2","target":"x86_64-unknown-linux-gnu","cross":false,"platform":"linux","rustflags":"","features":"pyroscope"},
-            {"target_id":"linux-aarch64-gnu","os":"sm-standard-2","target":"aarch64-unknown-linux-gnu","cross":true,"platform":"linux","rustflags":"","features":"pyroscope"},
+            {"target_id":"linux-x86_64-musl","os":"sm-standard-4","target":"x86_64-unknown-linux-musl","cross":false,"platform":"linux","rustflags":"","features":""},
+            {"target_id":"linux-aarch64-musl","os":"sm-standard-4","target":"aarch64-unknown-linux-musl","cross":true,"platform":"linux","rustflags":"","features":""},
+            {"target_id":"linux-x86_64-gnu","os":"sm-standard-4","target":"x86_64-unknown-linux-gnu","cross":false,"platform":"linux","rustflags":"","features":"pyroscope"},
+            {"target_id":"linux-aarch64-gnu","os":"sm-standard-4","target":"aarch64-unknown-linux-gnu","cross":true,"platform":"linux","rustflags":"","features":"pyroscope"},
             {"target_id":"macos-aarch64","os":"macos-latest","target":"aarch64-apple-darwin","cross":false,"platform":"macos","rustflags":"","features":"pyroscope"},
             {"target_id":"windows-x86_64","os":"windows-latest","target":"x86_64-pc-windows-msvc","cross":false,"platform":"windows","rustflags":"","features":""}
           ]}'

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/src/store/init.rs

@@ -2981,8 +2981,15 @@ mod tests {
                     assert!(backend.contains(&remote).await, "commit must not delete remote bytes before cleanup");
                     let removed_before = backend.remove_count().await;
 
-                    // Restart before queue delivery. The new runtime must
-                    // reconstruct ownership solely from the committed xl.meta.
+                    // Later matrix cases reuse this restarted store. If its
+                    // expiry worker has already dequeued this failed cleanup,
+                    // drain it before cancelling the context so the next
+                    // store cannot inherit object locks that block recovery.
+                    wait_for_expiry_workers_idle(&store).await;
+
+                    // Restart after the failed cleanup attempt. The new
+                    // runtime must reconstruct ownership from committed
+                    // xl.meta regardless of whether the old queue delivered it.
                     let tier_config = ctx
                         .tier_config_mgr()
                         .read()
@@ -3022,6 +3029,7 @@ mod tests {
                     backend.set_remove_failure(false);
                     // This fixture starts expiry workers without the runtime's
                     // recovery loop, so drive its durable rescan explicitly.
+                    eprintln!("Recovering overwrite cleanup: state={state:?}, suspended={suspended}, copy={self_copy}");
                     wait_for_tier_free_version_recovery(Arc::clone(&store), &backend, removed_before + 1).await;
                     let versions = set
                         .load_file_info_versions_exact(&bucket, object)
@@ -12593,19 +12601,33 @@ mod tests {
         expected_removes: usize,
     ) {
         ExpiryState::resize_workers(1, store.clone()).await;
-        tokio::time::timeout(Duration::from_secs(30), async {
+        let mut last_progress = None;
+        let result = tokio::time::timeout(Duration::from_secs(30), async {
             loop {
                 let stats = recover_tier_free_versions(store.clone(), 100, None, None)
                     .await
                     .expect("tier free-version recovery scan should succeed");
-                if backend.remove_versions().await.len() >= expected_removes && stats.enqueued == 0 && stats.failed == 0 {
+                let removes = backend.remove_versions().await.len();
+                let recovered = removes >= expected_removes && stats.enqueued == 0 && stats.failed == 0;
+                last_progress = Some((removes, stats));
+                if recovered {
                     return;
                 }
                 tokio::time::sleep(Duration::from_millis(50)).await;
             }
         })
-        .await
-        .expect("tier free-version recovery should complete");
+        .await;
+        if result.is_err() {
+            let expiry_state = store.ctx.expiry_state();
+            let workers = expiry_state
+                .try_read()
+                .ok()
+                .map(|state| (state.pending_tasks(), state.active_tasks()));
+            panic!(
+                "tier free-version recovery should complete: expected_removes={expected_removes}, \
+                 last_progress={last_progress:?}, workers(pending, active)={workers:?}"
+            );
+        }
         wait_for_expiry_workers_idle(&store).await;
     }
 

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/src/error.rs

@@ -66,6 +66,12 @@ pub enum Error {
     #[error("Replacement ownership conflict: {0}")]
     ReplacementOwnershipConflict(String),
 
+    #[error("Replacement generation conflict for task {task_id}: {reason}")]
+    ReplacementGenerationConflict { task_id: String, reason: String },
+
+    #[error("Replacement target is not ready: {0}")]
+    ReplacementTargetNotReady(String),
+
     #[error("replacement recovery retry budget exhausted")]
     ReplacementRetryBudgetExhausted,
 
@@ -117,6 +123,7 @@ impl Error {
     pub(crate) fn is_recoverable_heal(&self) -> bool {
         match self {
             Error::TaskCancelled | Error::TaskTimeout | Error::StaleBucketIncarnation { .. } => false,
+            Error::ReplacementTargetNotReady(_) => true,
             Error::TransientSkip { .. } => true,
             // Lock failures classify by LockError's own taxonomy: only the
             // fatal variants (ResourceNotFound / PermissionDenied /
@@ -348,6 +355,11 @@ mod tests {
         assert!(!Error::TaskTimeout.is_recoverable_heal());
     }
 
+    #[test]
+    fn replacement_target_restart_is_recoverable() {
+        assert!(Error::ReplacementTargetNotReady("mount is restarting".to_string()).is_recoverable_heal());
+    }
+
     #[test]
     fn lock_contention_and_transport_variants_stay_recoverable() {
         use rustfs_lock::LockError;

crates/heal/src/heal/erasure_healer.rs

@@ -937,11 +937,28 @@ impl ErasureSetHealer {
         // later heal cycle via the same bounded-retry mechanism as failures —
         // never hot-retried in place here.
         if failed_objects > 0 || skipped_objects > 0 || failed_buckets > 0 {
-            if self.replacement_task_id.is_some() && resume_manager.schedule_retry().await? {
-                checkpoint_manager.reset_for_retry().await?;
-                return Err(Error::transient_skip(format!(
-                    "Replacement erasure set heal incomplete: {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped; retry scheduled"
-                )));
+            if self.replacement_task_id.is_some() {
+                let state = resume_manager.get_state().await;
+                let targets_ready = self
+                    .storage
+                    .replacement_targets_ready_for_retry(
+                        set_disk_id,
+                        &state.replacement_targets,
+                        &state.replacement_target_identities,
+                    )
+                    .await?;
+                if !targets_ready {
+                    // The target may be between process restart, mount
+                    // admission, and format publication. Rewind only the
+                    // object checkpoint so the same generation can retry once
+                    // its identity is ready; do not spend a generation retry
+                    // or release its healing markers.
+                    checkpoint_manager.reset_for_retry().await?;
+                    resume_manager.defer_retry_until_target_ready().await?;
+                    return Err(Error::transient_skip(format!(
+                        "Replacement erasure set heal incomplete: {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped; target readiness deferred retry"
+                    )));
+                }
             }
             if resume_manager.schedule_retry().await? {
                 // Both persistence layers must be reset together: schedule_retry
@@ -966,8 +983,13 @@ impl ErasureSetHealer {
                     state = "retry_scheduled",
                     "Erasure set heal pass finished with unhealed versions; scheduled full re-heal retry"
                 );
+                let prefix = if self.replacement_task_id.is_some() {
+                    "Replacement erasure set heal incomplete"
+                } else {
+                    "Erasure set heal incomplete"
+                };
                 return Err(Error::transient_skip(format!(
-                    "Erasure set heal incomplete: {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped; retry scheduled"
+                    "{prefix}: {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped; retry scheduled"
                 )));
             }
 

crates/heal/src/heal/manager.rs

@@ -15,7 +15,10 @@
 use crate::heal::{
     outcome::HealTaskOutcome,
     progress::{HealProgress, HealStatistics},
-    resume::{ReplacementPhase, ResumeGc, ResumeManager, ResumeState, ResumeUtils},
+    resume::{
+        ReplacementPhase, ReplacementRecoveryCandidate, ResumeGc, ResumeManager, ResumeState, ResumeUtils,
+        merge_replacement_recovery_candidate, replacement_retry_is_exhausted_active,
+    },
     storage::HealStorageAPI,
     task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType, demote_to_debug_when},
 };

crates/heal/src/heal/manager/auto_scan.rs

@@ -47,6 +47,12 @@ impl HealManager {
 
         tokio::spawn(async move {
             let mut interval = interval(duration);
+            let block_recovery_set = |set_disk_id: &str| {
+                replacement_recovery_blocked_sets
+                    .lock()
+                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                    .insert(set_disk_id.to_string());
+            };
 
             loop {
                 let mut candidate_count = 0usize;
@@ -72,7 +78,10 @@ impl HealManager {
                         // Build list of endpoints that need healing
                         let mut endpoints = HashMap::<String, Vec<Endpoint>>::new();
                         let mut durable_recoveries = HashMap::<String, (String, Vec<Endpoint>, Vec<String>, String)>::new();
+                        let mut durable_recovery_candidates = HashMap::<String, ReplacementRecoveryCandidate>::new();
+                        let mut conflicted_recovery_tasks = HashSet::<String>::new();
                         let mut conflicted_recovery_sets = HashSet::<String>::new();
+                        let mut deferred_recovery_sets = HashSet::<String>::new();
                         let mut deferred_replacement_endpoints = HashSet::<String>::new();
                         let local_disks = {
                             let local_disk_map = local_disk_map_read().await;
@@ -317,69 +326,265 @@ impl HealManager {
                                     }
                                 };
                                 let state = resume_manager.get_state().await;
-                                let targets = state
-                                    .replacement_targets
-                                    .iter()
-                                    .filter_map(|target| {
-                                        local_endpoints
-                                            .iter()
-                                            .find(|endpoint| endpoint.to_string() == *target)
-                                            .cloned()
-                                    })
-                                    .collect::<Vec<_>>();
+                                if disk_set_disk_id.as_deref().is_some_and(|disk_set| disk_set != state.set_disk_id) {
+                                    let disk_set = disk_set_disk_id.as_deref().unwrap_or_default();
+                                    conflicted_recovery_tasks.insert(task_id.clone());
+                                    conflicted_recovery_sets.insert(state.set_disk_id.clone());
+                                    conflicted_recovery_sets.insert(disk_set.to_string());
+                                    block_recovery_set(&state.set_disk_id);
+                                    if !disk_set.is_empty() {
+                                        block_recovery_set(disk_set);
+                                    }
+                                    warn!(
+                                        target: "rustfs::heal::manager",
+                                        event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
+                                        component = LOG_COMPONENT_HEAL,
+                                        subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
+                                        endpoint = %endpoint,
+                                        task_id,
+                                        state_set_disk_id = %state.set_disk_id,
+                                        observed_set_disk_id = disk_set,
+                                        "Replacement recovery intent set binding is ambiguous"
+                                    );
+                                    continue;
+                                }
                                 // A remote survivor may own the durable intent, but only the
                                 // node mounting every replacement target may advance or replay it.
-                                if targets.len() != state.replacement_targets.len() {
+                                if state.replacement_targets.iter().any(|target| {
+                                    !local_endpoints.iter().any(|endpoint| endpoint.to_string() == *target)
+                                }) {
+                                    deferred_recovery_sets.insert(state.set_disk_id.clone());
                                     continue;
                                 }
-                                if !durable_replacement_recovery_is_due(&state, &task_id) {
-                                    if durable_replacement_reserves_targets(&state) {
-                                        conflicted_recovery_sets.insert(state.set_disk_id.clone());
-                                    }
+                                if conflicted_recovery_tasks.contains(&task_id) {
                                     continue;
                                 }
-                                let state = match resume_manager.resolve_replacement_recovery(storage.as_ref()).await {
-                                    Ok(state) => state,
-                                    Err(_) => {
+                                let mut selected = durable_recovery_candidates.remove(&task_id);
+                                let previous_set = selected.as_ref().map(|candidate| candidate.state.set_disk_id.clone());
+                                let candidate = match ReplacementRecoveryCandidate::new(
+                                    state.clone(),
+                                    disk.endpoint().to_string(),
+                                    disk.endpoint().is_local,
+                                ) {
+                                    Ok(candidate) => candidate,
+                                    Err(error) => {
+                                        conflicted_recovery_tasks.insert(task_id.clone());
                                         conflicted_recovery_sets.insert(state.set_disk_id.clone());
+                                        if let Some(previous_set) = &previous_set {
+                                            conflicted_recovery_sets.insert(previous_set.clone());
+                                        }
+                                        block_recovery_set(&state.set_disk_id);
+                                        if let Some(previous_set) = &previous_set {
+                                            block_recovery_set(previous_set);
+                                        }
+                                        warn!(
+                                            target: "rustfs::heal::manager",
+                                            event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
+                                            component = LOG_COMPONENT_HEAL,
+                                            subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
+                                            endpoint = %endpoint,
+                                            task_id,
+                                            error = %error,
+                                            "Replacement recovery generation binding is ambiguous"
+                                        );
                                         continue;
                                     }
                                 };
-                                let task_id = state.task_id.clone();
-                                let Some(set_disk_id) = crate::heal::utils::format_set_disk_id_from_i32(
-                                    targets[0].pool_idx,
-                                    targets[0].set_idx,
-                                ) else {
-                                    continue;
-                                };
-                                if targets.iter().any(|target| {
-                                    crate::heal::utils::format_set_disk_id_from_i32(target.pool_idx, target.set_idx)
-                                        .as_deref()
-                                        != Some(set_disk_id.as_str())
-                                }) {
-                                    continue;
-                                }
-                                let resume_endpoint = disk.endpoint().to_string();
-                                match durable_recoveries.get(&set_disk_id) {
-                                    Some((existing_task_id, _, _, existing_anchor))
-                                        if existing_task_id != &task_id || existing_anchor != &resume_endpoint => {
-                                        replacement_recovery_blocked_sets
-                                            .lock()
-                                            .expect("replacement recovery blocked set lock poisoned")
-                                            .insert(set_disk_id.clone());
-                                        conflicted_recovery_sets.insert(set_disk_id);
+                                let task_set = candidate.state.set_disk_id.clone();
+                                if let Err(error) = merge_replacement_recovery_candidate(&mut selected, candidate) {
+                                    conflicted_recovery_tasks.insert(task_id.clone());
+                                    conflicted_recovery_sets.insert(task_set.clone());
+                                    if let Some(previous_set) = &previous_set {
+                                        conflicted_recovery_sets.insert(previous_set.clone());
                                     }
-                                    Some(_) => {}
-                                    None => {
-                                        durable_recoveries.insert(
-                                            set_disk_id,
-                                            (task_id, targets, state.replacement_buckets, resume_endpoint),
-                                        );
+                                    block_recovery_set(&task_set);
+                                    if let Some(previous_set) = &previous_set {
+                                        block_recovery_set(previous_set);
                                     }
+                                    warn!(
+                                        target: "rustfs::heal::manager",
+                                        event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
+                                        component = LOG_COMPONENT_HEAL,
+                                        subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
+                                        endpoint = %endpoint,
+                                        task_id,
+                                        error = %error,
+                                        "Replacement recovery generation merge failed"
+                                    );
+                                } else if let Some(selected) = selected {
+                                    durable_recovery_candidates.insert(task_id, selected);
                                 }
                             }
                         }
 
+                        // Resolve only the canonical copy.  Resolving every
+                        // replica independently can advance separate revisions
+                        // and recreates the duplicate-generation race this scan
+                        // is meant to recover from.
+                        for (task_id, candidate) in durable_recovery_candidates {
+                            // Discovery can observe a remote survivor before it
+                            // confirms that every replacement target is mounted
+                            // locally.  Do not resolve (and therefore mutate)
+                            // that generation in the same pass; the next scan
+                            // must make the readiness decision again.
+                            if deferred_recovery_sets.contains(&candidate.state.set_disk_id)
+                                || conflicted_recovery_sets.contains(&candidate.state.set_disk_id)
+                            {
+                                continue;
+                            }
+                            let Some(anchor_disk) = local_disks
+                                .iter()
+                                .chain(recovery_disks.iter())
+                                .find(|disk| disk.endpoint().to_string() == candidate.anchor)
+                                .cloned()
+                            else {
+                                deferred_recovery_sets.insert(candidate.state.set_disk_id.clone());
+                                continue;
+                            };
+                            let resume_manager = match ResumeManager::load_replacement_intent(anchor_disk, &task_id).await {
+                                Ok(resume_manager) => resume_manager,
+                                Err(error) => {
+                                    conflicted_recovery_sets.insert(candidate.state.set_disk_id.clone());
+                                    block_recovery_set(&candidate.state.set_disk_id);
+                                    warn!(
+                                        target: "rustfs::heal::manager",
+                                        event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
+                                        component = LOG_COMPONENT_HEAL,
+                                        subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
+                                        task_id,
+                                        error = %error,
+                                        "Canonical replacement recovery intent load failed"
+                                    );
+                                    continue;
+                                }
+                            };
+                            let mut state = resume_manager.get_state().await;
+                            if replacement_retry_is_exhausted_active(&state) {
+                                match resume_manager
+                                    .rearm_replacement_recovery_if_needed(storage.as_ref())
+                                    .await
+                                {
+                                    Ok(true) => state = resume_manager.get_state().await,
+                                    Ok(false) => {
+                                        // A ready generation without the
+                                        // compatibility transient-skip marker
+                                        // remains bounded; do not turn it into
+                                        // a fresh retry.
+                                        if durable_replacement_reserves_targets(&state) {
+                                            conflicted_recovery_sets.insert(state.set_disk_id.clone());
+                                            block_recovery_set(&state.set_disk_id);
+                                        }
+                                        continue;
+                                    }
+                                    Err(error) => {
+                                        conflicted_recovery_sets.insert(state.set_disk_id.clone());
+                                        block_recovery_set(&state.set_disk_id);
+                                        warn!(
+                                            target: "rustfs::heal::manager",
+                                            event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
+                                            component = LOG_COMPONENT_HEAL,
+                                            subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
+                                            task_id,
+                                            error = %error,
+                                            "Replacement recovery readiness probe failed"
+                                        );
+                                        continue;
+                                    }
+                                }
+                            }
+                            if !durable_replacement_recovery_is_due(&state, &task_id) {
+                                if durable_replacement_reserves_targets(&state) {
+                                    conflicted_recovery_sets.insert(state.set_disk_id.clone());
+                                    block_recovery_set(&state.set_disk_id);
+                                }
+                                continue;
+                            }
+                            let state = match resume_manager.resolve_replacement_recovery(storage.as_ref()).await {
+                                Ok(state) => state,
+                                Err(Error::ReplacementTargetNotReady(_)) => {
+                                    deferred_recovery_sets.insert(state.set_disk_id.clone());
+                                    continue;
+                                }
+                                Err(error) => {
+                                    conflicted_recovery_sets.insert(state.set_disk_id.clone());
+                                    block_recovery_set(&state.set_disk_id);
+                                    warn!(
+                                        target: "rustfs::heal::manager",
+                                        event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
+                                        component = LOG_COMPONENT_HEAL,
+                                        subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
+                                        task_id,
+                                        error = %error,
+                                        "Canonical replacement recovery resolution failed"
+                                    );
+                                    continue;
+                                }
+                            };
+                            let targets = state
+                                .replacement_targets
+                                .iter()
+                                .filter_map(|target| {
+                                    local_endpoints
+                                        .iter()
+                                        .find(|endpoint| endpoint.to_string() == *target)
+                                        .cloned()
+                                })
+                                .collect::<Vec<_>>();
+                            if targets.len() != state.replacement_targets.len() {
+                                deferred_recovery_sets.insert(state.set_disk_id.clone());
+                                continue;
+                            }
+                            let Some(first_target) = targets.first() else {
+                                conflicted_recovery_sets.insert(state.set_disk_id.clone());
+                                block_recovery_set(&state.set_disk_id);
+                                continue;
+                            };
+                            // A pending handoff may resolve the predecessor to
+                            // its durable successor.  Queue the resolved owner,
+                            // not the stale filename that initiated replay.
+                            let resolved_task_id = state.task_id.clone();
+                            let Some(set_disk_id) = crate::heal::utils::format_set_disk_id_from_i32(
+                                first_target.pool_idx,
+                                first_target.set_idx,
+                            ) else {
+                                conflicted_recovery_sets.insert(state.set_disk_id.clone());
+                                block_recovery_set(&state.set_disk_id);
+                                continue;
+                            };
+                            if state.set_disk_id != set_disk_id
+                                || targets.iter().any(|target| {
+                                    crate::heal::utils::format_set_disk_id_from_i32(target.pool_idx, target.set_idx)
+                                        .as_deref()
+                                        != Some(set_disk_id.as_str())
+                                })
+                            {
+                                block_recovery_set(&set_disk_id);
+                                conflicted_recovery_sets.insert(set_disk_id);
+                                continue;
+                            }
+                            let resume_endpoint = candidate.anchor;
+                            match durable_recoveries.get(&set_disk_id) {
+                                Some((existing_task_id, existing_targets, existing_buckets, _))
+                                    if existing_task_id != &resolved_task_id
+                                        || existing_targets != &targets
+                                        || existing_buckets != &state.replacement_buckets => {
+                                    block_recovery_set(&set_disk_id);
+                                    conflicted_recovery_sets.insert(set_disk_id);
+                                }
+                                Some(_) => {}
+                                None => {
+                                    durable_recoveries.insert(
+                                        set_disk_id,
+                                        (resolved_task_id, targets, state.replacement_buckets, resume_endpoint),
+                                    );
+                                }
+                            }
+                        }
+
+                        for set_disk_id in &deferred_recovery_sets {
+                            durable_recoveries.remove(set_disk_id);
+                            endpoints.remove(set_disk_id);
+                        }
                         for set_disk_id in &conflicted_recovery_sets {
                             durable_recoveries.remove(set_disk_id);
                             endpoints.remove(set_disk_id);

crates/heal/src/heal/manager/tests.rs

@@ -2500,6 +2500,12 @@ fn replacement_recovery_blocks_only_confirmed_conflicts() {
             message: "replacement recovery corruption: malformed legacy intent".to_string(),
         }
     ));
+    assert!(crate::heal::resume::replacement_recovery_error_requires_block(
+        &Error::ReplacementGenerationConflict {
+            task_id: "generation-a".to_string(),
+            reason: "divergent durable copies".to_string(),
+        }
+    ));
     assert!(!crate::heal::resume::replacement_recovery_error_requires_block(&Error::Disk(
         DiskError::Timeout
     )));

crates/heal/src/heal/manager/unclean_shutdown.rs

@@ -83,6 +83,8 @@ impl HealManager {
         let mut set_disk_ids = HashSet::new();
         let mut reserved_replacement_sets = HashSet::new();
         let mut replacement_intents = HashMap::<String, (String, Vec<String>, Vec<String>, String)>::new();
+        let mut replacement_recovery_candidates = HashMap::<String, ReplacementRecoveryCandidate>::new();
+        let mut conflicted_replacement_tasks = HashSet::new();
         let mut conflicted_replacement_sets = HashSet::new();
 
         {
@@ -233,6 +235,31 @@ impl HealManager {
                         }
                     };
                     let state = manager.get_state().await;
+                    if disk_set_disk_id
+                        .as_deref()
+                        .is_some_and(|disk_set| disk_set != state.set_disk_id)
+                    {
+                        let disk_set = disk_set_disk_id.as_deref().unwrap_or_default();
+                        conflicted_replacement_tasks.insert(task_id.clone());
+                        conflicted_replacement_sets.insert(state.set_disk_id.clone());
+                        conflicted_replacement_sets.insert(disk_set.to_string());
+                        self.block_replacement_recovery_set(&state.set_disk_id);
+                        if !disk_set.is_empty() {
+                            self.block_replacement_recovery_set(disk_set);
+                        }
+                        warn!(
+                            target: "rustfs::heal::manager",
+                            event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
+                            component = LOG_COMPONENT_HEAL,
+                            subsystem = LOG_SUBSYSTEM_MANAGER,
+                            endpoint = %endpoint,
+                            task_id,
+                            state_set_disk_id = %state.set_disk_id,
+                            observed_set_disk_id = disk_set,
+                            "Replacement recovery intent set binding is ambiguous"
+                        );
+                        continue;
+                    }
                     if durable_replacement_reserves_targets(&state) {
                         reserved_replacement_sets.insert(state.set_disk_id.clone());
                     }
@@ -247,7 +274,7 @@ impl HealManager {
                                 | ReplacementPhase::HandoffPending
                                 | ReplacementPhase::Rebuilding
                         )
-                        && state.retry_count < state.max_retries;
+                        && (state.retry_count < state.max_retries || replacement_retry_is_exhausted_active(&state));
                     let verified_replacement = state.completed
                         && matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending);
                     if !active_replacement && !verified_replacement && durable_replacement_reserves_targets(&state) {
@@ -257,34 +284,150 @@ impl HealManager {
                         && state.replacement_generation.as_deref() == Some(task_id.as_str())
                         && !state.replacement_targets.is_empty()
                     {
-                        let state = match manager.resolve_replacement_recovery(self.storage.as_ref()).await {
-                            Ok(state) => state,
-                            Err(_) => {
-                                conflicted_replacement_sets.insert(state.set_disk_id.clone());
-                                continue;
-                            }
-                        };
-                        let resume_endpoint = endpoint.to_string();
-                        match replacement_intents.entry(state.task_id.clone()) {
-                            std::collections::hash_map::Entry::Vacant(entry) => {
-                                entry.insert((
-                                    state.set_disk_id,
-                                    state.replacement_targets,
-                                    state.replacement_buckets,
-                                    resume_endpoint,
-                                ));
-                            }
-                            std::collections::hash_map::Entry::Occupied(entry) => {
-                                let (existing_set, existing_targets, existing_buckets, existing_anchor) = entry.get();
-                                if existing_set != &state.set_disk_id
-                                    || existing_targets != &state.replacement_targets
-                                    || existing_buckets != &state.replacement_buckets
-                                    || existing_anchor != &resume_endpoint
-                                {
+                        if conflicted_replacement_tasks.contains(&task_id) {
+                            continue;
+                        }
+                        let mut selected = replacement_recovery_candidates.remove(&task_id);
+                        let previous_set = selected.as_ref().map(|candidate| candidate.state.set_disk_id.clone());
+                        let candidate =
+                            match ReplacementRecoveryCandidate::new(state.clone(), endpoint.to_string(), endpoint.is_local) {
+                                Ok(candidate) => candidate,
+                                Err(error) => {
+                                    conflicted_replacement_tasks.insert(task_id.clone());
                                     conflicted_replacement_sets.insert(state.set_disk_id.clone());
+                                    if let Some(previous_set) = &previous_set {
+                                        conflicted_replacement_sets.insert(previous_set.clone());
+                                    }
                                     self.block_replacement_recovery_set(&state.set_disk_id);
+                                    if let Some(previous_set) = &previous_set {
+                                        self.block_replacement_recovery_set(previous_set);
+                                    }
+                                    warn!(
+                                        target: "rustfs::heal::manager",
+                                        event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
+                                        component = LOG_COMPONENT_HEAL,
+                                        subsystem = LOG_SUBSYSTEM_MANAGER,
+                                        endpoint = %endpoint,
+                                        task_id,
+                                        error = %error,
+                                        "Replacement recovery generation binding is ambiguous"
+                                    );
+                                    continue;
                                 }
+                            };
+                        let task_set = candidate.state.set_disk_id.clone();
+                        if let Err(error) = merge_replacement_recovery_candidate(&mut selected, candidate) {
+                            conflicted_replacement_tasks.insert(task_id.clone());
+                            conflicted_replacement_sets.insert(task_set.clone());
+                            self.block_replacement_recovery_set(&task_set);
+                            if let Some(previous_set) = &previous_set {
+                                conflicted_replacement_sets.insert(previous_set.clone());
+                                self.block_replacement_recovery_set(previous_set);
                             }
+                            warn!(
+                                target: "rustfs::heal::manager",
+                                event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
+                                component = LOG_COMPONENT_HEAL,
+                                subsystem = LOG_SUBSYSTEM_MANAGER,
+                                endpoint = %endpoint,
+                                task_id,
+                                error = %error,
+                                "Replacement recovery generation merge failed"
+                            );
+                        } else if let Some(selected) = selected {
+                            replacement_recovery_candidates.insert(task_id, selected);
+                        }
+                    }
+                }
+            }
+
+            // Replay only the canonical durable copy.  Replica locations are
+            // observations of one generation and must never each advance it.
+            for (task_id, candidate) in replacement_recovery_candidates {
+                let Some(anchor_disk) = recovery_disks
+                    .iter()
+                    .find(|disk| disk.endpoint().to_string() == candidate.anchor)
+                    .cloned()
+                else {
+                    conflicted_replacement_sets.insert(candidate.state.set_disk_id.clone());
+                    self.block_replacement_recovery_set(&candidate.state.set_disk_id);
+                    continue;
+                };
+                let manager = match ResumeManager::load_replacement_intent(anchor_disk, &task_id).await {
+                    Ok(manager) => manager,
+                    Err(error) => {
+                        conflicted_replacement_sets.insert(candidate.state.set_disk_id.clone());
+                        self.block_replacement_recovery_set(&candidate.state.set_disk_id);
+                        warn!(
+                            target: "rustfs::heal::manager",
+                            event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
+                            component = LOG_COMPONENT_HEAL,
+                            subsystem = LOG_SUBSYSTEM_MANAGER,
+                            task_id,
+                            error = %error,
+                            "Canonical replacement recovery intent load failed"
+                        );
+                        continue;
+                    }
+                };
+                let mut state = manager.get_state().await;
+                if replacement_retry_is_exhausted_active(&state) {
+                    match manager.rearm_replacement_recovery_if_needed(self.storage.as_ref()).await {
+                        Ok(true) => state = manager.get_state().await,
+                        Ok(false) => {
+                            // A ready target with an exhausted all-skip pass
+                            // is terminal for this generation; preserve its
+                            // reservation and do not enqueue a fresh task.
+                            conflicted_replacement_sets.insert(state.set_disk_id.clone());
+                            self.block_replacement_recovery_set(&state.set_disk_id);
+                            continue;
+                        }
+                        Err(error) => {
+                            conflicted_replacement_sets.insert(state.set_disk_id.clone());
+                            self.block_replacement_recovery_set(&state.set_disk_id);
+                            warn!(
+                                target: "rustfs::heal::manager",
+                                event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
+                                component = LOG_COMPONENT_HEAL,
+                                subsystem = LOG_SUBSYSTEM_MANAGER,
+                                task_id,
+                                error = %error,
+                                "Replacement recovery readiness probe failed"
+                            );
+                            continue;
+                        }
+                    }
+                }
+                let state = match manager.resolve_replacement_recovery(self.storage.as_ref()).await {
+                    Ok(state) => state,
+                    Err(Error::ReplacementTargetNotReady(_)) => continue,
+                    Err(error) => {
+                        conflicted_replacement_sets.insert(state.set_disk_id.clone());
+                        self.block_replacement_recovery_set(&state.set_disk_id);
+                        warn!(
+                            target: "rustfs::heal::manager",
+                            event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
+                            component = LOG_COMPONENT_HEAL,
+                            subsystem = LOG_SUBSYSTEM_MANAGER,
+                            task_id,
+                            error = %error,
+                            "Canonical replacement recovery resolution failed"
+                        );
+                        continue;
+                    }
+                };
+                let resolved = (state.set_disk_id, state.replacement_targets, state.replacement_buckets, candidate.anchor);
+                match replacement_intents.entry(state.task_id.clone()) {
+                    std::collections::hash_map::Entry::Vacant(entry) => {
+                        entry.insert(resolved);
+                    }
+                    std::collections::hash_map::Entry::Occupied(entry) => {
+                        let existing = entry.get();
+                        if existing.0 != resolved.0 || existing.1 != resolved.1 || existing.2 != resolved.2 {
+                            conflicted_replacement_sets.insert(existing.0.clone());
+                            conflicted_replacement_sets.insert(resolved.0.clone());
+                            self.block_replacement_recovery_set(&resolved.0);
+                            self.block_replacement_recovery_set(&existing.0);
                         }
                     }
                 }

crates/heal/src/heal/replacement_execution.rs

@@ -37,7 +37,7 @@ impl ReplacementExecution {
     pub(crate) async fn acquire(targets: &[String]) -> Result<Arc<Self>> {
         let identities = auto_replacement_target_identities(targets)
             .await
-            .ok_or_else(|| Error::ReplacementOwnershipConflict("target mount admission failed".to_string()))?;
+            .ok_or_else(|| Error::ReplacementTargetNotReady("target mount admission failed".to_string()))?;
         let local_disks = ecstore_local_disk_map_read()
             .await
             .values()
@@ -51,12 +51,12 @@ impl ReplacementExecution {
         for identity in &identities {
             let disk = replacement_target_disk(&identity.endpoint, &local_disks)
                 .await
-                .ok_or_else(|| Error::ReplacementOwnershipConflict("replacement target is unavailable".to_string()))?;
+                .ok_or_else(|| Error::ReplacementTargetNotReady("replacement target is unavailable".to_string()))?;
             leases.push(disk.acquire_replacement_execution_lease().await?);
             disks.push(disk);
         }
         if auto_replacement_target_identities(targets).await.as_ref() != Some(&identities) {
-            return Err(Error::ReplacementOwnershipConflict(
+            return Err(Error::ReplacementTargetNotReady(
                 "target changed while acquiring execution leases".to_string(),
             ));
         }

crates/heal/src/heal/resume.rs

@@ -44,10 +44,15 @@ use replacement::replacement_targets_match_identities;
 pub use replacement::{
     ReplacementPhase, ReplacementRecoveryRecord, ReplacementRecoveryState, ReplacementTargetIdentity, compose_key,
 };
+pub(crate) use replacement::{
+    ReplacementRecoveryCandidate, merge_replacement_recovery_candidate, replacement_retry_is_exhausted_active,
+    replacement_retry_is_legacy_transient_skip,
+};
 pub use utils::ResumeUtils;
 
 const LOG_COMPONENT_HEAL: &str = "heal";
 const LOG_SUBSYSTEM_RESUME: &str = "resume";
+pub(crate) const REPLACEMENT_TARGET_READINESS_DEFERRED: &str = "Replacement target readiness deferred; retry pending";
 const EVENT_HEAL_RESUME_STATE: &str = "heal_resume_state";
 
 /// resume state file constants
@@ -128,7 +133,7 @@ pub(crate) fn replacement_recovery_error_requires_block(error: &Error) -> bool {
         Error::TaskExecutionFailed { message }
             if message.starts_with(REPLACEMENT_RECOVERY_CONFLICT_PREFIX)
                 || message.starts_with(REPLACEMENT_RECOVERY_CORRUPTION_PREFIX)
-    )
+    ) || matches!(error, Error::ReplacementGenerationConflict { .. })
 }
 
 fn replacement_recovery_corruption_for_state_load(message: impl std::fmt::Display, error: Error) -> Error {
@@ -345,6 +350,17 @@ pub struct ResumeState {
     pub replacement_legacy_import: Option<LegacyReplacementApproval>,
     #[serde(default)]
     pub replacement_legacy_successor: Option<String>,
+    /// Whether the legacy transient-skip compatibility path is no longer
+    /// applicable to this generation (or has already been consumed). New
+    /// generations set this true; older schema-seven snapshots default false
+    /// and may receive one bounded compatibility re-arm.
+    #[serde(default)]
+    pub replacement_legacy_retry_compatibility_done: bool,
+    /// The current pass was rewound because the replacement mount was not
+    /// ready.  Task-level error persistence must not turn this readiness wait
+    /// back into a consumed retry attempt.
+    #[serde(default)]
+    pub replacement_retry_waiting_for_target: bool,
     /// start time
     pub start_time: u64,
     /// last update time
@@ -421,6 +437,8 @@ impl ResumeState {
             replacement_lineage: Vec::new(),
             replacement_legacy_import: None,
             replacement_legacy_successor: None,
+            replacement_legacy_retry_compatibility_done: false,
+            replacement_retry_waiting_for_target: false,
             start_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(),
             last_update: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(),
             completed: false,
@@ -461,6 +479,9 @@ impl ResumeState {
         state.replacement_generation = Some(task_id);
         state.replacement_phase = ReplacementPhase::Intent;
         state.replacement_execution_protocol = 1;
+        // New generations already use readiness-aware persistence and must
+        // never enter the one-shot compatibility path for legacy snapshots.
+        state.replacement_legacy_retry_compatibility_done = true;
         state
     }
 
@@ -1127,6 +1148,7 @@ impl ResumeManager {
         if !state.can_retry() {
             return Ok(false);
         }
+        state.replacement_retry_waiting_for_target = false;
         state.increment_retry();
         state.reset_for_retry();
         drop(state);
@@ -1134,6 +1156,73 @@ impl ResumeManager {
         Ok(true)
     }
 
+    /// Rewind one replacement pass while its target is not ready, without
+    /// consuming the generation retry budget. The caller resets the companion
+    /// checkpoint first; this method rewinds the resume ledger and keeps the
+    /// durable generation in `Rebuilding` so the existing markers remain owned
+    /// until the target can be admitted again.
+    pub(crate) async fn defer_retry_until_target_ready(&self) -> Result<()> {
+        self.defer_retry_until_target_ready_with_rearm(false).await
+    }
+
+    async fn defer_retry_until_target_ready_with_rearm(&self, legacy_rearm: bool) -> Result<()> {
+        let mut state = self.state.write().await;
+        // Readiness deferral is not a heal attempt. Preserve an in-budget
+        // counter, but reserve one bounded pass when compatibility recovery
+        // has to re-arm an exhausted legacy generation.
+        if state.retry_count >= state.max_retries {
+            state.retry_count = state.max_retries.saturating_sub(1);
+        }
+        if legacy_rearm {
+            state.replacement_legacy_retry_compatibility_done = true;
+        }
+        state.replacement_retry_waiting_for_target = true;
+        state.reset_for_retry();
+        state.error_message = Some(REPLACEMENT_TARGET_READINESS_DEFERRED.to_string());
+        drop(state);
+        self.save_state_strict().await
+    }
+
+    /// Re-arm an exhausted generation when its target is unavailable, or when
+    /// a pre-readiness-aware release left the durable transient-skip marker at
+    /// the retry ceiling.  The compatibility path is one-shot; an exhausted
+    /// ready generation with no such marker remains terminal.
+    pub(crate) async fn rearm_replacement_recovery_if_needed(
+        &self,
+        storage: &dyn crate::heal::storage::HealStorageAPI,
+    ) -> Result<bool> {
+        let state = self.get_state().await;
+        if !replacement_retry_is_exhausted_active(&state) {
+            return Ok(false);
+        }
+        if state.max_retries == 0 {
+            return Ok(false);
+        }
+        let legacy_rearm = replacement_retry_is_legacy_transient_skip(&state);
+        let ready = storage
+            .replacement_targets_ready_for_retry(
+                &state.set_disk_id,
+                &state.replacement_targets,
+                &state.replacement_target_identities,
+            )
+            .await?;
+        if ready && !legacy_rearm {
+            return Ok(false);
+        }
+        // Legacy exhausted passes may already have a checkpoint containing
+        // the skipped object ledger.  Rewinding only the resume summary would
+        // make the next executor suppress those objects again, so clear the
+        // companion checkpoint before publishing the re-armed state.
+        if CheckpointManager::has_checkpoint(&self.disk, &state.task_id).await {
+            CheckpointManager::load_from_disk(self.disk.clone(), &state.task_id)
+                .await?
+                .reset_for_retry()
+                .await?;
+        }
+        self.defer_retry_until_target_ready_with_rearm(legacy_rearm).await?;
+        Ok(true)
+    }
+
     /// cleanup resume state
     pub async fn cleanup(&self) -> Result<()> {
         let state = self.state.read().await;

crates/heal/src/heal/resume/handoff.rs

@@ -104,6 +104,10 @@ impl ResumeManager {
             Err(error) if matches!(&error, Error::Disk(DiskError::Io(io)) if io.kind() == std::io::ErrorKind::WouldBlock) => {
                 Err(error)
             }
+            // A target can be temporarily absent while its process or mount
+            // restarts.  This is a readiness observation, not a failed
+            // generation attempt, so do not consume the durable retry budget.
+            Err(error @ Error::ReplacementTargetNotReady(_)) => Err(error),
             Err(failure) => match self.record_replacement_failure(&failure, attempt).await {
                 Ok(()) => Err(failure),
                 Err(persistence) => Err(Error::ReplacementFailurePersistence {

crates/heal/src/heal/resume/replacement.rs

@@ -168,6 +168,133 @@ pub(super) fn replacement_targets_match_identities(targets: &[String], identitie
         && identities.iter().map(|identity| &identity.endpoint).eq(targets.iter())
 }
 
+/// A durable intent can be visible through more than one survivor after a
+/// restart.  The endpoint that exposes the record is a replica location, not
+/// a generation identity, so discovery must merge those observations before it
+/// decides whether a set is conflicted.
+#[derive(Debug, Clone)]
+pub(crate) struct ReplacementRecoveryCandidate {
+    pub(crate) state: ResumeState,
+    pub(crate) anchor: String,
+    pub(crate) anchor_is_local: bool,
+}
+
+/// An exhausted replacement data pass is eligible for a readiness probe.
+/// Handoff-pending generations stay on their durable handoff protocol; an
+/// ordinary rebuild may re-arm only for a currently unavailable target or the
+/// one-shot legacy transient-skip compatibility marker.
+pub(crate) fn replacement_retry_is_exhausted_active(state: &ResumeState) -> bool {
+    !state.completed
+        && state.retry_count >= state.max_retries
+        && replacement_targets_match_identities(&state.replacement_targets, &state.replacement_target_identities)
+        && matches!(
+            state.replacement_phase,
+            ReplacementPhase::Intent | ReplacementPhase::OwnershipPending | ReplacementPhase::Rebuilding
+        )
+}
+
+/// Releases before readiness-aware retry persistence encoded a target restart
+/// as an ordinary transient-skip retry.  Such a state is recoverable once
+/// after the same generation is rediscovered, but must not reopen an already
+/// bounded generation on every scanner tick.
+pub(crate) fn replacement_retry_is_legacy_transient_skip(state: &ResumeState) -> bool {
+    !state.replacement_legacy_retry_compatibility_done
+        && state.error_message.as_deref().is_some_and(|message| {
+            message.starts_with("Transient heal skip:")
+                && message.contains("Replacement erasure set heal incomplete")
+                && message.contains("retry scheduled")
+                && !message.contains("target readiness deferred retry")
+        })
+}
+
+impl ReplacementRecoveryCandidate {
+    pub(crate) fn new(state: ResumeState, anchor: impl Into<String>, anchor_is_local: bool) -> Result<Self> {
+        let candidate = Self {
+            state,
+            anchor: anchor.into(),
+            anchor_is_local,
+        };
+        candidate.validate()?;
+        Ok(candidate)
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_replacement_intent(&self.state)
+            || self.state.replacement_generation.as_deref() != Some(self.state.task_id.as_str())
+            || crate::heal::utils::parse_set_disk_id(&self.state.set_disk_id).is_err()
+            || !replacement_targets_match_identities(&self.state.replacement_targets, &self.state.replacement_target_identities)
+        {
+            return Err(Error::ReplacementGenerationConflict {
+                task_id: self.state.task_id.clone(),
+                reason: "durable intent does not bind one generation to its target identities".to_string(),
+            });
+        }
+        Ok(())
+    }
+}
+
+/// Merge one durable observation into the generation selected for a set.
+///
+/// `replacement_revision` is a per-generation CAS fence.  A higher revision
+/// wins; equal revisions must describe the exact same state, otherwise the
+/// on-disk copies cannot be ordered safely and the caller must surface a typed
+/// conflict.  Local anchors are preferred only when the durable revisions and
+/// state are identical, followed by endpoint order for deterministic replay.
+pub(crate) fn merge_replacement_recovery_candidate(
+    selected: &mut Option<ReplacementRecoveryCandidate>,
+    candidate: ReplacementRecoveryCandidate,
+) -> Result<()> {
+    candidate.validate()?;
+    let Some(current) = selected.as_ref() else {
+        *selected = Some(candidate);
+        return Ok(());
+    };
+
+    if !same_replacement_generation_binding(&current.state, &candidate.state) {
+        return Err(Error::ReplacementGenerationConflict {
+            task_id: candidate.state.task_id,
+            reason: "durable copies disagree on set, target slot, target incarnation, or lineage".to_string(),
+        });
+    }
+
+    let candidate_revision = candidate.state.replacement_revision;
+    let current_revision = current.state.replacement_revision;
+    if candidate_revision == current_revision && candidate.state != current.state {
+        return Err(Error::ReplacementGenerationConflict {
+            task_id: candidate.state.task_id,
+            reason: format!("durable copies diverge at replacement revision {candidate_revision}"),
+        });
+    }
+
+    let candidate_wins = candidate_revision > current_revision
+        || (candidate_revision == current_revision
+            && (candidate.anchor_is_local, std::cmp::Reverse(candidate.anchor.as_str()))
+                > (current.anchor_is_local, std::cmp::Reverse(current.anchor.as_str())));
+    if candidate_wins {
+        *selected = Some(candidate);
+    }
+    Ok(())
+}
+
+/// Fields that identify a generation independently of its mutable progress.
+/// Handoff phase and counters intentionally remain mutable and are ordered by
+/// the revision fence above.
+fn same_replacement_generation_binding(left: &ResumeState, right: &ResumeState) -> bool {
+    left.task_id == right.task_id
+        && left.task_type == right.task_type
+        && left.schema_version == right.schema_version
+        && left.set_disk_id == right.set_disk_id
+        && left.replacement_targets == right.replacement_targets
+        && left.replacement_target_identities == right.replacement_target_identities
+        && left.replacement_buckets == right.replacement_buckets
+        && left.replacement_generation == right.replacement_generation
+        && left.replacement_execution_protocol == right.replacement_execution_protocol
+        && left.replacement_predecessor == right.replacement_predecessor
+        && left.replacement_lineage == right.replacement_lineage
+        && left.replacement_legacy_import == right.replacement_legacy_import
+        && left.max_retries == right.max_retries
+}
+
 /// Stable evidence for the mounted replacement instance that owns a repair
 /// generation. Endpoint text alone is not sufficient because a later disk can
 /// be mounted at the same configured path.
@@ -367,6 +494,8 @@ impl ResumeManager {
         }
         state.replacement_phase = ReplacementPhase::Rebuilding;
         state.error_message = None;
+        state.replacement_legacy_retry_compatibility_done = true;
+        state.replacement_retry_waiting_for_target = false;
         state.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
         drop(state);
         self.save_state_strict().await
@@ -377,9 +506,27 @@ impl ResumeManager {
         if state.replacement_generation.as_deref() != Some(state.task_id.as_str()) {
             return Err(replacement_recovery_conflict("replacement failure has no matching generation"));
         }
+        let readiness_deferred = matches!(error, Error::ReplacementTargetNotReady(_))
+            || (state.replacement_retry_waiting_for_target
+                && matches!(error, Error::TransientSkip { message } if message.contains("target readiness deferred retry")));
+        if readiness_deferred {
+            if state.retry_count >= state.max_retries && state.max_retries > 0 {
+                state.retry_count = state.max_retries.saturating_sub(1);
+            }
+            state.replacement_retry_waiting_for_target = true;
+            state.error_message = Some(super::REPLACEMENT_TARGET_READINESS_DEFERRED.to_string());
+            state.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+            drop(state);
+            return self.save_state_strict().await;
+        }
         state.error_message = Some(error.to_string());
         state.retry_count = state.retry_count.max(retry_attempts);
-        if matches!(error, Error::ReplacementRetryBudgetExhausted | Error::ReplacementOwnershipConflict(_)) {
+        if matches!(
+            error,
+            Error::ReplacementRetryBudgetExhausted
+                | Error::ReplacementOwnershipConflict(_)
+                | Error::ReplacementGenerationConflict { .. }
+        ) {
             state.retry_count = state.max_retries;
         }
         state.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();

crates/heal/src/heal/resume/tests.rs

@@ -85,6 +85,137 @@ fn replacement_intent_binds_a_generation_before_format() {
     );
 }
 
+fn replacement_candidate_state(task_id: &str) -> ResumeState {
+    ResumeState::replacement_intent(
+        task_id.to_string(),
+        "erasure_set".to_string(),
+        "pool_0_set_0".to_string(),
+        vec!["bucket-a".to_string()],
+        vec!["replacement-a".to_string()],
+        vec![ReplacementTargetIdentity {
+            endpoint: "replacement-a".to_string(),
+            canonical_path: "/mnt/replacement-a".to_string(),
+            physical_device_ids: vec!["device-a".to_string()],
+            filesystem_identity: "1:2:3".to_string(),
+        }],
+    )
+}
+
+#[test]
+fn replacement_recovery_merges_same_generation_replicas() {
+    let task_id = Uuid::new_v4().to_string();
+    let mut stale = replacement_candidate_state(&task_id);
+    stale.replacement_revision = 4;
+    let mut current = None;
+    merge_replacement_recovery_candidate(
+        &mut current,
+        ReplacementRecoveryCandidate::new(stale, "http://survivor-a/recovery", false).expect("valid candidate"),
+    )
+    .expect("first candidate should be admitted");
+
+    let mut latest = replacement_candidate_state(&task_id);
+    latest.replacement_revision = 5;
+    latest.processed_objects = 7;
+    merge_replacement_recovery_candidate(
+        &mut current,
+        ReplacementRecoveryCandidate::new(latest, "http://survivor-b/recovery", true).expect("valid candidate"),
+    )
+    .expect("same generation replica should be merged");
+
+    let selected = current.expect("a canonical candidate should remain");
+    assert_eq!(selected.state.replacement_revision, 5);
+    assert_eq!(selected.state.processed_objects, 7);
+    assert_eq!(selected.anchor, "http://survivor-b/recovery");
+}
+
+#[test]
+fn replacement_recovery_rejects_ambiguous_equal_revision_copies() {
+    let task_id = Uuid::new_v4().to_string();
+    let mut left = replacement_candidate_state(&task_id);
+    left.replacement_revision = 2;
+    let mut right = left.clone();
+    right.processed_objects = 1;
+    let mut current = None;
+    merge_replacement_recovery_candidate(
+        &mut current,
+        ReplacementRecoveryCandidate::new(left, "survivor-a", false).expect("valid candidate"),
+    )
+    .expect("first candidate should be admitted");
+    let error = merge_replacement_recovery_candidate(
+        &mut current,
+        ReplacementRecoveryCandidate::new(right, "survivor-b", false).expect("valid candidate"),
+    )
+    .expect_err("equal revisions with divergent progress must remain a typed conflict");
+    assert!(matches!(error, Error::ReplacementGenerationConflict { .. }));
+}
+
+#[test]
+fn replacement_recovery_rejects_different_generation_binding() {
+    let task_id = Uuid::new_v4().to_string();
+    let first = replacement_candidate_state(&task_id);
+    let mut second = first.clone();
+    second.replacement_targets = vec!["replacement-b".to_string()];
+    second.replacement_target_identities[0].endpoint = "replacement-b".to_string();
+    let mut current = None;
+    merge_replacement_recovery_candidate(
+        &mut current,
+        ReplacementRecoveryCandidate::new(first, "survivor-a", true).expect("valid candidate"),
+    )
+    .expect("first candidate should be admitted");
+    let error = merge_replacement_recovery_candidate(
+        &mut current,
+        ReplacementRecoveryCandidate::new(second, "survivor-b", true).expect("candidate shape is still valid"),
+    )
+    .expect_err("different target slots must not be merged");
+    assert!(matches!(error, Error::ReplacementGenerationConflict { .. }));
+}
+
+#[test]
+fn exhausted_replacement_is_probeable_without_losing_failed_progress() {
+    let task_id = Uuid::new_v4().to_string();
+    let mut state = replacement_candidate_state(&task_id);
+    state.replacement_phase = ReplacementPhase::Rebuilding;
+    state.retry_count = state.max_retries;
+    state.skipped_objects = 24;
+    assert!(replacement_retry_is_exhausted_active(&state));
+
+    state.failed_objects = 1;
+    assert!(replacement_retry_is_exhausted_active(&state));
+}
+
+#[test]
+fn exhausted_legacy_transient_skip_is_rearmed_only_once() {
+    let task_id = Uuid::new_v4().to_string();
+    let mut state = replacement_candidate_state(&task_id);
+    state.replacement_phase = ReplacementPhase::Rebuilding;
+    state.retry_count = state.max_retries;
+    state.error_message = Some(
+        "Transient heal skip: Replacement erasure set heal incomplete: 0 bucket(s) failed, 0 object(s) failed, 24 object(s) skipped; retry scheduled"
+            .to_string(),
+    );
+    state.replacement_legacy_retry_compatibility_done = false;
+    assert!(replacement_retry_is_legacy_transient_skip(&state));
+
+    state.replacement_legacy_retry_compatibility_done = true;
+    assert!(!replacement_retry_is_legacy_transient_skip(&state));
+}
+
+#[test]
+fn readiness_aware_retry_marker_defaults_old_schema_seven_intents_to_legacy() {
+    let task_id = Uuid::new_v4().to_string();
+    let current = replacement_candidate_state(&task_id);
+    assert!(current.replacement_legacy_retry_compatibility_done);
+
+    let mut encoded = serde_json::to_value(current).expect("serialize current replacement intent");
+    encoded
+        .as_object_mut()
+        .expect("replacement intent is a JSON object")
+        .remove("replacement_legacy_retry_compatibility_done");
+    let legacy: ResumeState = serde_json::from_value(encoded).expect("deserialize prior schema-seven replacement intent");
+
+    assert!(!legacy.replacement_legacy_retry_compatibility_done);
+}
+
 #[tokio::test]
 async fn replacement_terminal_phases_are_durable() {
     use super::super::{DiskOption, Endpoint, new_disk};
@@ -1416,6 +1547,142 @@ fn reset_for_retry_clears_progress_but_keeps_retry_budget() {
     assert_eq!(state.retry_count, 1, "retry budget must be preserved");
 }
 
+#[tokio::test]
+async fn replacement_readiness_deferral_rewinds_progress_without_spending_budget() {
+    let (_temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = ResumeManager::new_replacement_intent(
+        disk.clone(),
+        task_id.clone(),
+        "pool_0_set_0".to_string(),
+        vec!["bucket".to_string()],
+        vec!["replacement-a".to_string()],
+        vec![ReplacementTargetIdentity {
+            endpoint: "replacement-a".to_string(),
+            canonical_path: "/mnt/replacement-a".to_string(),
+            physical_device_ids: vec!["device-a".to_string()],
+            filesystem_identity: "1:2:3".to_string(),
+        }],
+    )
+    .await
+    .expect("replacement intent should persist");
+    {
+        let mut state = manager.state.write().await;
+        state.replacement_phase = ReplacementPhase::Rebuilding;
+        state.processed_objects = 9;
+        state.skipped_objects = 4;
+        state.retry_count = 2;
+    }
+    manager
+        .save_state_strict()
+        .await
+        .expect("progress should persist before deferral");
+
+    manager
+        .defer_retry_until_target_ready()
+        .await
+        .expect("readiness deferral should persist without incrementing retry count");
+    let state = ResumeManager::load_replacement_intent(disk, &task_id)
+        .await
+        .expect("deferred intent should remain loadable")
+        .get_state()
+        .await;
+    assert_eq!(state.retry_count, 2, "readiness deferral must not spend the in-budget counter");
+    assert_eq!(state.processed_objects, 0);
+    assert_eq!(state.skipped_objects, 0);
+    assert_eq!(state.replacement_phase, ReplacementPhase::Rebuilding);
+    assert!(!state.completed);
+    assert!(state.replacement_retry_waiting_for_target);
+    assert_eq!(state.error_message.as_deref(), Some(REPLACEMENT_TARGET_READINESS_DEFERRED));
+}
+
+#[tokio::test]
+async fn exhausted_replacement_deferral_reserves_one_bounded_retry() {
+    let (_temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = ResumeManager::new_replacement_intent(
+        disk.clone(),
+        task_id.clone(),
+        "pool_0_set_0".to_string(),
+        vec!["bucket".to_string()],
+        vec!["replacement-a".to_string()],
+        vec![ReplacementTargetIdentity {
+            endpoint: "replacement-a".to_string(),
+            canonical_path: "/mnt/replacement-a".to_string(),
+            physical_device_ids: vec!["device-a".to_string()],
+            filesystem_identity: "1:2:3".to_string(),
+        }],
+    )
+    .await
+    .expect("replacement intent should persist");
+    {
+        let mut state = manager.state.write().await;
+        state.replacement_phase = ReplacementPhase::Rebuilding;
+        state.retry_count = state.max_retries;
+    }
+    manager.save_state_strict().await.expect("exhausted state should persist");
+
+    manager
+        .defer_retry_until_target_ready()
+        .await
+        .expect("compatibility deferral should persist");
+    manager
+        .record_replacement_failure(
+            &Error::transient_skip("Replacement erasure set heal incomplete: target readiness deferred retry"),
+            99,
+        )
+        .await
+        .expect("task-level persistence must preserve a readiness deferral");
+    let state = ResumeManager::load_replacement_intent(disk, &task_id)
+        .await
+        .expect("deferred intent should remain loadable")
+        .get_state()
+        .await;
+    assert_eq!(state.retry_count, state.max_retries.saturating_sub(1));
+    assert!(state.replacement_legacy_retry_compatibility_done);
+    assert!(state.replacement_retry_waiting_for_target);
+}
+
+#[tokio::test]
+async fn replacement_target_not_ready_failure_does_not_consume_budget() {
+    let (_temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = ResumeManager::new_replacement_intent(
+        disk.clone(),
+        task_id.clone(),
+        "pool_0_set_0".to_string(),
+        vec!["bucket".to_string()],
+        vec!["replacement-a".to_string()],
+        vec![ReplacementTargetIdentity {
+            endpoint: "replacement-a".to_string(),
+            canonical_path: "/mnt/replacement-a".to_string(),
+            physical_device_ids: vec!["device-a".to_string()],
+            filesystem_identity: "1:2:3".to_string(),
+        }],
+    )
+    .await
+    .expect("replacement intent should persist");
+    {
+        let mut state = manager.state.write().await;
+        state.replacement_phase = ReplacementPhase::Rebuilding;
+        state.retry_count = 1;
+    }
+    manager.save_state_strict().await.expect("active state should persist");
+
+    manager
+        .record_replacement_failure(&Error::ReplacementTargetNotReady("target is restarting".to_string()), 99)
+        .await
+        .expect("target readiness should be persisted as a deferral");
+    let state = ResumeManager::load_replacement_intent(disk, &task_id)
+        .await
+        .expect("deferred intent should remain loadable")
+        .get_state()
+        .await;
+    assert_eq!(state.retry_count, 1);
+    assert!(state.replacement_retry_waiting_for_target);
+    assert_eq!(state.error_message.as_deref(), Some(REPLACEMENT_TARGET_READINESS_DEFERRED));
+}
+
 #[test]
 fn can_retry_is_bounded_by_max_retries() {
     let mut state = ResumeState::new("t".to_string(), "erasure_set".to_string(), "pool_0_set_0".to_string(), vec![]);

crates/heal/src/heal/storage.rs

@@ -19,7 +19,7 @@ use rustfs_common::mrf_channel::MrfDeleteMarkerPurge;
 use rustfs_heal_contracts::heal_channel::{DriveState, HealOpts, HealScanMode};
 use rustfs_madmin::heal_commands::HealResultItem;
 use serde::{Deserialize, Serialize};
-use std::sync::Arc;
+use std::{collections::HashSet, sync::Arc};
 use tracing::{debug, error, warn};
 use uuid::Uuid;
 
@@ -31,7 +31,10 @@ use super::storage_api::storage::{
     BucketInfo, BucketOperations, DiskSetSelector, EcstoreHealObjectStorageResult, HealOperations as _, ListOperations as _,
     ObjectIO as _, ObjectOperations as _, StorageAdminApi,
 };
-use super::{DiskError, DiskStore, ECStore, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity};
+use super::{
+    DiskError, DiskStore, ECStore, HealDiskExt as _, StorageError, local_disk_map_read,
+    resume::{ReplacementRecoveryCandidate, ReplacementTargetIdentity, merge_replacement_recovery_candidate},
+};
 pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader};
 
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
@@ -161,6 +164,19 @@ pub enum ReplacementResumeDisk {
     Existing(DiskStore),
 }
 
+fn replacement_resume_conflict_for_other_tasks(
+    requested_task_id: &str,
+    observed_other_tasks: &std::collections::HashSet<String>,
+    set_disk_id: &str,
+) -> Option<Error> {
+    let mut task_ids = observed_other_tasks.iter().collect::<Vec<_>>();
+    task_ids.sort_unstable();
+    task_ids.first().map(|other_task_id| Error::ReplacementGenerationConflict {
+        task_id: requested_task_id.to_string(),
+        reason: format!("durable generation {other_task_id} already owns set {set_disk_id}"),
+    })
+}
+
 pub(crate) fn next_heal_listing_token(
     bucket: &str,
     prefix: &str,
@@ -650,6 +666,20 @@ pub trait HealStorageAPI: Send + Sync {
         Err(Error::other("replacement target identity collection is unsupported"))
     }
 
+    /// Check whether a replacement generation may consume another retry.
+    /// Implementations must verify the mounted incarnation and that every
+    /// target is formatted in the expected erasure-set slot.  The default is
+    /// deliberately permissive for alternate/mock backends that do not expose
+    /// those disk-level probes.
+    async fn replacement_targets_ready_for_retry(
+        &self,
+        _set_disk_id: &str,
+        _targets: &[String],
+        _expected_identities: &[ReplacementTargetIdentity],
+    ) -> Result<bool> {
+        Ok(true)
+    }
+
     async fn replacement_execution(&self, _targets: &[String]) -> Result<Arc<ReplacementExecution>> {
         Err(Error::other("replacement execution lease acquisition is unsupported"))
     }
@@ -1849,39 +1879,106 @@ impl HealStorageAPI for ECStoreHealStorage {
         excluded_targets: &[String],
     ) -> Result<ReplacementResumeDisk> {
         let disks = self.resume_disk_inventory(set_disk_id).await?;
-        let mut existing = None;
-        for prefer_local in [true, false] {
-            for disk_store in &disks {
-                if disk_store.endpoint().is_local != prefer_local
-                    || excluded_targets.contains(&disk_store.endpoint().to_string())
-                    || !matches!(disk_store.get_disk_id().await, Ok(Some(id)) if !id.is_nil())
-                {
+        let mut selected = None;
+        let mut observed_other_tasks = HashSet::new();
+        for disk_store in &disks {
+            if excluded_targets.contains(&disk_store.endpoint().to_string())
+                || !matches!(disk_store.get_disk_id().await, Ok(Some(id)) if !id.is_nil())
+            {
+                continue;
+            }
+            let mut replacement_tasks = super::resume::ResumeUtils::get_replacement_intent_tasks(disk_store).await?;
+            if !replacement_tasks.iter().any(|candidate| candidate == task_id)
+                && super::resume::ResumeManager::has_replacement_intent(disk_store, task_id).await
+            {
+                replacement_tasks.push(task_id.to_string());
+            }
+            if replacement_tasks.is_empty() {
+                continue;
+            }
+            for candidate_task_id in replacement_tasks {
+                // A request with a fresh UUID must not silently create a new
+                // generation while another durable intent already owns this
+                // set. Inspect one copy of each other task and let the
+                // caller reconcile it through the normal recovery scanner.
+                if candidate_task_id != task_id && !observed_other_tasks.insert(candidate_task_id.clone()) {
                     continue;
                 }
-                let replacement_tasks = super::resume::ResumeUtils::get_replacement_intent_tasks(disk_store).await?;
-                let has_intent = replacement_tasks.iter().any(|candidate| candidate == task_id)
-                    || (prefer_local && super::resume::ResumeManager::has_replacement_intent(disk_store, task_id).await);
-                if has_intent && existing.replace(disk_store.clone()).is_some() {
-                    return Err(Error::TaskExecutionFailed {
-                        message: format!("Replacement resume intent is duplicated for set_disk_id: {set_disk_id}"),
+                let manager =
+                    super::resume::ResumeManager::load_replacement_intent(disk_store.clone(), &candidate_task_id).await?;
+                let state = manager.get_state().await;
+                if state.set_disk_id != set_disk_id {
+                    return Err(Error::ReplacementGenerationConflict {
+                        task_id: candidate_task_id,
+                        reason: format!("replacement intent belongs to {}, not {set_disk_id}", state.set_disk_id),
                     });
                 }
-            }
-            // Existing releases only write replacement intents locally, and
-            // this release also prefers a local survivor. Keep that path free
-            // of remote availability and latency; remote discovery is needed
-            // only when no local survivor owns the task.
-            if existing.is_some() {
-                break;
+                if candidate_task_id != task_id {
+                    continue;
+                }
+                let candidate =
+                    ReplacementRecoveryCandidate::new(state, disk_store.endpoint().to_string(), disk_store.endpoint().is_local)?;
+                merge_replacement_recovery_candidate(&mut selected, candidate)?;
             }
         }
-        Ok(existing.map_or(ReplacementResumeDisk::Fresh, ReplacementResumeDisk::Existing))
+
+        let Some(candidate) = selected else {
+            if let Some(error) = replacement_resume_conflict_for_other_tasks(task_id, &observed_other_tasks, set_disk_id) {
+                return Err(error);
+            }
+            return Ok(ReplacementResumeDisk::Fresh);
+        };
+        let disk = disks
+            .iter()
+            .find(|disk| disk.endpoint().to_string() == candidate.anchor)
+            .ok_or_else(|| Error::TaskExecutionFailed {
+                message: format!("Selected replacement resume anchor disappeared for set_disk_id: {set_disk_id}"),
+            })?;
+        Ok(ReplacementResumeDisk::Existing(disk.clone()))
     }
 
     async fn replacement_target_identities(&self, targets: &[String]) -> Result<Vec<ReplacementTargetIdentity>> {
         super::replacement_readiness::auto_replacement_target_identities(targets)
             .await
-            .ok_or_else(|| Error::other("replacement target is not a stable mounted disk"))
+            .ok_or_else(|| Error::ReplacementTargetNotReady("replacement target is not a stable mounted disk".to_string()))
+    }
+
+    async fn replacement_targets_ready_for_retry(
+        &self,
+        set_disk_id: &str,
+        targets: &[String],
+        expected_identities: &[ReplacementTargetIdentity],
+    ) -> Result<bool> {
+        let identities = match self.replacement_target_identities(targets).await {
+            Ok(identities) => identities,
+            Err(_) => return Ok(false),
+        };
+        if identities != expected_identities {
+            return Ok(false);
+        }
+
+        let local_disks = local_disk_map_read()
+            .await
+            .values()
+            .flatten()
+            .filter(|disk| disk.endpoint().is_local)
+            .cloned()
+            .collect::<Vec<_>>();
+        for target in targets {
+            let Some(disk) = local_disks.iter().find(|disk| disk.endpoint().to_string() == *target) else {
+                return Ok(false);
+            };
+            let endpoint = disk.endpoint();
+            if crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx).as_deref()
+                != Some(set_disk_id)
+            {
+                return Ok(false);
+            }
+            if !matches!(disk.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) {
+                return Ok(false);
+            }
+        }
+        Ok(true)
     }
 
     async fn replacement_execution(&self, targets: &[String]) -> Result<Arc<ReplacementExecution>> {
@@ -1894,8 +1991,9 @@ mod tests {
     use super::super::StorageError;
     use super::{
         decode_disk_walk_token, decode_heal_token, encode_disk_walk_token, encode_heal_token, is_transient_object_exists_error,
-        is_transient_object_exists_message, next_heal_listing_token,
+        is_transient_object_exists_message, next_heal_listing_token, replacement_resume_conflict_for_other_tasks,
     };
+    use std::collections::HashSet;
 
     #[test]
     fn object_receipt_requires_integrity_and_resolved_version_evidence() {
@@ -2104,4 +2202,15 @@ mod tests {
             "object".to_string(),
         )));
     }
+
+    #[test]
+    fn fresh_replacement_cannot_ignore_another_durable_generation() {
+        let other = "00000000-0000-4000-8000-000000000001".to_string();
+        let observed = HashSet::from([other]);
+        let error =
+            replacement_resume_conflict_for_other_tasks("00000000-0000-4000-8000-000000000002", &observed, "pool_0_set_0")
+                .expect("an unmatched durable owner must produce a conflict");
+        assert!(matches!(error, crate::Error::ReplacementGenerationConflict { .. }));
+        assert!(replacement_resume_conflict_for_other_tasks("task", &HashSet::new(), "pool_0_set_0").is_none());
+    }
 }

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());