fix(heal): fall back to set-wide format for directory-backed targets (#7331)

fix(heal): fall back to set-wide format for directory-backed replacement targets

Since #7018 `renew_disk` routes an unformatted local endpoint through the automatic replacement heal, which requires the target to be an independently mounted disk. Directory-backed deployments (the operator set `RUSTFS_UNSAFE_BYPASS_DISK_CHECK`, which the startup disk-independence check already requires for endpoints sharing a device) can never pass that admission, so a runtime-wiped or replaced directory disk stayed unformatted forever: the heal task failed with "replacement target is not a stable mounted disk" and the auto-scan kept deferring the endpoint. This broke the Issue #1533 contract and the `heal_erasure_disk_rebuild_test` lane on main since 2026-09-02.

When the disk-check bypass is set and the replacement target fails admission, the erasure-set heal now logs a warning and downgrades to the ordinary set-wide `heal_format` path that predated replacement admission, and the auto-scan no longer defers such endpoints. The mount admission itself is unchanged and still cannot be bypassed by any environment variable; deployments without the bypass keep failing closed.

The endpoint-blackhole heal scenario now probes whether `iptables` can read the OUTPUT chain and logs an explicit skip when the host lacks `CAP_NET_ADMIN` (containerised runners report "Permission denied" from the nf_tables backend even under sudo); `RUSTFS_E2E_REQUIRE_NET_FAULT_INJECTION=1` turns that into a failure for lanes that provision the capability. The CI full-gate job surfaces the missing capability as a workflow warning, and the runtime-wipe fixture retries `remove_dir_all` on the listing race macOS surfaces as `DirectoryNotEmpty`.

Refs rustfs/backlog#2357.

crates/e2e_test/README.md

@@ -184,6 +184,8 @@ the wiring source of truth. Committed test-ID digests under
 
 ## Troubleshooting
 
+**Endpoint blackhole scenario skipped** — `heal_erasure_disk_rebuild_test::tests::test_cluster_root_heal_recovers_after_target_endpoint_blackhole` installs a loopback `iptables` DROP rule and therefore needs `CAP_NET_ADMIN` (root or passwordless `sudo -n iptables`). A host where `iptables` is missing or cannot read the OUTPUT chain (typical inside an unprivileged container, where the nf_tables backend reports "Permission denied" even under `sudo`) logs a `heal_interruption_skipped` warning and returns without exercising heal. Set `RUSTFS_E2E_REQUIRE_NET_FAULT_INJECTION=1` on lanes that do provision the capability so a broken runner fails instead of skipping.
+
 **Reproduce a CI failure locally** — run the exact profile/lane:
 
 ```bash

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -34,6 +34,8 @@ mod tests {
     use tokio::net::TcpStream;
     use tokio::time::{Duration, Instant, sleep, timeout};
     use tracing::info;
+    #[cfg(target_os = "linux")]
+    use tracing::warn;
 
     const POOL_METADATA_OBJECT: &str = "pool.bin";
 
@@ -115,6 +117,49 @@ mod tests {
     }
 
     impl TcpPortBlackhole {
+        /// Environment flag that turns an unusable fault-injection host into a
+        /// hard failure instead of a logged skip. Lanes that provision
+        /// `CAP_NET_ADMIN` set it so a broken runner cannot pass silently.
+        #[cfg(target_os = "linux")]
+        const REQUIRE_ENV: &str = "RUSTFS_E2E_REQUIRE_NET_FAULT_INJECTION";
+
+        /// Probe whether this host can manipulate the OUTPUT chain at all.
+        ///
+        /// Returns `Ok(Some(reason))` when `iptables` is missing or lacks
+        /// `CAP_NET_ADMIN` (the nf_tables backend reports "Permission denied"
+        /// even under `sudo` inside an unprivileged container) and the lane did
+        /// not demand fault injection; returns an error when the lane demands
+        /// it; returns `Ok(None)` when the blackhole can be installed.
+        #[cfg(target_os = "linux")]
+        fn unavailable_reason() -> Result<Option<String>, Box<dyn Error + Send + Sync>> {
+            let id = Command::new("id").arg("-u").output()?;
+            if !id.status.success() {
+                return Err(format!("failed to determine the test process uid: {}", String::from_utf8_lossy(&id.stderr)).into());
+            }
+            let use_sudo = String::from_utf8_lossy(&id.stdout).trim() != "0";
+            let mut command = if use_sudo {
+                let mut command = Command::new("sudo");
+                command.args(["-n", "iptables"]);
+                command
+            } else {
+                Command::new("iptables")
+            };
+            let probe = command.args(["-w", "5", "-S", "OUTPUT"]).output();
+            let reason = match probe {
+                Ok(output) if output.status.success() => return Ok(None),
+                Ok(output) => format!(
+                    "iptables cannot read the OUTPUT chain (status {}): {}",
+                    output.status,
+                    String::from_utf8_lossy(&output.stderr).trim()
+                ),
+                Err(err) => format!("iptables is not runnable: {err}"),
+            };
+            if std::env::var_os(Self::REQUIRE_ENV).is_some() {
+                return Err(format!("{} is set but network fault injection is unavailable: {reason}", Self::REQUIRE_ENV).into());
+            }
+            Ok(Some(reason))
+        }
+
         fn install(address: &str) -> Result<Self, Box<dyn Error + Send + Sync>> {
             let address = address.parse::<SocketAddr>()?;
             if !address.ip().is_loopback() {
@@ -199,6 +244,27 @@ mod tests {
         }
     }
 
+    /// Remove a disk directory underneath a running server. Background writers
+    /// (scanner, usage cache, heal markers) can recreate entries between the
+    /// recursive listing and the final `rmdir`, which surfaces as
+    /// `DirectoryNotEmpty` on macOS; retry briefly so the wipe reflects the
+    /// operator action rather than a listing race.
+    fn wipe_directory_while_server_runs(disk: &Path) -> std::io::Result<()> {
+        let mut last_err = None;
+        for _ in 0..20 {
+            match std::fs::remove_dir_all(disk) {
+                Ok(()) => return Ok(()),
+                Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()),
+                Err(err) if err.kind() == std::io::ErrorKind::DirectoryNotEmpty => {
+                    last_err = Some(err);
+                    std::thread::sleep(std::time::Duration::from_millis(100));
+                }
+                Err(err) => return Err(err),
+            }
+        }
+        Err(last_err.expect("retry loop only exits without success after recording an error"))
+    }
+
     fn has_file_under(path: &Path) -> bool {
         let Ok(entries) = std::fs::read_dir(path) else {
             return false;
@@ -481,7 +547,7 @@ mod tests {
             );
         }
 
-        std::fs::remove_dir_all(&disk0).expect("disk0 wipe should succeed while server is running");
+        wipe_directory_while_server_runs(&disk0).expect("disk0 wipe should succeed while server is running");
         std::fs::create_dir_all(&disk0).expect("disk0 should be recreated empty while server is running");
         assert!(!has_file_under(&disk0), "disk0 must be empty immediately after runtime wipe");
 
@@ -868,6 +934,18 @@ mod tests {
     #[cfg(target_os = "linux")]
     #[tokio::test(flavor = "multi_thread")]
     async fn test_cluster_root_heal_recovers_after_target_endpoint_blackhole() -> Result<(), Box<dyn Error + Send + Sync>> {
+        if let Some(reason) = TcpPortBlackhole::unavailable_reason()? {
+            init_logging();
+            warn!(
+                event = "heal_interruption_skipped",
+                component = "e2e_test",
+                subsystem = "heal",
+                interruption_kind = "target_endpoint_blackhole",
+                reason,
+                "Skipping endpoint blackhole scenario: network fault injection is unavailable on this host"
+            );
+            return Ok(());
+        }
         timeout(
             Duration::from_secs(420),
             run_cluster_root_heal_interruption(InterruptionScenario::TargetEndpointBlackhole),

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

@@ -150,6 +150,7 @@ impl HealManager {
                                 Err(DiskError::UnformattedDisk) => {
                                     if !super::super::replacement_readiness::auto_replacement_target_ready(disk, &local_disks)
                                         .await
+                                        && !super::super::replacement_readiness::directory_backed_replacement_fallback_enabled()
                                     {
                                         deferred_replacement_endpoints.insert(endpoint.to_string());
                                         skipped_invalid_count += 1;

crates/heal/src/heal/replacement_readiness.rs

@@ -18,6 +18,25 @@ use super::{
     DiskOption, DiskStore, Endpoint, HealDiskExt as _, local_disk_map_read, new_disk, resume::ReplacementTargetIdentity,
 };
 
+/// Whether automatic replacement may fall back to the set-wide format heal when
+/// a target cannot pass the independent-mount admission.
+///
+/// Directory-backed deployments already declare, through
+/// `RUSTFS_UNSAFE_BYPASS_DISK_CHECK`, that their endpoints are plain
+/// directories sharing a device with the host root. Those endpoints can never
+/// satisfy [`auto_replacement_target_identity`], so without this fallback a
+/// runtime-wiped or replaced directory disk would stay deferred forever. The
+/// admission check itself is never bypassed; the fallback only routes the heal
+/// through the ordinary format path that formats every unformatted disk in the
+/// set, which is exactly what the pre-admission `heal_disk` path did.
+pub(crate) fn directory_backed_replacement_fallback_enabled() -> bool {
+    rustfs_utils::get_env_bool_with_aliases(
+        rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK,
+        &[rustfs_config::ENV_MINIO_CI],
+        rustfs_config::DEFAULT_UNSAFE_BYPASS_DISK_CHECK,
+    )
+}
+
 pub(crate) async fn auto_replacement_target_ready(disk: &DiskStore, local_disks: &[DiskStore]) -> bool {
     auto_replacement_target_identity(disk, local_disks).await.is_some()
 }
@@ -184,12 +203,47 @@ mod tests {
         assert!(endpoint.is_local);
     }
 
+    #[test]
+    fn directory_backed_fallback_is_off_by_default() {
+        temp_env::with_vars(
+            [
+                (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>),
+                (rustfs_config::ENV_MINIO_CI, None::<&str>),
+            ],
+            || assert!(!directory_backed_replacement_fallback_enabled()),
+        );
+    }
+
+    #[test]
+    fn directory_backed_fallback_follows_the_disk_check_bypass() {
+        temp_env::with_vars(
+            [
+                (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, Some("true")),
+                (rustfs_config::ENV_MINIO_CI, None::<&str>),
+            ],
+            || assert!(directory_backed_replacement_fallback_enabled()),
+        );
+        temp_env::with_vars(
+            [
+                (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, Some("false")),
+                (rustfs_config::ENV_MINIO_CI, Some("true")),
+            ],
+            || {
+                assert!(
+                    !directory_backed_replacement_fallback_enabled(),
+                    "the canonical key must win over the alias"
+                )
+            },
+        );
+    }
+
     #[tokio::test]
     async fn runtime_environment_cannot_bypass_mount_admission() {
         temp_env::async_with_vars(
             [
                 ("RUSTFS_TEST_AUTO_REPLACEMENT_READINESS_BYPASS", Some("1")),
                 ("RUSTFS_E2E_AUTO_REPLACEMENT_READINESS_BYPASS", Some("1")),
+                (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, Some("true")),
             ],
             async {
                 let temp = TempDir::new().expect("temporary replacement root should be created");

crates/heal/src/heal/task/heal_erasure_set.rs

@@ -42,7 +42,36 @@ impl HealTask {
             progress.update_stage(0, 4);
         }
 
-        let is_auto_replacement = matches!(self.source, HealRequestSource::AutoHeal) && !self.heal_endpoints.is_empty();
+        let mut is_auto_replacement = matches!(self.source, HealRequestSource::AutoHeal) && !self.heal_endpoints.is_empty();
+        if is_auto_replacement
+            && crate::heal::replacement_readiness::directory_backed_replacement_fallback_enabled()
+            && self
+                .await_with_control(self.storage.replacement_target_identities(&self.heal_endpoints))
+                .await
+                .is_err()
+        {
+            // Directory-backed endpoints cannot pass replacement admission; the
+            // operator opted out of disk checks, so heal the set the way the
+            // pre-admission `heal_disk` path did instead of deferring forever.
+            warn!(
+                target: "rustfs::heal::task",
+                event = EVENT_HEAL_ERASURE_SET_STAGE,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_TASK,
+                task_id = %self.id,
+                set_disk_id,
+                stage = "replacement_admission",
+                result = "directory_backed_fallback",
+                target_count = self.heal_endpoints.len(),
+                "Heal erasure set falls back to set-wide format heal for a replacement target that is not an independently mounted disk"
+            );
+            is_auto_replacement = false;
+        }
+        let replacement_targets = if is_auto_replacement {
+            self.heal_endpoints.clone()
+        } else {
+            Vec::new()
+        };
         let replacement_resume_disk = if is_auto_replacement {
             let mut requested_targets = self.heal_endpoints.clone();
             requested_targets.sort_unstable();
@@ -421,7 +450,7 @@ impl HealTask {
             heal_opts,
             self.source,
         )
-        .with_replacement_targets(self.heal_endpoints.clone(), is_auto_replacement.then(|| self.id.clone()))
+        .with_replacement_targets(replacement_targets, is_auto_replacement.then(|| self.id.clone()))
         .with_replacement_identity_fence(replacement_target_identities.clone())
         .with_mainline_pacer(self.mainline_pacer.clone());
 

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

@@ -480,6 +480,95 @@ async fn automatic_replacement_uses_target_scoped_format() {
     );
 }
 
+fn directory_backed_replacement_request() -> HealRequest {
+    let mut request = HealRequest::new(
+        HealType::ErasureSet {
+            buckets: Vec::new(),
+            set_disk_id: "pool_0_set_0".to_string(),
+        },
+        HealOptions {
+            pool_index: Some(0),
+            set_index: Some(0),
+            ..Default::default()
+        },
+        HealPriority::Low,
+    );
+    request.source = HealRequestSource::AutoHeal;
+    request.heal_endpoints = vec!["/data/disk0".to_string()];
+    request
+}
+
+#[tokio::test]
+async fn directory_backed_replacement_falls_back_to_set_format_when_disk_checks_are_bypassed() {
+    temp_env::async_with_vars(
+        [
+            (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, Some("true")),
+            (rustfs_config::ENV_MINIO_CI, None::<&str>),
+        ],
+        async {
+            let storage = Arc::new(MockStorage {
+                replacement_target_identities_ready: Mutex::new(false),
+                global_format_ok_endpoints: Mutex::new(vec!["/data/disk0".to_string()]),
+                ..Default::default()
+            });
+            let task = HealTask::from_request(directory_backed_replacement_request(), storage.clone());
+
+            // The mock has no local disk behind "/data/disk0", so the run stops at
+            // the healing-marker step that follows the format stage, exactly like
+            // `automatic_replacement_uses_target_scoped_format`. The assertions
+            // below pin which format path ran before that point.
+            let err = task.execute().await.expect_err("the mock has no local healing marker target");
+            assert!(
+                err.to_string().contains("healing marker target is unavailable"),
+                "the fallback must reach the post-format marker step, got: {err}"
+            );
+
+            assert_eq!(
+                *storage.global_format_calls.lock().unwrap(),
+                1,
+                "the fallback must run exactly one set-wide format heal"
+            );
+            assert!(
+                storage.replacement_format_calls.lock().unwrap().is_empty(),
+                "the fallback must not run the target-scoped replacement format"
+            );
+        },
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn directory_backed_replacement_stays_fail_closed_without_disk_check_bypass() {
+    temp_env::async_with_vars(
+        [
+            (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>),
+            (rustfs_config::ENV_MINIO_CI, None::<&str>),
+        ],
+        async {
+            let storage = Arc::new(MockStorage {
+                replacement_target_identities_ready: Mutex::new(false),
+                ..Default::default()
+            });
+            let task = HealTask::from_request(directory_backed_replacement_request(), storage.clone());
+
+            task.execute()
+                .await
+                .expect_err("an inadmissible replacement target must keep failing closed");
+
+            assert_eq!(
+                *storage.global_format_calls.lock().unwrap(),
+                0,
+                "fail-closed admission must not format the set"
+            );
+            assert!(
+                storage.replacement_format_calls.lock().unwrap().is_empty(),
+                "fail-closed admission must not format the target"
+            );
+        },
+    )
+    .await;
+}
+
 #[tokio::test]
 async fn automatic_replacement_persists_intent_before_format() {
     let storage = Arc::new(MockStorage {
@@ -962,6 +1051,8 @@ struct MockStorage {
     format_no_heal_required: Mutex<bool>,
     format_error: Mutex<Option<Error>>,
     global_format_calls: Mutex<u32>,
+    /// Endpoints the set-wide format mock reports as freshly formatted (`state == "ok"`).
+    global_format_ok_endpoints: Mutex<Vec<String>>,
     replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
     replacement_target_identities_ready: Mutex<bool>,
     replacement_target_identity_sequences: Mutex<VecDeque<Vec<crate::heal::resume::ReplacementTargetIdentity>>>,
@@ -1338,10 +1429,26 @@ impl HealStorageAPI for MockStorage {
             return Err(error);
         }
         let no_heal_required = *self.format_no_heal_required.lock().unwrap();
+        let result = HealResultItem {
+            after: Infos {
+                drives: self
+                    .global_format_ok_endpoints
+                    .lock()
+                    .unwrap()
+                    .iter()
+                    .map(|endpoint| HealDriveInfo {
+                        endpoint: endpoint.clone(),
+                        state: "ok".to_string(),
+                        ..Default::default()
+                    })
+                    .collect(),
+            },
+            ..Default::default()
+        };
         if no_heal_required {
-            Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired))))
+            Ok((result, Some(Error::Storage(EcstoreError::NoHealRequired))))
         } else {
-            Ok((HealResultItem::default(), None))
+            Ok((result, None))
         }
     }