fix(heal): resume remote rebuilds after target restart (#6941)

* fix(heal): retry unavailable recreate targets

* fix(heal): refresh put-file epochs after target restart

* test(e2e): harden heal restart evidence

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(e2e): cancel competing heal before restart

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

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

.config/e2e-nightly-selection.txt

@@ -1 +1 @@
-sha256=26003ce03eca11391d1c080491e4f408526717e4b47967b62db08abe1edd189a
+sha256=9c2b958035a038ffd5ab98cac5f59a1b8e6a16e141f109ec7fb956afc0f11105

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -16,13 +16,14 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::chaos::signed_admin_post;
+    use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post};
-    use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging};
+    use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request, init_logging};
     use aws_sdk_s3::primitives::ByteStream;
+    use http::Method;
     use std::collections::HashSet;
     use std::error::Error;
     use std::path::{Path, PathBuf};
-    use tokio::time::{Duration, sleep, timeout};
+    use tokio::time::{Duration, Instant, sleep, timeout};
     use tracing::info;
 
     fn has_file_under(path: &Path) -> bool {
@@ -48,6 +49,110 @@ mod tests {
         disk.join(bucket).join(key).join("xl.meta").is_file()
     }
 
+    // Healing may rewrite non-identity bookkeeping in xl.meta. The census
+    // therefore compares the canonical selected metadata fields plus every
+    // physical shard, while the payload seed makes object mix-ups observable.
+    #[derive(Debug)]
+    struct PhysicalObjectManifest {
+        key: String,
+        payload_seed: u8,
+        shard_census: VersionShardCensus,
+    }
+
+    fn deterministic_object_body(len: usize, seed: u8) -> Vec<u8> {
+        let mut value = seed;
+        std::iter::repeat_with(|| {
+            value = value.wrapping_mul(31).wrapping_add(17);
+            value
+        })
+        .take(len)
+        .collect()
+    }
+
+    fn matching_manifest_count(
+        disk: &Path,
+        bucket: &str,
+        expected_manifests: &[PhysicalObjectManifest],
+    ) -> Result<usize, Box<dyn Error + Send + Sync>> {
+        let mut matching = 0;
+        for expected in expected_manifests {
+            let actual = census_object_version_on_disk(disk, bucket, &expected.key, None)?;
+            if actual.matches_manifest(&expected.shard_census) {
+                matching += 1;
+            }
+        }
+        Ok(matching)
+    }
+
+    fn metadata_count(disk: &Path, bucket: &str, expected_manifests: &[PhysicalObjectManifest]) -> usize {
+        expected_manifests
+            .iter()
+            .filter(|expected| object_metadata_exists_on_disk(disk, bucket, &expected.key))
+            .count()
+    }
+
+    fn heal_task_status_diagnostic(body: &str) -> String {
+        let Ok(status) = serde_json::from_str::<serde_json::Value>(body) else {
+            return body.to_string();
+        };
+        let items = status["items"].as_array();
+        let mut unresolved_states = HashSet::new();
+        for item in items.into_iter().flatten() {
+            for drive in item["after"]["drives"].as_array().into_iter().flatten() {
+                if let Some(state) = drive["state"].as_str()
+                    && state != "ok"
+                {
+                    unresolved_states.insert(state.to_string());
+                }
+            }
+        }
+        let mut unresolved_states = unresolved_states.into_iter().collect::<Vec<_>>();
+        unresolved_states.sort();
+        format!(
+            "summary={:?}, detail={:?}, item_count={}, unresolved_drive_states={unresolved_states:?}",
+            status["summary"].as_str(),
+            status["detail"].as_str(),
+            items.map_or(0, Vec::len)
+        )
+    }
+
+    fn cluster_heal_is_idle(status: &serde_json::Value) -> bool {
+        let operations = &status["healOperations"];
+        status["clusterStatusComplete"] == serde_json::Value::Bool(true)
+            && status["state"].as_str() == Some("idle")
+            && operations["queueLength"].as_u64() == Some(0)
+            && operations["activeTasks"].as_u64() == Some(0)
+            && operations["retryingTasks"].as_u64() == Some(0)
+    }
+
+    fn only_admin_heal_is_active(status: &serde_json::Value) -> bool {
+        let operations = &status["healOperations"];
+        status["clusterStatusComplete"] == serde_json::Value::Bool(true)
+            && status["state"].as_str() == Some("active")
+            && operations["queueLength"].as_u64() == Some(0)
+            && operations["activeTasks"].as_u64() == Some(1)
+            && operations["retryingTasks"].as_u64() == Some(0)
+            && operations["activeBySource"]["admin"].as_u64() == Some(1)
+    }
+
+    async fn replacement_recovery_status(
+        cluster: &RustFSTestClusterEnvironment,
+    ) -> Result<serde_json::Value, Box<dyn Error + Send + Sync>> {
+        let (status, body) = admin_request(
+            &cluster.nodes[0].url,
+            Method::GET,
+            "/rustfs/admin/v4/heal/replacement-recovery",
+            None,
+            &cluster.access_key,
+            &cluster.secret_key,
+        )
+        .await?;
+        if !status.is_success() {
+            return Err(format!("replacement recovery status failed: {status} {body}").into());
+        }
+        serde_json::from_str(&body).map_err(|err| format!("replacement recovery status is not JSON ({err}): {body}").into())
+    }
+
     async fn assert_object_body(env: &RustFSTestEnvironment, bucket: &str, key: &str, expected: &[u8]) {
         let client = env.create_s3_client();
         let response = client
@@ -442,6 +547,380 @@ mod tests {
         .into())
     }
 
+    // Keep the original unformatted-disk scenario above. This case retains the
+    // format identity so only the explicit admin task can rebuild missing data.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_cluster_root_heal_resumes_missing_remote_shards_after_node_restart() -> Result<(), Box<dyn Error + Send + Sync>>
+    {
+        init_logging();
+        info!(
+            event = "heal_restart_started",
+            component = "e2e_test",
+            subsystem = "heal",
+            "Starting root-heal restart test"
+        );
+
+        let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
+        cluster.set_env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true");
+        cluster.set_env("RUSTFS_HEAL_ENABLED", "true");
+        cluster.set_env("RUSTFS_HEAL_AUTO_HEAL_ENABLE", "false");
+        cluster.set_env("RUSTFS_HEAL_MRF_ENABLE", "false");
+        cluster.set_env("RUSTFS_SCANNER_ENABLED", "false");
+        cluster.set_env("RUSTFS_HEAL_MAX_CONCURRENT_HEALS", "1");
+        cluster.set_env("RUSTFS_HEAL_MAX_CONCURRENT_PER_SET", "1");
+        cluster.set_env("RUSTFS_HEAL_PAGE_OBJECT_CONCURRENCY", "1");
+        cluster.set_env("RUSTFS_HEAL_PAGE_PARALLEL_ENABLE", "false");
+        // Keep all storage nodes' Heal runtimes enabled so their disk services
+        // complete normal registration after restart. Scanner, auto-heal and
+        // MRF are disabled; the pre-root idle barrier below drains the direct
+        // outage-object repair before the explicit admin task starts.
+        let server_rust_log = std::env::var("RUSTFS_HEAL_CHAOS_SERVER_RUST_LOG")
+            .unwrap_or_else(|_| "rustfs::heal::task=info,rustfs=error".to_string());
+        cluster.set_env("RUST_LOG", server_rust_log);
+        if let Ok(log_dir) = std::env::var("RUSTFS_HEAL_CHAOS_LOG_DIR") {
+            std::fs::create_dir_all(&log_dir)?;
+            for node_index in 0..cluster.nodes.len() {
+                cluster.set_node_capture_log_path(node_index, format!("{log_dir}/node{node_index}.log"))?;
+            }
+        }
+        cluster.start().await?;
+        let clients = cluster.create_all_clients()?;
+
+        let bucket = "heal-restart-during-rebuild";
+        clients[0].create_bucket().bucket(bucket).send().await?;
+
+        let replaced_disk = PathBuf::from(&cluster.nodes[1].data_dir);
+        let replacement_format_path = replaced_disk.join(".rustfs.sys").join("format.json");
+        let replacement_format = std::fs::read(&replacement_format_path).map_err(|err| {
+            format!("failed to capture target format before replacement wipe at {replacement_format_path:?}: {err}")
+        })?;
+        let online_object_count = std::env::var("RUSTFS_HEAL_CHAOS_OBJECT_COUNT")
+            .ok()
+            .and_then(|value| value.parse::<usize>().ok())
+            .unwrap_or(24)
+            .clamp(8, 64);
+        let object_size_bytes = std::env::var("RUSTFS_HEAL_CHAOS_OBJECT_SIZE_BYTES")
+            .ok()
+            .and_then(|value| value.parse::<usize>().ok())
+            .unwrap_or(4 * 1024 * 1024)
+            .clamp(1024 * 1024, 16 * 1024 * 1024);
+        let mut expected_manifests = Vec::with_capacity(online_object_count);
+        for index in 0..online_object_count {
+            let key = format!("cluster/online/object-{index:04}.bin");
+            let payload_seed = u8::try_from(index + 1).expect("clamped object count must fit in u8");
+            timeout(
+                Duration::from_secs(30),
+                clients[0]
+                    .put_object()
+                    .bucket(bucket)
+                    .key(&key)
+                    .body(ByteStream::from(deterministic_object_body(object_size_bytes, payload_seed)))
+                    .send(),
+            )
+            .await??;
+            let shard_census = census_object_version_on_disk(&replaced_disk, bucket, &key, None)?;
+            assert!(
+                shard_census.is_complete(),
+                "node 1 should hold a complete baseline shard for {key}: {shard_census:?}"
+            );
+            assert!(
+                !shard_census.expected_part_numbers.is_empty(),
+                "chaos objects must use physical part shards rather than inline data: {shard_census:?}"
+            );
+            expected_manifests.push(PhysicalObjectManifest {
+                key,
+                payload_seed,
+                shard_census,
+            });
+        }
+
+        cluster.stop_node(1)?;
+        std::fs::remove_dir_all(&replaced_disk)?;
+        std::fs::create_dir_all(
+            replacement_format_path
+                .parent()
+                .ok_or("replacement format path has no parent")?,
+        )?;
+        std::fs::write(&replacement_format_path, replacement_format)?;
+        assert!(
+            replacement_format_path.is_file(),
+            "replacement target must retain only its preformatted topology identity"
+        );
+
+        let outage_key = "cluster/written-while-node-down.bin";
+        let outage_payload_seed = 0xf1;
+        timeout(
+            Duration::from_secs(30),
+            clients[2]
+                .put_object()
+                .bucket(bucket)
+                .key(outage_key)
+                .body(ByteStream::from(deterministic_object_body(object_size_bytes, outage_payload_seed)))
+                .send(),
+        )
+        .await??;
+
+        let mut outage_peer_erasure_indices = HashSet::new();
+        for (node_index, node) in cluster.nodes.iter().enumerate() {
+            if node_index == 1 {
+                continue;
+            }
+            let census = census_object_version_on_disk(Path::new(&node.data_dir), bucket, outage_key, None)?;
+            assert!(
+                census.is_complete(),
+                "online node {node_index} must hold a complete outage-object shard: {census:?}"
+            );
+            let erasure_index = census
+                .erasure_index
+                .ok_or_else(|| format!("online node {node_index} outage-object shard has no erasure index: {census:?}"))?;
+            assert!(
+                (1..=cluster.nodes.len()).contains(&erasure_index),
+                "online node {node_index} outage-object erasure index is out of range: {census:?}"
+            );
+            assert!(
+                outage_peer_erasure_indices.insert(erasure_index),
+                "outage-object erasure index {erasure_index} is duplicated across online nodes"
+            );
+        }
+        assert_eq!(
+            outage_peer_erasure_indices.len(),
+            cluster.nodes.len().saturating_sub(1),
+            "every online node must contribute one unique outage-object erasure index"
+        );
+        let expected_outage_target_erasure_index = (1..=cluster.nodes.len())
+            .find(|index| !outage_peer_erasure_indices.contains(index))
+            .ok_or("online outage-object shards leave no erasure index for the replacement target")?;
+
+        // The PUT path may have admitted a direct Internal object repair while
+        // node 1 was offline. Cancel the isolated bucket path before the target
+        // returns; otherwise it could rebuild the outage object and invalidate
+        // the explicit-root ownership assertion below.
+        let cancel_outage_heal_path = format!("/rustfs/admin/v3/heal/{bucket}?forceStop=true");
+        let (cancel_status, cancel_body) = admin_request(
+            &cluster.nodes[0].url,
+            Method::POST,
+            &cancel_outage_heal_path,
+            Some(
+                r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#
+                    .to_string(),
+            ),
+            &cluster.access_key,
+            &cluster.secret_key,
+        )
+        .await?;
+        if !cancel_status.is_success() {
+            return Err(format!("cancel outage heal failed: {cancel_status} {cancel_body}").into());
+        }
+
+        cluster.start_node(1).await?;
+
+        let status_url = format!("{}/rustfs/admin/v3/background-heal/status", cluster.nodes[0].url);
+        let recovery_deadline = Instant::now() + Duration::from_secs(60);
+        loop {
+            let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?;
+            assert!(
+                !status_body.contains("MissingContentLength"),
+                "background heal status should not fail without an explicit Content-Length: {status_body}"
+            );
+            let recovered: serde_json::Value = serde_json::from_str(&status_body)
+                .map_err(|err| format!("background heal status is not JSON ({err}): {status_body}"))?;
+            if cluster_heal_is_idle(&recovered) {
+                break;
+            }
+            if Instant::now() >= recovery_deadline {
+                return Err(format!("cluster heal operations did not become idle before root heal: {recovered}").into());
+            }
+            sleep(Duration::from_millis(250)).await;
+        }
+        assert_eq!(
+            matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?,
+            0,
+            "non-admin Heal is disabled, so the replacement target must remain empty before the explicit root heal"
+        );
+        assert!(
+            !census_object_version_on_disk(&replaced_disk, bucket, outage_key, None)?.has_xl_meta,
+            "the object written during the outage must be absent before the explicit root heal"
+        );
+        let pre_heal_replacement = replacement_recovery_status(&cluster).await?;
+        assert_eq!(
+            pre_heal_replacement["cluster"]["records"].as_array().map(Vec::len),
+            Some(0),
+            "isolated target must not retain an automatic replacement generation: {pre_heal_replacement}"
+        );
+
+        let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
+        let heal_url = format!("{}/rustfs/admin/v3/heal/?forceStart=true", cluster.nodes[0].url);
+        let heal_start_body = signed_admin_post(&heal_url, Some(heal_body), &cluster.access_key, &cluster.secret_key).await?;
+        let heal_start: serde_json::Value = serde_json::from_str(&heal_start_body)
+            .map_err(|err| format!("heal start response is not JSON ({err}): {heal_start_body}"))?;
+        let client_token = heal_start["clientToken"]
+            .as_str()
+            .filter(|token| !token.is_empty())
+            .ok_or_else(|| format!("heal start response has no client token: {heal_start}"))?;
+        let task_status_url = format!("{}/rustfs/admin/v3/heal/?clientToken={client_token}", cluster.nodes[0].url);
+
+        let partial_timeout_secs = std::env::var("RUSTFS_HEAL_CHAOS_PARTIAL_TIMEOUT_SECS")
+            .ok()
+            .and_then(|value| value.parse::<u64>().ok())
+            .unwrap_or(60);
+        let partial_deadline = Instant::now() + Duration::from_secs(partial_timeout_secs);
+        let pre_interrupt_status = loop {
+            let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?;
+            let active_status: serde_json::Value = serde_json::from_str(&status_body)
+                .map_err(|err| format!("background heal status is not JSON ({err}): {status_body}"))?;
+            if only_admin_heal_is_active(&active_status) {
+                break active_status;
+            }
+            if Instant::now() >= partial_deadline {
+                return Err(format!("root heal never became active within {partial_timeout_secs}s: {active_status}").into());
+            }
+            sleep(Duration::from_millis(50)).await;
+        };
+        let partial_count = loop {
+            let matching = matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?;
+            if matching > 0 && matching < expected_manifests.len() {
+                break matching;
+            }
+            if matching == expected_manifests.len() {
+                return Err(format!(
+                    "root heal rebuilt all {} baseline objects before the target could be interrupted",
+                    expected_manifests.len()
+                )
+                .into());
+            }
+            if Instant::now() >= partial_deadline {
+                return Err(format!(
+                    "root heal made no observable partial progress on the replacement target within {partial_timeout_secs}s"
+                )
+                .into());
+            }
+            sleep(Duration::from_millis(10)).await;
+        };
+        info!(
+            event = "heal_restart_checkpoint",
+            component = "e2e_test",
+            subsystem = "heal",
+            partial_count,
+            "Verified unique admin owner before target interruption"
+        );
+
+        cluster.stop_node(1)?;
+        let stopped_count = matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?;
+        assert!(
+            stopped_count > 0 && stopped_count < expected_manifests.len(),
+            "the target must stop after a partial rebuild, observed before stop={partial_count}, after stop={stopped_count}, total={}",
+            expected_manifests.len()
+        );
+        let unclean_shutdown_marker = replaced_disk.join(".rustfs.sys").join("unclean-shutdown");
+        match std::fs::remove_file(&unclean_shutdown_marker) {
+            Ok(()) => {}
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+            Err(error) => {
+                return Err(format!("failed to isolate unclean recovery marker {unclean_shutdown_marker:?}: {error}").into());
+            }
+        }
+        cluster.start_node(1).await?;
+
+        let heal_timeout_secs = std::env::var("RUSTFS_HEAL_REPLACED_DISK_TIMEOUT_SECS")
+            .ok()
+            .and_then(|value| value.parse::<u64>().ok())
+            .unwrap_or(180);
+        let heal_deadline = Instant::now() + Duration::from_secs(heal_timeout_secs);
+        loop {
+            if metadata_count(&replaced_disk, bucket, &expected_manifests) == expected_manifests.len()
+                && object_metadata_exists_on_disk(&replaced_disk, bucket, outage_key)
+            {
+                let matching = matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?;
+                let outage_census = census_object_version_on_disk(&replaced_disk, bucket, outage_key, None)?;
+                if matching == expected_manifests.len() && outage_census.is_complete() {
+                    break;
+                }
+            }
+            if Instant::now() >= heal_deadline {
+                let matching = matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?;
+                let outage_census = census_object_version_on_disk(&replaced_disk, bucket, outage_key, None)?;
+                let final_status = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key)
+                    .await
+                    .unwrap_or_else(|err| format!("status request failed: {err}"));
+                let task_status = match timeout(
+                    Duration::from_secs(5),
+                    signed_admin_post(&task_status_url, None, &cluster.access_key, &cluster.secret_key),
+                )
+                .await
+                {
+                    Ok(Ok(body)) => heal_task_status_diagnostic(&body),
+                    Ok(Err(err)) => format!("task status request failed: {err}"),
+                    Err(_) => "task status request exceeded 5s diagnostic budget".to_string(),
+                };
+                let replacement_status = match timeout(Duration::from_secs(5), replacement_recovery_status(&cluster)).await {
+                    Ok(Ok(status)) => status.to_string(),
+                    Ok(Err(err)) => format!("replacement status request failed: {err}"),
+                    Err(_) => "replacement status request exceeded 5s diagnostic budget".to_string(),
+                };
+                return Err(format!(
+                    "root heal did not resume after target restart within {heal_timeout_secs}s: baseline={matching}/{}, outage={outage_census:?}, status={final_status}, task_status={task_status}, pre_interrupt_status={pre_interrupt_status}, pre_heal_replacement={pre_heal_replacement}, replacement_status={replacement_status}",
+                    expected_manifests.len()
+                )
+                .into());
+            }
+            sleep(Duration::from_millis(250)).await;
+        }
+
+        for expected in &expected_manifests {
+            let actual = census_object_version_on_disk(&replaced_disk, bucket, &expected.key, None)?;
+            assert!(
+                actual.matches_manifest(&expected.shard_census),
+                "rebuilt target shard differs from its baseline for {}: {actual:?}",
+                expected.key
+            );
+        }
+        let outage_census = census_object_version_on_disk(&replaced_disk, bucket, outage_key, None)?;
+        assert!(
+            outage_census.is_complete(),
+            "outage object must have a complete target shard: {outage_census:?}"
+        );
+        assert_eq!(
+            outage_census.erasure_index,
+            Some(expected_outage_target_erasure_index),
+            "the outage object must be rebuilt into its own missing erasure slot"
+        );
+
+        let target_client = cluster.create_s3_client(1)?;
+        for expected in &expected_manifests {
+            let response = target_client.get_object().bucket(bucket).key(&expected.key).send().await?;
+            let actual = response.body.collect().await?.into_bytes();
+            let expected_body = deterministic_object_body(object_size_bytes, expected.payload_seed);
+            assert_eq!(actual.as_ref(), expected_body.as_slice(), "object body changed for {}", expected.key);
+        }
+        let response = target_client.get_object().bucket(bucket).key(outage_key).send().await?;
+        let actual = response.body.collect().await?.into_bytes();
+        let expected_outage_body = deterministic_object_body(object_size_bytes, outage_payload_seed);
+        assert_eq!(actual.as_ref(), expected_outage_body.as_slice(), "object body changed for {outage_key}");
+
+        let terminal_deadline = Instant::now() + Duration::from_secs(30);
+        loop {
+            let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?;
+            let status: serde_json::Value = serde_json::from_str(&status_body)
+                .map_err(|err| format!("background heal status is not JSON ({err}): {status_body}"))?;
+            if cluster_heal_is_idle(&status) {
+                break;
+            }
+            if Instant::now() >= terminal_deadline {
+                return Err(format!("heal data rebuilt but operations did not converge to terminal idle: {status}").into());
+            }
+            sleep(Duration::from_millis(250)).await;
+        }
+
+        let task_status_body = signed_admin_post(&task_status_url, None, &cluster.access_key, &cluster.secret_key).await?;
+        let task_status: serde_json::Value = serde_json::from_str(&task_status_body)
+            .map_err(|err| format!("heal task status is not JSON ({err}): {task_status_body}"))?;
+        if task_status["summary"].as_str() != Some("finished") {
+            return Err(format!("heal data rebuilt but task did not finish successfully: {task_status}").into());
+        }
+
+        Ok(())
+    }
+
     /// Issue #5850: `background-heal/status` must answer while a peer is down.
     ///
     /// Exercises the production path in `read_cluster_heal_status` end to end,

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

@@ -36,6 +36,7 @@ use rustfs_rio::{ChunkReaderBox, HttpChunkReader, HttpReader, HttpWriter};
 use sha2::{Digest, Sha256};
 use std::collections::HashMap;
 use std::future::Future;
+use std::io;
 use std::pin::Pin;
 use std::sync::{Arc, LazyLock, OnceLock};
 use std::task::{Context, Poll};
@@ -105,9 +106,13 @@ struct PutFileCapabilityCacheState {
     cached: Option<PutFileCapabilityState>,
     generation: u64,
     in_flight: Option<PutFileCapabilityFlight>,
+    rejected_server_epoch: Option<Uuid>,
 }
 
-type PutFileCapabilityCacheEntry = Arc<tokio::sync::RwLock<PutFileCapabilityCacheState>>;
+// The registry lock is released before taking an entry lock. Entry guards cover
+// only cache transitions, never a probe or await; poll-based writers must be
+// able to reject an epoch atomically with those transitions.
+type PutFileCapabilityCacheEntry = Arc<parking_lot::RwLock<PutFileCapabilityCacheState>>;
 
 static PUT_FILE_CAPABILITY_CACHE: LazyLock<parking_lot::RwLock<HashMap<String, PutFileCapabilityCacheEntry>>> =
     LazyLock::new(|| parking_lot::RwLock::new(HashMap::new()));
@@ -119,7 +124,7 @@ fn put_file_capability_cache_entry(endpoint: &str) -> PutFileCapabilityCacheEntr
     PUT_FILE_CAPABILITY_CACHE
         .write()
         .entry(endpoint.to_owned())
-        .or_insert_with(|| Arc::new(tokio::sync::RwLock::new(PutFileCapabilityCacheState::default())))
+        .or_insert_with(|| Arc::new(parking_lot::RwLock::new(PutFileCapabilityCacheState::default())))
         .clone()
 }
 
@@ -134,6 +139,23 @@ fn fresh_put_file_capability(state: Option<PutFileCapabilityState>, now: Instant
     }
 }
 
+fn reject_put_file_server_epoch(endpoint: &str, server_epoch: Uuid) {
+    let entry = PUT_FILE_CAPABILITY_CACHE.read().get(endpoint).cloned();
+    if let Some(entry) = entry {
+        let mut state = entry.write();
+        if matches!(state.cached, Some(PutFileCapabilityState::V1 { server_epoch: cached, .. }) if cached == server_epoch) {
+            state.rejected_server_epoch = Some(server_epoch);
+        }
+    }
+}
+
+fn usable_put_file_capability(state: &PutFileCapabilityCacheState, now: Instant) -> Option<Option<Uuid>> {
+    match fresh_put_file_capability(state.cached, now)? {
+        Some(server_epoch) if state.rejected_server_epoch == Some(server_epoch) => None,
+        capability => Some(capability),
+    }
+}
+
 fn put_file_capability_status_is_legacy(status: u16) -> bool {
     status == 404
 }
@@ -322,13 +344,14 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
 
     async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter> {
         let server_epoch = self.put_file_auth_capability(&request.endpoint).await?;
-        let nonce = server_epoch.map(|_| Uuid::new_v4());
+        let auth_scope = server_epoch.map(|server_epoch| (Uuid::new_v4(), server_epoch));
-        let url = build_put_file_stream_url(&request, nonce.zip(server_epoch));
+        let url = build_put_file_stream_url(&request, auth_scope);
+        let endpoint = request.endpoint;
         let mut headers = json_headers();
         build_auth_headers(&url, &Method::PUT, &mut headers)?;
         let writer = HttpWriter::new(url.clone(), Method::PUT, headers).await?;
-        match nonce {
+        match auth_scope {
-            Some(nonce) => Ok(Box::new(PutFileAuthWriter::new(writer, url, nonce))),
+            Some((nonce, server_epoch)) => Ok(Box::new(PutFileAuthWriter::new(writer, url, nonce, endpoint, server_epoch))),
             None => Ok(Box::new(writer)),
         }
     }
@@ -498,15 +521,15 @@ where
 {
     let entry = put_file_capability_cache_entry(endpoint);
     {
-        let state = entry.read().await;
+        let state = entry.read();
-        if let Some(cached) = fresh_put_file_capability(state.cached, Instant::now()) {
+        if let Some(cached) = usable_put_file_capability(&state, Instant::now()) {
             return Ok(cached);
         }
     }
 
     let flight = {
-        let mut state = entry.write().await;
+        let mut state = entry.write();
-        if let Some(cached) = fresh_put_file_capability(state.cached, Instant::now()) {
+        if let Some(cached) = usable_put_file_capability(&state, Instant::now()) {
             return Ok(cached);
         }
         if let Some(flight) = state.in_flight.clone() {
@@ -532,7 +555,7 @@ where
         .await;
 
     {
-        let mut state = entry.write().await;
+        let mut state = entry.write();
         let is_current_flight = state
             .in_flight
             .as_ref()
@@ -540,6 +563,9 @@ where
         if is_current_flight {
             match outcome {
                 Ok(Some(server_epoch)) => {
+                    if state.rejected_server_epoch != Some(*server_epoch) {
+                        state.rejected_server_epoch = None;
+                    }
                     state.cached = Some(PutFileCapabilityState::V1 {
                         server_epoch: *server_epoch,
                         revalidate_after: Instant::now() + PUT_FILE_V1_CAPABILITY_TTL,
@@ -630,17 +656,23 @@ struct PutFileAuthWriter<W> {
     inner: W,
     url: String,
     nonce: Uuid,
+    endpoint: String,
+    server_epoch: Uuid,
+    server_epoch_rejected: bool,
     hasher: Sha256,
     trailer: Option<Vec<u8>>,
     trailer_offset: usize,
 }
 
 impl<W> PutFileAuthWriter<W> {
-    fn new(inner: W, url: String, nonce: Uuid) -> Self {
+    fn new(inner: W, url: String, nonce: Uuid, endpoint: String, server_epoch: Uuid) -> Self {
         Self {
             inner,
             url,
             nonce,
+            endpoint,
+            server_epoch,
+            server_epoch_rejected: false,
             hasher: Sha256::new(),
             trailer: None,
             trailer_offset: 0,
@@ -656,6 +688,14 @@ impl<W> PutFileAuthWriter<W> {
         Ok(())
     }
 
+    fn reject_server_epoch_on_conflict(&mut self, error: &io::Error) {
+        if self.server_epoch_rejected || !io_error_has_put_file_epoch_conflict(error) {
+            return;
+        }
+        reject_put_file_server_epoch(&self.endpoint, self.server_epoch);
+        self.server_epoch_rejected = true;
+    }
+
     fn poll_write_trailer(&mut self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>>
     where
         W: AsyncWrite + Unpin,
@@ -673,7 +713,10 @@ impl<W> PutFileAuthWriter<W> {
                     )));
                 }
                 Poll::Ready(Ok(written)) => written,
-                Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+                Poll::Ready(Err(err)) => {
+                    self.reject_server_epoch_on_conflict(&err);
+                    return Poll::Ready(Err(err));
+                }
                 Poll::Pending => return Poll::Pending,
             };
             self.trailer_offset += written;
@@ -682,6 +725,15 @@ impl<W> PutFileAuthWriter<W> {
     }
 }
 
+fn io_error_has_put_file_epoch_conflict(error: &io::Error) -> bool {
+    error
+        .get_ref()
+        .and_then(|source| source.downcast_ref::<rustfs_rio::InternodeHttpError>())
+        .is_some_and(
+            |error| matches!(error.kind(), rustfs_rio::InternodeHttpErrorKind::HttpStatus(status) if status.as_u16() == 409),
+        )
+}
+
 impl<W> AsyncWrite for PutFileAuthWriter<W>
 where
     W: AsyncWrite + Unpin,
@@ -698,12 +750,22 @@ where
                 self.hasher.update(&buf[..written]);
                 Poll::Ready(Ok(written))
             }
-            other => other,
+            Poll::Ready(Err(err)) => {
+                self.reject_server_epoch_on_conflict(&err);
+                Poll::Ready(Err(err))
+            }
+            Poll::Pending => Poll::Pending,
         }
     }
 
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        Pin::new(&mut self.inner).poll_flush(cx)
+        match Pin::new(&mut self.inner).poll_flush(cx) {
+            Poll::Ready(Err(err)) => {
+                self.reject_server_epoch_on_conflict(&err);
+                Poll::Ready(Err(err))
+            }
+            other => other,
+        }
     }
 
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
@@ -712,7 +774,13 @@ where
             Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
             Poll::Pending => return Poll::Pending,
         }
-        Pin::new(&mut self.inner).poll_shutdown(cx)
+        match Pin::new(&mut self.inner).poll_shutdown(cx) {
+            Poll::Ready(Err(err)) => {
+                self.reject_server_epoch_on_conflict(&err);
+                Poll::Ready(Err(err))
+            }
+            other => other,
+        }
     }
 }
 
@@ -840,7 +908,6 @@ mod tests {
             loop {
                 let strong_count = entry
                     .read()
-                    .await
                     .in_flight
                     .as_ref()
                     .map(|flight| Arc::strong_count(&flight.outcome))
@@ -858,6 +925,50 @@ mod tests {
     #[derive(Debug)]
     struct LegacyTestTransport;
 
+    #[derive(Clone, Copy, Debug)]
+    enum PutFileFailurePhase {
+        Write,
+        Flush,
+        Shutdown,
+    }
+
+    struct PutFileFailureWriter {
+        phase: PutFileFailurePhase,
+        status: reqwest::StatusCode,
+    }
+
+    impl PutFileFailureWriter {
+        fn error(&self) -> io::Error {
+            rustfs_rio::new_test_internode_http_io_error(rustfs_rio::InternodeHttpErrorKind::HttpStatus(self.status))
+        }
+    }
+
+    impl tokio::io::AsyncWrite for PutFileFailureWriter {
+        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
+            Poll::Ready(if matches!(self.phase, PutFileFailurePhase::Write) {
+                Err(self.error())
+            } else {
+                Ok(buf.len())
+            })
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(if matches!(self.phase, PutFileFailurePhase::Flush) {
+                Err(self.error())
+            } else {
+                Ok(())
+            })
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(if matches!(self.phase, PutFileFailurePhase::Shutdown) {
+                Err(self.error())
+            } else {
+                Ok(())
+            })
+        }
+    }
+
     #[async_trait::async_trait]
     impl InternodeDataTransport for LegacyTestTransport {
         async fn open_read(&self, _request: ReadStreamRequest) -> Result<FileReader> {
@@ -1048,7 +1159,7 @@ mod tests {
         let v1_endpoint = format!("http://v1-{}.invalid", Uuid::new_v4());
         let v1_entry = put_file_capability_cache_entry(&v1_endpoint);
         let server_epoch = Uuid::new_v4();
-        v1_entry.write().await.cached = Some(PutFileCapabilityState::V1 {
+        v1_entry.write().cached = Some(PutFileCapabilityState::V1 {
             server_epoch,
             revalidate_after: Instant::now() + PUT_FILE_V1_CAPABILITY_TTL,
         });
@@ -1067,7 +1178,7 @@ mod tests {
             Some(server_epoch)
         );
         assert!(!cache_probe_called.load(Ordering::SeqCst));
-        v1_entry.write().await.cached = Some(PutFileCapabilityState::V1 {
+        v1_entry.write().cached = Some(PutFileCapabilityState::V1 {
             server_epoch,
             revalidate_after: Instant::now(),
         });
@@ -1086,8 +1197,7 @@ mod tests {
 
         let legacy_endpoint = format!("http://legacy-{}.invalid", Uuid::new_v4());
         let legacy_entry = put_file_capability_cache_entry(&legacy_endpoint);
-        legacy_entry.write().await.cached =
+        legacy_entry.write().cached = Some(PutFileCapabilityState::LegacyUntil(Instant::now() + PUT_FILE_LEGACY_CAPABILITY_TTL));
-            Some(PutFileCapabilityState::LegacyUntil(Instant::now() + PUT_FILE_LEGACY_CAPABILITY_TTL));
         assert!(
             transport
                 .put_file_auth_capability(&legacy_endpoint)
@@ -1098,7 +1208,7 @@ mod tests {
 
         let expired_endpoint = format!("http://expired-legacy-{}.invalid", Uuid::new_v4());
         let expired_entry = put_file_capability_cache_entry(&expired_endpoint);
-        expired_entry.write().await.cached = Some(PutFileCapabilityState::LegacyUntil(Instant::now()));
+        expired_entry.write().cached = Some(PutFileCapabilityState::LegacyUntil(Instant::now()));
         let reprobed = std::sync::atomic::AtomicBool::new(false);
         assert_eq!(
             resolve_put_file_auth_capability(&expired_endpoint, || async {
@@ -1349,7 +1459,7 @@ mod tests {
         };
         probe_started.notified().await;
         {
-            let mut state = entry.write().await;
+            let mut state = entry.write();
             state.generation = state.generation.checked_add(1).expect("test generation should advance");
             state.cached = Some(PutFileCapabilityState::V1 {
                 server_epoch: newer_epoch,
@@ -1362,10 +1472,7 @@ mod tests {
             task.await.expect("stale task should finish").expect("stale probe result"),
             Some(stale_epoch)
         );
-        assert_eq!(
+        assert_eq!(fresh_put_file_capability(entry.read().cached, Instant::now()), Some(Some(newer_epoch)));
-            fresh_put_file_capability(entry.read().await.cached, Instant::now()),
-            Some(Some(newer_epoch))
-        );
     }
 
     #[test]
@@ -1398,6 +1505,8 @@ mod tests {
 
         let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-writer-test-secret".to_string());
         let nonce = Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("nonce");
+        let server_epoch = Uuid::parse_str("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee").expect("server epoch");
+        let endpoint = "http://node1:9000".to_string();
         let url = concat!(
             "http://node1:9000/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1",
             "&append=false&size=11&put_file_auth=digest-trailer-v1&put_file_nonce=11111111-2222-4333-8444-555555555555"
@@ -1406,7 +1515,7 @@ mod tests {
         let mut sink = Vec::new();
 
         {
-            let mut writer = PutFileAuthWriter::new(&mut sink, url.clone(), nonce);
+            let mut writer = PutFileAuthWriter::new(&mut sink, url.clone(), nonce, endpoint, server_epoch);
             writer.write_all(b"hello world").await.expect("body write should succeed");
             writer.shutdown().await.expect("shutdown should append auth trailer");
             let err = writer
@@ -1424,6 +1533,143 @@ mod tests {
         assert_eq!(verified, expected_digest);
     }
 
+    #[tokio::test]
+    async fn put_file_auth_writer_reprobes_after_server_epoch_conflict() {
+        use tokio::io::AsyncWriteExt;
+
+        let _ = rustfs_credentials::set_global_rpc_secret("put-file-epoch-conflict-test-secret".to_string());
+        for status in [reqwest::StatusCode::CONFLICT, reqwest::StatusCode::BAD_REQUEST] {
+            for (phase, trailer_write) in [
+                (PutFileFailurePhase::Write, false),
+                (PutFileFailurePhase::Write, true),
+                (PutFileFailurePhase::Flush, false),
+                (PutFileFailurePhase::Shutdown, false),
+            ] {
+                let endpoint = format!("http://epoch-conflict-{}.invalid", Uuid::new_v4());
+                let stale_epoch = Uuid::new_v4();
+                let replacement_epoch = Uuid::new_v4();
+                resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(stale_epoch)) })
+                    .await
+                    .expect("initial capability should resolve");
+                let mut writer = PutFileAuthWriter::new(
+                    PutFileFailureWriter { phase, status },
+                    format!("{endpoint}{PUT_FILE_AUTH_STREAM_PATH}"),
+                    Uuid::new_v4(),
+                    endpoint.clone(),
+                    stale_epoch,
+                );
+                let error = match (phase, trailer_write) {
+                    (PutFileFailurePhase::Write, false) => writer.write_all(b"body").await,
+                    (PutFileFailurePhase::Flush, _) => writer.flush().await,
+                    _ => writer.shutdown().await,
+                }
+                .expect_err("injected writer error must reach the caller");
+                let conflict = status == reqwest::StatusCode::CONFLICT;
+                assert_eq!(io_error_has_put_file_epoch_conflict(&error), conflict);
+
+                let probe_called = AtomicBool::new(false);
+                let resolved = resolve_put_file_auth_capability(&endpoint, || async {
+                    probe_called.store(true, Ordering::SeqCst);
+                    Ok(Some(replacement_epoch))
+                })
+                .await
+                .expect("capability should remain usable or be reprobed");
+                assert_eq!(probe_called.load(Ordering::SeqCst), conflict, "phase={phase:?}, trailer={trailer_write}");
+                assert_eq!(resolved, Some(if conflict { replacement_epoch } else { stale_epoch }));
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn late_put_file_epoch_rejection_preserves_current_rejection() {
+        let endpoint = format!("http://late-epoch-conflict-{}.invalid", Uuid::new_v4());
+        let old_epoch = Uuid::new_v4();
+        let current_epoch = Uuid::new_v4();
+        let replacement_epoch = Uuid::new_v4();
+        assert_eq!(
+            resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(old_epoch)) })
+                .await
+                .expect("initial epoch should be cached"),
+            Some(old_epoch)
+        );
+        reject_put_file_server_epoch(&endpoint, old_epoch);
+        assert_eq!(
+            resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(current_epoch)) })
+                .await
+                .expect("first restart should install a new epoch"),
+            Some(current_epoch)
+        );
+
+        reject_put_file_server_epoch(&endpoint, current_epoch);
+        // A writer opened before the first restart can report its 409 after
+        // a newer writer has already rejected the second server incarnation.
+        reject_put_file_server_epoch(&endpoint, old_epoch);
+        let probe_called = AtomicBool::new(false);
+        let resolved = resolve_put_file_auth_capability(&endpoint, || async {
+            probe_called.store(true, Ordering::SeqCst);
+            Ok(Some(replacement_epoch))
+        })
+        .await
+        .expect("late old-epoch rejection must preserve the current rejection");
+
+        assert!(probe_called.load(Ordering::SeqCst), "known-rejected current epoch must be reprobed");
+        assert_eq!(resolved, Some(replacement_epoch));
+    }
+
+    #[tokio::test]
+    async fn put_file_epoch_rejection_is_endpoint_and_epoch_scoped() {
+        let endpoint = format!("http://scoped-epoch-{}.invalid", Uuid::new_v4());
+        let other_endpoint = format!("http://other-epoch-{}.invalid", Uuid::new_v4());
+        let current_epoch = Uuid::new_v4();
+        for endpoint in [&endpoint, &other_endpoint] {
+            resolve_put_file_auth_capability(endpoint, || async { Ok(Some(current_epoch)) })
+                .await
+                .expect("initial epoch should resolve");
+        }
+        reject_put_file_server_epoch(&endpoint, Uuid::new_v4());
+        assert_eq!(
+            resolve_put_file_auth_capability(&endpoint, || async { panic!("old writer must not invalidate a new epoch") })
+                .await
+                .expect("new epoch must remain cached"),
+            Some(current_epoch)
+        );
+        reject_put_file_server_epoch(&endpoint, current_epoch);
+        assert_eq!(
+            resolve_put_file_auth_capability(&other_endpoint, || async { panic!("another endpoint must stay cached") })
+                .await
+                .expect("other endpoint must remain cached"),
+            Some(current_epoch)
+        );
+    }
+
+    #[tokio::test]
+    async fn put_file_rejected_epoch_survives_failed_stale_and_downgrade_probes() {
+        let endpoint = format!("http://rejected-probe-{}.invalid", Uuid::new_v4());
+        let rejected_epoch = Uuid::new_v4();
+        let replacement_epoch = Uuid::new_v4();
+        resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(rejected_epoch)) })
+            .await
+            .expect("initial epoch should resolve");
+        reject_put_file_server_epoch(&endpoint, rejected_epoch);
+        let failure = resolve_put_file_auth_capability(&endpoint, || async { Err(Error::other("injected probe failure")) })
+            .await
+            .expect_err("probe failure must be returned");
+        assert!(failure.to_string().contains("injected probe failure"));
+        let downgrade = resolve_put_file_auth_capability(&endpoint, || async { Ok(None) })
+            .await
+            .expect_err("rejection must not unpin authenticated v1");
+        assert!(downgrade.to_string().contains("downgrade rejected"));
+        resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(rejected_epoch)) })
+            .await
+            .expect("a probe racing a restart can still return the old epoch");
+        assert_eq!(
+            resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(replacement_epoch)) })
+                .await
+                .expect("same-epoch probe must not clear known rejection"),
+            Some(replacement_epoch)
+        );
+    }
+
     #[test]
     fn walk_dir_url_encodes_disk_ref() {
         let url = build_walk_dir_url(&WalkDirStreamRequest {

crates/ecstore/src/disk/error.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM;
 use rustfs_rio::{InternodeHttpError, InternodeHttpErrorKind};
 use std::error::Error as StdError;
 use std::hash::{Hash, Hasher};
@@ -229,6 +230,19 @@ fn classify_internode_missing_error(error: &InternodeHttpError) -> Option<DiskEr
     None
 }
 
+fn internode_write_error_is_retryable(error: &InternodeHttpError) -> bool {
+    error.kind().is_retryable()
+        || (matches!(error.kind(), InternodeHttpErrorKind::HttpStatus(status) if status.as_u16() == 409)
+            && error.context().operation() == Some(INTERNODE_OPERATION_PUT_FILE_STREAM))
+}
+
+fn io_error_contains_retryable_internode_write(error: &io::Error) -> bool {
+    error
+        .get_ref()
+        .and_then(|source| source.downcast_ref::<InternodeHttpError>())
+        .is_some_and(internode_write_error_is_retryable)
+}
+
 /// Wrap a terminal shard-read failure without changing its typed
 /// classification.  Timeout-like disk errors retain `TimedOut`; other errors
 /// retain their inner I/O kind or use `Other` when no more specific kind exists.
@@ -336,10 +350,7 @@ impl DiskError {
 
     pub fn is_retryable_internode_write_failure(&self) -> bool {
         match self {
-            DiskError::Io(io_error) => io_error
+            DiskError::Io(io_error) => io_error_contains_retryable_internode_write(io_error),
-                .get_ref()
-                .and_then(|source| source.downcast_ref::<InternodeHttpError>())
-                .is_some_and(|err| err.kind().is_retryable()),
             _ => false,
         }
     }
@@ -1240,6 +1251,68 @@ mod tests {
         assert!(!DiskError::FileNotFound.is_internode_http_status(429));
     }
 
+    #[test]
+    fn test_put_file_server_epoch_conflict_is_retryable_write_failure() {
+        let conflict = DiskError::from(rustfs_rio::new_test_internode_http_io_error(
+            rustfs_rio::InternodeHttpErrorKind::HttpStatus(http::StatusCode::CONFLICT),
+        ));
+        let bad_request = DiskError::from(rustfs_rio::new_test_internode_http_io_error(
+            rustfs_rio::InternodeHttpErrorKind::HttpStatus(http::StatusCode::BAD_REQUEST),
+        ));
+
+        assert!(conflict.is_retryable_internode_write_failure());
+        assert!(!bad_request.is_retryable_internode_write_failure());
+    }
+
+    #[tokio::test]
+    async fn read_stream_conflict_is_not_a_retryable_put_file_failure() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+                .await
+                .expect("bind isolated HTTP fixture");
+            let address = listener.local_addr().expect("fixture address");
+            let server = tokio::spawn(async move {
+                let (mut stream, _) = listener.accept().await.expect("accept read request");
+                let mut request = [0_u8; 4096];
+                let mut read = 0;
+                loop {
+                    let count = stream.read(&mut request[read..]).await.expect("read HTTP request");
+                    assert!(count > 0, "request ended before its complete headers");
+                    read += count;
+                    if request[..read].windows(4).any(|bytes| bytes == b"\r\n\r\n") {
+                        break;
+                    }
+                    assert!(read < request.len(), "fixture request headers exceed their budget");
+                }
+                stream
+                    .write_all(b"HTTP/1.1 409 Conflict\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                    .await
+                    .expect("send typed conflict response");
+            });
+            let error = match rustfs_rio::HttpReader::new(
+                format!("http://{address}/rustfs/rpc/read_file_stream"),
+                http::Method::GET,
+                http::HeaderMap::new(),
+                None,
+            )
+            .await
+            {
+                Ok(_) => panic!("HTTP 409 must fail the read"),
+                Err(error) => DiskError::from(error),
+            };
+            server.await.expect("fixture task should complete");
+            assert!(error.is_internode_http_status(409));
+            assert!(
+                !error.is_retryable_internode_write_failure(),
+                "read-operation 409 must not trigger put-file retry"
+            );
+        })
+        .await
+        .expect("isolated read-conflict test must finish within its budget");
+    }
+
     #[test]
     fn test_internode_missing_errors_preserve_disk_error_types() {
         let file_missing = DiskError::from(rustfs_rio::new_test_remote_file_not_found_http_io_error());

crates/ecstore/src/erasure/coding/encode.rs

@@ -321,6 +321,13 @@ impl<'a> MultiWriter<'a> {
         }
     }
 
+    pub(super) fn take_retryable_internode_write_failure(&mut self) -> Option<Error> {
+        self.errs
+            .iter_mut()
+            .find(|error| error.as_ref().is_some_and(Error::is_retryable_internode_write_failure))
+            .and_then(Option::take)
+    }
+
     /// Effective budget for one shard operation: the smaller of the per-shard
     /// stall timeout and the time remaining until the object's absolute cap.
     /// Returns `None` when neither deadline is configured (wait indefinitely).

crates/ecstore/src/erasure/coding/heal.rs

@@ -108,6 +108,13 @@ where
     (shards, errs)
 }
 
+fn heal_writer_failure(writers: &mut MultiWriter<'_>, error: io::Error) -> Error {
+    writers
+        .take_retryable_internode_write_failure()
+        .map(|error| Error::RemoteClientUnavailable(error.to_string()))
+        .unwrap_or_else(|| error.into())
+}
+
 impl super::Erasure {
     pub async fn heal<R>(
         &self,
@@ -202,10 +209,14 @@ impl super::Erasure {
                 .map(|s| Bytes::from(s.unwrap_or_default()))
                 .collect::<Vec<_>>();
 
-            writers.write(shards).await?;
+            if let Err(error) = writers.write(shards).await {
+                return Err(heal_writer_failure(&mut writers, error));
+            }
         }
 
-        writers.shutdown().await?;
+        if let Err(error) = writers.shutdown().await {
+            return Err(heal_writer_failure(&mut writers, error));
+        }
         Ok(())
     }
 }
@@ -246,6 +257,35 @@ mod tests {
         }
     }
 
+    struct InternodeFailureWriter {
+        fail_on_write: bool,
+        status: http::StatusCode,
+    }
+
+    impl InternodeFailureWriter {
+        fn error(&self) -> io::Error {
+            rustfs_rio::new_test_internode_http_io_error(rustfs_rio::InternodeHttpErrorKind::HttpStatus(self.status))
+        }
+    }
+
+    impl AsyncWrite for InternodeFailureWriter {
+        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
+            Poll::Ready(if self.fail_on_write {
+                Err(self.error())
+            } else {
+                Ok(buf.len())
+            })
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Err(self.error()))
+        }
+    }
+
     struct PendingReader;
 
     impl AsyncRead for PendingReader {
@@ -331,6 +371,94 @@ mod tests {
         assert!(writers.iter().all(Option::is_some));
     }
 
+    #[tokio::test]
+    async fn heal_maps_put_file_epoch_conflict_to_retryable_remote_unavailable() {
+        for status in [http::StatusCode::CONFLICT, http::StatusCode::BAD_REQUEST] {
+            for (fail_on_write, data) in [
+                (false, b"".as_slice()),
+                (false, b"payload".as_slice()),
+                (true, b"payload".as_slice()),
+            ] {
+                let erasure = Erasure::new(2, 1, 64);
+                let encoded = erasure.encode_data(data).expect("source shards should encode");
+                let readers = encoded
+                    .iter()
+                    .enumerate()
+                    .map(|(index, shard)| {
+                        (index < erasure.data_shards).then(|| {
+                            BitrotReader::new(Cursor::new(shard.to_vec()), erasure.shard_size(), HashAlgorithm::None, false)
+                        })
+                    })
+                    .collect::<Vec<_>>();
+                let mut writers = (0..erasure.total_shard_count())
+                    .map(|index| {
+                        (index == erasure.data_shards).then(|| {
+                            BitrotWriterWrapper::new(
+                                CustomWriter::new_tokio_writer(InternodeFailureWriter { fail_on_write, status }),
+                                erasure.shard_size(),
+                                HashAlgorithm::None,
+                            )
+                        })
+                    })
+                    .collect::<Vec<_>>();
+                let error = erasure
+                    .heal(&mut writers, readers, data.len(), &[])
+                    .await
+                    .expect_err("failed sole target must not satisfy heal write quorum");
+                assert_eq!(
+                    matches!(error, Error::RemoteClientUnavailable(_)),
+                    status == http::StatusCode::CONFLICT,
+                    "status={status}, fail_on_write={fail_on_write}, len={}, error={error:?}",
+                    data.len()
+                );
+                assert!(writers.iter().all(Option::is_none), "failed target must not be committed");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn heal_epoch_conflict_does_not_abort_healthy_target() {
+        for fail_on_write in [false, true] {
+            let erasure = Erasure::new(2, 2, 64);
+            let data = b"healthy target must retain exact reconstructed bytes";
+            let encoded = erasure.encode_data(data).expect("source shards should encode");
+            let readers = encoded
+                .iter()
+                .enumerate()
+                .map(|(index, shard)| {
+                    (index < erasure.data_shards)
+                        .then(|| BitrotReader::new(Cursor::new(shard.to_vec()), erasure.shard_size(), HashAlgorithm::None, false))
+                })
+                .collect::<Vec<_>>();
+            let mut writers = vec![
+                None,
+                None,
+                Some(BitrotWriterWrapper::new(
+                    CustomWriter::new_tokio_writer(InternodeFailureWriter {
+                        fail_on_write,
+                        status: http::StatusCode::CONFLICT,
+                    }),
+                    erasure.shard_size(),
+                    HashAlgorithm::None,
+                )),
+                Some(inline_writer(erasure.shard_size())),
+            ];
+            erasure
+                .heal(&mut writers, readers, data.len(), &[])
+                .await
+                .expect("one healthy target must still satisfy the existing heal quorum");
+            assert!(writers[2].is_none(), "conflicting target must be dropped");
+            assert_eq!(
+                writers[3]
+                    .take()
+                    .expect("healthy target remains")
+                    .into_inline_data()
+                    .expect("inline target data"),
+                encoded[3].to_vec()
+            );
+        }
+    }
+
     #[tokio::test]
     async fn heal_reconstructs_missing_parity_shard() {
         let erasure = Erasure::new(2, 2, 64);

crates/ecstore/src/set_disk/ops/heal.rs

@@ -36,6 +36,20 @@ const EVENT_HEAL_OBJECT_RENAME: &str = "heal_object_rename";
 const HEAL_RENAME_INCOMPLETE: &str = "heal rename incomplete";
 const READ_REPAIR_DATA_PHASE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60 * 60);
 
+fn heal_drive_state_for_error(error: &DiskError) -> DriveState {
+    match error {
+        DiskError::DiskNotFound | DiskError::RemoteClientUnavailable(_) => DriveState::Offline,
+        DiskError::FaultyDisk | DiskError::FaultyRemoteDisk => DriveState::Faulty,
+        DiskError::FileNotFound
+        | DiskError::FileVersionNotFound
+        | DiskError::VolumeNotFound
+        | DiskError::PartMissingOrCorrupt
+        | DiskError::OutdatedXLMeta => DriveState::Missing,
+        DiskError::FileCorrupt => DriveState::Corrupt,
+        _ => DriveState::Unknown(error.to_string()),
+    }
+}
+
 #[cfg(test)]
 static HEAL_RENAME_FAILURES: std::sync::Mutex<Vec<(String, String, usize)>> = std::sync::Mutex::new(Vec::new());
 
@@ -892,16 +906,7 @@ impl SetDisks {
                             }
 
                             let drive_state = match reason {
-                                Some(err) => match err {
+                                Some(err) => heal_drive_state_for_error(&err).to_string(),
-                                    DiskError::DiskNotFound => DriveState::Offline.to_string(),
-                                    DiskError::FileNotFound
-                                    | DiskError::FileVersionNotFound
-                                    | DiskError::VolumeNotFound
-                                    | DiskError::PartMissingOrCorrupt
-                                    | DiskError::OutdatedXLMeta => DriveState::Missing.to_string(),
-                                    DiskError::FileCorrupt => DriveState::Corrupt.to_string(),
-                                    _ => DriveState::Unknown(err.to_string()).to_string(),
-                                },
                                 None => DriveState::Ok.to_string(),
                             };
                             result.before.drives.push(HealDriveInfo {
@@ -2673,6 +2678,17 @@ mod heal_result_report_tests {
         assert!(!super::metadata_less_part_file("xl.meta"));
     }
 
+    #[test]
+    fn unavailable_heal_errors_use_stable_drive_states() {
+        for error in [DiskError::FaultyDisk, DiskError::FaultyRemoteDisk] {
+            assert_eq!(super::heal_drive_state_for_error(&error).to_string(), DriveState::Faulty.to_string());
+        }
+        assert_eq!(
+            super::heal_drive_state_for_error(&DiskError::RemoteClientUnavailable("peer restarting".to_string())).to_string(),
+            DriveState::Offline.to_string()
+        );
+    }
+
     #[test]
     fn read_repair_commit_fingerprint_tracks_commit_identity_only() {
         let data_dir = Uuid::parse_str("11111111-1111-1111-1111-111111111111").expect("data dir should parse");

crates/heal/src/heal/task.rs

@@ -24,7 +24,7 @@ use crate::heal::{
 use crate::{Error, Result};
 use metrics::{counter, histogram};
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit};
-use rustfs_heal_contracts::heal_channel::{HealOpts, HealRequestSource, HealScanMode};
+use rustfs_heal_contracts::heal_channel::{DriveState, HealOpts, HealRequestSource, HealScanMode};
 use rustfs_madmin::heal_commands::HealResultItem;
 use rustfs_utils::path::SLASH_SEPARATOR;
 use serde::{Deserialize, Serialize};

crates/heal/src/heal/task/heal_bucket.rs

@@ -16,6 +16,22 @@ use super::*;
 use crate::heal::progress::{add_bytes, increment_counter, stable_generation};
 use crate::heal::utils::format_set_disk_id;
 
+fn unavailable_recreate_error(result: &HealResultItem, opts: &HealOpts) -> Option<Error> {
+    if opts.dry_run || !opts.recreate {
+        return None;
+    }
+
+    let mut offline = false;
+    for drive in &result.after.drives {
+        if drive.state == DriveState::Faulty.to_str() {
+            return Some(Error::Disk(DiskError::FaultyDisk));
+        }
+        offline |= drive.state == DriveState::Offline.to_str();
+    }
+
+    offline.then_some(Error::Disk(DiskError::DiskNotFound))
+}
+
 impl HealTask {
     pub(super) async fn heal_bucket(&self, bucket: &str) -> Result<()> {
         debug!(
@@ -335,13 +351,16 @@ impl HealTask {
                             )
                             .await
                         {
-                            Ok((result, None)) => {
+                            Ok((result, None)) => match unavailable_recreate_error(&result, &heal_opts) {
+                                Some(error) => Some(error),
+                                None => {
                                     telemetry_unknown |= !increment_counter(&mut healed);
                                     telemetry_unknown |=
                                         !add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX));
                                     self.record_result_item(result).await;
                                     None
                                 }
+                            },
                             Ok((_, Some(err))) if is_missing_object_dir_heal_result(object, &err) => {
                                 telemetry_unknown |= !increment_counter(&mut healed);
                                 debug!(

crates/heal/src/heal/task/tests.rs

@@ -706,11 +706,28 @@ enum MockHealObjectOutcome {
     OkWithOtherError(&'static str),
     ErrOther(&'static str),
     DanglingGraceDeferred,
+    UnavailableDrive(DriveState),
     RetryableReadQuorum,
     RetryableSlowDown,
     PermanentOther(&'static str),
 }
 
+fn unavailable_drive_heal_result(state: DriveState) -> (HealResultItem, Option<Error>) {
+    (
+        HealResultItem {
+            after: Infos {
+                drives: vec![HealDriveInfo {
+                    endpoint: "remote-target".to_string(),
+                    state: state.to_string(),
+                    ..Default::default()
+                }],
+            },
+            ..Default::default()
+        },
+        None,
+    )
+}
+
 #[derive(Clone, Copy)]
 enum MockObjectExists {
     Exists(bool),
@@ -813,6 +830,7 @@ impl HealStorageAPI for MockStorage {
                         "dangling object deletion deferred by heal grace window; retry_after_secs=3599; grace_secs=3600",
                     ))),
                 )),
+                MockHealObjectOutcome::UnavailableDrive(state) => Ok(unavailable_drive_heal_result(state)),
                 MockHealObjectOutcome::RetryableReadQuorum => Err(Error::Storage(EcstoreError::InsufficientReadQuorum(
                     bucket.to_string(),
                     object.to_string(),
@@ -833,6 +851,7 @@ impl HealStorageAPI for MockStorage {
                         "dangling object deletion deferred by heal grace window; retry_after_secs=3599; grace_secs=3600",
                     ))),
                 )),
+                MockHealObjectOutcome::UnavailableDrive(state) => Ok(unavailable_drive_heal_result(state)),
                 MockHealObjectOutcome::OkWithOtherError(message) => Ok((HealResultItem::default(), Some(Error::other(message)))),
                 MockHealObjectOutcome::ErrOther(message) | MockHealObjectOutcome::PermanentOther(message) => {
                     Err(Error::other(message))
@@ -1449,6 +1468,46 @@ async fn test_recursive_bucket_heal_retries_only_retryable_objects() {
     assert_eq!(progress.objects_failed, 0);
 }
 
+#[tokio::test(start_paused = true)]
+async fn recursive_bucket_heal_retries_when_recreate_target_is_unavailable() {
+    for state in [DriveState::Offline, DriveState::Faulty] {
+        let state_name = state.to_string();
+        let storage = Arc::new(MockStorage::default());
+        storage
+            .heal_object_outcomes
+            .lock()
+            .unwrap()
+            .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::UnavailableDrive(state)]));
+        let request = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket-a".to_string(),
+            },
+            HealOptions {
+                recursive: true,
+                recreate_missing: true,
+                timeout: None,
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        );
+        let task = HealTask::from_request(request, storage.clone());
+
+        task.heal_bucket("bucket-a")
+            .await
+            .expect("an unavailable recreate target should be retried after it returns");
+
+        assert_eq!(
+            storage.heal_object_calls.lock().unwrap().as_slice(),
+            ["object-a".to_string(), "object-b".to_string(), "object-a".to_string()],
+            "unexpected calls for unavailable state {state_name}"
+        );
+        let progress = task.get_progress().await;
+        assert_eq!(progress.objects_scanned, 2, "unexpected scanned count for state {state_name}");
+        assert_eq!(progress.objects_healed, 2, "unexpected healed count for state {state_name}");
+        assert_eq!(progress.objects_failed, 0, "unexpected failed count for state {state_name}");
+    }
+}
+
 #[tokio::test(start_paused = true)]
 async fn recursive_bucket_heal_skips_dangling_delete_grace_without_batch_failure() {
     let storage = Arc::new(MockStorage::default());
@@ -1486,6 +1545,90 @@ async fn recursive_bucket_heal_skips_dangling_delete_grace_without_batch_failure
     assert_eq!(progress.skipped_objects, 1);
 }
 
+#[tokio::test(start_paused = true)]
+async fn recursive_bucket_heal_preserves_non_recreate_and_non_availability_results() {
+    for (dry_run, recreate_missing, state) in [
+        (true, true, DriveState::Offline),
+        (true, true, DriveState::Faulty),
+        (false, false, DriveState::Offline),
+        (false, false, DriveState::Faulty),
+        (false, true, DriveState::Ok),
+        (false, true, DriveState::Missing),
+        (false, true, DriveState::Corrupt),
+        (false, true, DriveState::PermissionDenied),
+        (false, true, DriveState::Unknown("other failure".to_string())),
+    ] {
+        let storage = Arc::new(MockStorage::default());
+        storage
+            .heal_object_outcomes
+            .lock()
+            .expect("test outcome lock")
+            .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::UnavailableDrive(state)]));
+        let task = HealTask::from_request(
+            HealRequest::new(
+                HealType::Bucket {
+                    bucket: "bucket-a".to_string(),
+                },
+                HealOptions {
+                    recursive: true,
+                    dry_run,
+                    recreate_missing,
+                    timeout: None,
+                    ..Default::default()
+                },
+                HealPriority::Normal,
+            ),
+            storage.clone(),
+        );
+        task.heal_bucket("bucket-a")
+            .await
+            .expect("unchanged best-effort result should not schedule an availability retry");
+        assert_eq!(
+            storage.heal_object_calls.lock().expect("test call lock").as_slice(),
+            ["object-a".to_string(), "object-b".to_string()]
+        );
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn recursive_bucket_heal_exhausts_unavailable_target_without_rescanning_healthy_objects() {
+    let storage = Arc::new(MockStorage::default());
+    storage.heal_object_outcomes.lock().expect("test outcome lock").insert(
+        "object-a".to_string(),
+        (0..4)
+            .map(|_| MockHealObjectOutcome::UnavailableDrive(DriveState::Faulty))
+            .collect(),
+    );
+    let task = HealTask::from_request(
+        HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket-a".to_string(),
+            },
+            HealOptions {
+                recursive: true,
+                recreate_missing: true,
+                timeout: None,
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        ),
+        storage.clone(),
+    );
+    let error = task
+        .heal_bucket("bucket-a")
+        .await
+        .expect_err("persistent unavailable target must not report success");
+    assert!(matches!(error, Error::TaskExecutionFailed { .. }));
+    let failure = task
+        .take_batch_failure()
+        .await
+        .expect("exhausted availability failure should be retained");
+    assert_eq!((failure.failed, failure.retryable, failure.permanent), (1, 1, 0));
+    let calls = storage.heal_object_calls.lock().expect("test call lock");
+    assert_eq!(calls.iter().filter(|object| object.as_str() == "object-a").count(), 4);
+    assert_eq!(calls.iter().filter(|object| object.as_str() == "object-b").count(), 1);
+}
+
 #[tokio::test(start_paused = true)]
 async fn test_recursive_bucket_heal_reports_typed_exhausted_and_permanent_failures() {
     let storage = Arc::new(MockStorage::default());

docs/testing/e2e-suite-inventory.md

@@ -57,7 +57,7 @@
 | group_delete_test | 4 |  |
 | head_object_consistency_test | 1 | ✅ |
 | head_object_range_test | 1 | ✅ |
-| heal_erasure_disk_rebuild_test | 4 | 🌙 |
+| heal_erasure_disk_rebuild_test | 5 | 🌙 |
 | inline_fast_path_cluster_test | 16 |  |
 | internode_rpc_signature_e2e_test | 5 |  |
 | kms | 50 |  |
@@ -103,4 +103,4 @@
 | tls_hot_reload_test | 1 | ✅ |
 | version_id_regression_test | 10 | ✅ |
 
-**Total listed: 621 tests across 86 modules · PR smoke: 165 tests / 36 modules · merge/main full: 495 tests / 77 modules · nightly replication: 56 tests · nightly cluster faults: 31 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-31.
+**Total listed: 622 tests across 86 modules · PR smoke: 165 tests / 36 modules · merge/main full: 495 tests / 77 modules · nightly replication: 56 tests · nightly cluster faults: 32 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-31.