Merge branch 'main' into cxymds/fix-5416-k8s-quorum-startup

Cargo.lock

@@ -9128,6 +9128,7 @@ dependencies = [
  "google-cloud-storage",
  "hex-simd",
  "hmac 0.13.0",
+ "hostname",
  "hotpath",
  "http 1.5.0",
  "http-body 1.1.0",

Cargo.toml

@@ -256,6 +256,7 @@ hashbrown = { version = "0.17.1" }
 hex = "0.4.3"
 hex-simd = "0.8.0"
 highway = { version = "1.3.0" }
+hostname = "0.4.2"
 ipnetwork = { version = "0.21.1" }
 lazy_static = "1.5.0"
 libc = "0.2.189"

crates/config/README.md

@@ -66,6 +66,10 @@ Current guidance:
 
 - `RUSTFS_BROWSER_REDIRECT_URL` sets the externally reachable browser origin used for OIDC callback, console success redirect, and logout fallback URLs. Configure it to the public scheme and authority without a path, for example `https://console.example.com`. In load-balancer deployments, keep OIDC authorize and callback requests on the same backend node because the in-flight OIDC `state` is local to the RustFS node.
 
+## Distributed endpoint locality
+
+- `RUSTFS_LOCAL_ENDPOINT_HOST` identifies this server's host in a distributed `RUSTFS_VOLUMES` topology without resolving every peer during startup. Set it to exactly one host, without a scheme, port, or path. It is accepted only for orchestrated URL topologies and must match at least one endpoint on the RustFS server port; invalid or unmatched values fail startup. Leave it unset to retain DNS-based locality discovery.
+
 ## Scanner environment aliases
 
 - `RUSTFS_SCANNER_SPEED` (canonical, also accepts `MINIO_SCANNER_SPEED`)

crates/config/src/constants/app.rs

@@ -131,6 +131,10 @@ pub const ENV_RUSTFS_ADDRESS: &str = "RUSTFS_ADDRESS";
 /// Environment variable for server volumes.
 pub const ENV_RUSTFS_VOLUMES: &str = "RUSTFS_VOLUMES";
 
+/// Environment variable identifying this server's host in distributed endpoint
+/// lists without relying on DNS locality discovery.
+pub const ENV_LOCAL_ENDPOINT_HOST: &str = "RUSTFS_LOCAL_ENDPOINT_HOST";
+
 /// Environment variable to explicitly bypass local physical disk independence checks.
 pub const ENV_UNSAFE_BYPASS_DISK_CHECK: &str = "RUSTFS_UNSAFE_BYPASS_DISK_CHECK";
 

crates/ecstore/Cargo.toml

@@ -105,6 +105,7 @@ tempfile.workspace = true
 hyper = { workspace = true, features = ["http2", "http1", "server"] }
 hyper-util = { workspace = true, features = ["tokio", "server-auto", "server-graceful", "tracing"] }
 hyper-rustls = { workspace = true, default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "http2", "aws-lc-rs"] }
+hostname.workspace = true
 rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] }
 rustls-pki-types.workspace = true
 tokio = { workspace = true, features = ["io-util", "sync", "signal", "fs", "rt-multi-thread"] }

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

@@ -58,7 +58,7 @@ use std::{
     collections::HashMap,
     io::Cursor,
     sync::{
-        Arc,
+        Arc, Weak,
         atomic::{AtomicBool, Ordering},
     },
     time::SystemTime,
@@ -350,6 +350,44 @@ impl PeerRestClient {
         }
     }
 
+    fn parse_topology_host(peer_host_port: &str, grid_host: &str) -> Result<XHost> {
+        let url = url::Url::parse(grid_host).map_err(|_| Error::other("peer grid host is not a valid URL"))?;
+        if !matches!(url.scheme(), "http" | "https")
+            || !url.username().is_empty()
+            || url.password().is_some()
+            || url.query().is_some()
+            || url.fragment().is_some()
+            || url.path() != "/"
+        {
+            return Err(Error::other("peer grid host has an invalid URL shape"));
+        }
+        let url_host = url.host().ok_or_else(|| Error::other("peer grid host is missing a host"))?;
+        let topology_host = match url.port() {
+            Some(port) => format!("{url_host}:{port}"),
+            None => url_host.to_string(),
+        };
+        let explicit_port = url.port();
+        let name = match url_host {
+            url::Host::Domain(domain) => domain.to_string(),
+            url::Host::Ipv4(address) => address.to_string(),
+            url::Host::Ipv6(address) if explicit_port.is_none() => format!("[{address}]"),
+            url::Host::Ipv6(address) => address.to_string(),
+        };
+        let port = url
+            .port_or_known_default()
+            .filter(|port| *port > 0)
+            .ok_or_else(|| Error::other("peer grid host is missing a valid port"))?;
+        let host = XHost {
+            name,
+            port,
+            is_port_set: explicit_port.is_some(),
+        };
+        if topology_host != peer_host_port {
+            return Err(Error::other("peer topology host does not match its grid URL"));
+        }
+        Ok(host)
+    }
+
     fn build_clients_from_slots(
         slots: Vec<(String, Option<String>, bool)>,
     ) -> (Vec<Option<Self>>, Vec<Option<Self>>, Vec<String>) {
@@ -363,14 +401,14 @@ impl PeerRestClient {
             }
 
             let client = match grid_host {
-                Some(grid_host) => match XHost::try_from(peer_host_port.clone()) {
+                Some(grid_host) => match Self::parse_topology_host(&peer_host_port, &grid_host) {
                     Ok(host) => {
                         let mut client = PeerRestClient::new(host, grid_host);
                         client.topology_member = peer_host_port.clone();
                         Some(client)
                     }
                     Err(err) => {
-                        warn!(peer = %peer_host_port, "Xhost parse failed while constructing peer client: {err:?}");
+                        warn!(peer = %peer_host_port, "peer topology host parse failed while constructing peer client: {err:?}");
                         None
                     }
                 },
@@ -519,9 +557,8 @@ impl PeerRestClient {
         }
 
         let grid_host = self.grid_host.clone();
-        let offline = Arc::clone(&self.offline);
-        let recovery_running = Arc::clone(&self.recovery_running);
-        let span = Self::recovery_monitor_span(&grid_host);
+        let offline = Arc::downgrade(&self.offline);
+        let recovery_running = Arc::downgrade(&self.recovery_running);
         // The offline flag and its recovery are the silent half of
         // rustfs/backlog#888: log the monitor's start and its success so an
         // "offline then back" episode leaves a trace on the observing node.
@@ -530,13 +567,34 @@ impl PeerRestClient {
             grid_host = %self.grid_host,
             "peer RPC connection marked offline after a network-like failure; starting background recovery monitor"
         );
+        drop(Self::spawn_recovery_monitor(grid_host, offline, recovery_running));
+    }
+
+    fn spawn_recovery_monitor(
+        grid_host: String,
+        offline: Weak<AtomicBool>,
+        recovery_running: Weak<AtomicBool>,
+    ) -> tokio::task::JoinHandle<()> {
+        let span = Self::recovery_monitor_span(&grid_host);
         super::spawn_background_monitor(span, async move {
             let mut delay = get_drive_active_check_interval();
             let connect_timeout = get_drive_active_check_timeout();
 
             for attempt in 1..=PEER_REST_RECOVERY_MAX_ATTEMPTS {
+                if offline.strong_count() == 0 || recovery_running.strong_count() == 0 {
+                    return;
+                }
                 tokio::time::sleep(delay).await;
+                if offline.strong_count() == 0 || recovery_running.strong_count() == 0 {
+                    return;
+                }
                 if Self::perform_connectivity_check(&grid_host, connect_timeout).await.is_ok() {
+                    let Some(offline) = offline.upgrade() else {
+                        return;
+                    };
+                    let Some(recovery_running) = recovery_running.upgrade() else {
+                        return;
+                    };
                     offline.store(false, Ordering::Release);
                     recovery_running.store(false, Ordering::Release);
                     info!(
@@ -556,8 +614,10 @@ impl PeerRestClient {
                 attempts = PEER_REST_RECOVERY_MAX_ATTEMPTS,
                 "peer recovery monitor reached max attempts; will retry on next request"
             );
-            recovery_running.store(false, Ordering::Release);
-        });
+            if let Some(recovery_running) = recovery_running.upgrade() {
+                recovery_running.store(false, Ordering::Release);
+            }
+        })
     }
 
     #[cfg(test)]
@@ -1807,9 +1867,13 @@ fn tier_config_reload_status_outcome(status: tonic::Status) -> TierConfigReloadO
 mod tests {
     use super::*;
     use crate::config::com::STORAGE_CLASS_SUB_SYS;
+    use crate::layout::{disks_layout::DisksLayout, endpoints::SetupType};
+    use rustfs_config::{ENV_KUBERNETES_SERVICE_HOST, ENV_LOCAL_ENDPOINT_HOST, ENV_STARTUP_TOPOLOGY_WAIT_MODE};
     use serde_json::Value;
+    use serial_test::serial;
     use std::io::{self, Write};
     use std::sync::{Arc, Mutex};
+    use temp_env::async_with_vars;
     use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
 
     #[test]
@@ -1927,30 +1991,115 @@ mod tests {
     fn build_clients_from_slots_preserves_missing_remote_topology_slots() {
         let slots = vec![
             ("127.0.0.1:9000".to_string(), None, true),
-            ("127.0.0.1:9001".to_string(), Some("http://127.0.0.1:9001".to_string()), false),
+            (
+                "rustfs-1.invalid:9001".to_string(),
+                Some("http://rustfs-1.invalid:9001".to_string()),
+                false,
+            ),
+            ("rustfs-2.invalid".to_string(), Some("http://rustfs-2.invalid".to_string()), false),
             ("127.0.0.1:notaport".to_string(), Some("http://127.0.0.1:notaport".to_string()), false),
             ("127.0.0.1:9003".to_string(), None, false),
         ];
 
         let (remote, all, remote_topology_hosts) = PeerRestClient::build_clients_from_slots(slots);
 
-        assert_eq!(remote.len(), 3, "local node is excluded but remote slots are not compacted away");
-        assert_eq!(all.len(), 4, "all slots preserve the sorted cluster topology shape");
+        assert_eq!(remote.len(), 4, "local node is excluded but remote slots are not compacted away");
+        assert_eq!(all.len(), 5, "all slots preserve the sorted cluster topology shape");
         assert_eq!(
             remote_topology_hosts,
             vec![
-                "127.0.0.1:9001".to_string(),
+                "rustfs-1.invalid:9001".to_string(),
+                "rustfs-2.invalid".to_string(),
                 "127.0.0.1:notaport".to_string(),
                 "127.0.0.1:9003".to_string()
             ]
         );
-        assert!(remote[0].is_some(), "valid remote peer should get a client");
-        assert!(remote[1].is_none(), "unparseable remote peer should remain observable as a missing slot");
-        assert!(remote[2].is_none(), "missing grid host should remain observable as a missing slot");
+        let unresolved = remote[0]
+            .as_ref()
+            .expect("temporarily unresolved remote peer should retain a client");
+        assert_eq!(unresolved.host.to_string(), "rustfs-1.invalid:9001");
+        let default_port = remote[1]
+            .as_ref()
+            .expect("temporarily unresolved scheme-default remote peer should retain a client");
+        assert_eq!(default_port.host.to_string(), "rustfs-2.invalid");
+        assert_eq!(default_port.host.port, 80);
+        assert!(!default_port.host.is_port_set);
+        assert!(remote[2].is_none(), "unparseable remote peer should remain observable as a missing slot");
+        assert!(remote[3].is_none(), "missing grid host should remain observable as a missing slot");
         assert!(all[0].is_none(), "local node is represented by the local server_info row");
         assert!(all[1].is_some());
-        assert!(all[2].is_none());
+        assert!(all[2].is_some());
         assert!(all[3].is_none());
+        assert!(all[4].is_none());
+    }
+
+    #[test]
+    fn topology_host_parser_preserves_names_and_bracketed_ipv6() {
+        let domain = PeerRestClient::parse_topology_host("rustfs-1.invalid", "https://rustfs-1.invalid")
+            .expect("unresolved HTTPS topology host should parse without DNS");
+        assert_eq!(domain.to_string(), "rustfs-1.invalid");
+        assert_eq!(domain.port, 443);
+        assert!(!domain.is_port_set);
+
+        let ipv6 = PeerRestClient::parse_topology_host("[2001:db8::1]:9000", "http://[2001:db8::1]:9000")
+            .expect("bracketed IPv6 topology host should parse without changing its identity");
+        assert_eq!(ipv6.to_string(), "[2001:db8::1]:9000");
+
+        let default_port_ipv6 = PeerRestClient::parse_topology_host("[2001:db8::2]", "http://[2001:db8::2]")
+            .expect("scheme-default IPv6 topology host should parse without DNS");
+        assert_eq!(default_port_ipv6.to_string(), "[2001:db8::2]");
+        assert_eq!(default_port_ipv6.port, 80);
+        assert!(!default_port_ipv6.is_port_set);
+
+        assert!(PeerRestClient::parse_topology_host("peer.invalid:0", "http://peer.invalid:0").is_err());
+        assert!(PeerRestClient::parse_topology_host("peer-a.invalid:9000", "http://peer-b.invalid:9000").is_err());
+        assert!(PeerRestClient::parse_topology_host("peer.invalid:9000", "http://peer.invalid:9000/unexpected").is_err());
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn unresolved_default_port_endpoint_topology_retains_all_peer_clients() {
+        let volumes = (0..4)
+            .map(|index| format!("http://rustfs-{index}.invalid:80/data{index}"))
+            .collect::<Vec<_>>();
+        let layout = DisksLayout::from_volumes(&volumes).expect("distributed default-port topology should parse");
+
+        async_with_vars(
+            [
+                (ENV_STARTUP_TOPOLOGY_WAIT_MODE, Some("orchestrated")),
+                (ENV_LOCAL_ENDPOINT_HOST, Some("rustfs-0.invalid")),
+                (ENV_KUBERNETES_SERVICE_HOST, None),
+            ],
+            async {
+                let (server_pools, setup_type) = EndpointServerPools::create_server_endpoints("0.0.0.0:80", &layout)
+                    .await
+                    .expect("explicit local identity should avoid peer DNS during endpoint construction");
+                assert_eq!(setup_type, SetupType::DistErasure);
+
+                let (remote, all, remote_topology_hosts) =
+                    PeerRestClient::build_clients_from_slots(server_pools.peer_grid_host_slots_sorted());
+                assert_eq!(remote.len(), 3);
+                assert!(
+                    remote.iter().all(Option::is_some),
+                    "unresolved remote peers must retain reconnectable clients"
+                );
+                assert_eq!(all.len(), 4);
+                assert_eq!(all.iter().filter(|client| client.is_none()).count(), 1);
+                assert_eq!(remote_topology_hosts.len(), 3);
+                assert!(
+                    remote_topology_hosts.iter().all(|host| !host.contains(':')),
+                    "scheme-default ports must preserve the legacy topology identity"
+                );
+                assert!(
+                    remote
+                        .iter()
+                        .flatten()
+                        .all(|client| client.host.port == 80 && !client.host.is_port_set),
+                    "scheme-default peers must retain the effective dial port"
+                );
+            },
+        )
+        .await;
     }
 
     #[test]
@@ -2740,6 +2889,31 @@ mod tests {
         assert!(!client.offline.load(Ordering::Acquire));
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn dropped_peer_client_releases_and_stops_its_recovery_monitor() {
+        let client = test_peer_client();
+        client.offline.store(true, Ordering::Release);
+        client.recovery_running.store(true, Ordering::Release);
+        let offline = Arc::downgrade(&client.offline);
+        let recovery_running = Arc::downgrade(&client.recovery_running);
+        let handle = PeerRestClient::spawn_recovery_monitor(client.grid_host.clone(), offline.clone(), recovery_running.clone());
+        let started = tokio::time::Instant::now();
+
+        drop(client);
+
+        assert!(offline.upgrade().is_none(), "detached recovery must not retain offline state");
+        assert!(
+            recovery_running.upgrade().is_none(),
+            "detached recovery must not retain its running state"
+        );
+        handle.await.expect("recovery monitor should not panic");
+        assert_eq!(
+            tokio::time::Instant::now(),
+            started,
+            "recovery monitor should stop before advancing to its first delayed probe"
+        );
+    }
+
     #[tokio::test]
     async fn peer_rest_client_finalize_result_keeps_online_for_app_errors_mentioning_unavailable() {
         // Regression: application error text containing "unavailable" (a

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

@@ -4814,9 +4814,11 @@ mod tests {
     async fn test_remote_disk_endpoints_with_different_schemes() {
         let test_cases = vec![
             ("http://server:9000", "server:9000"),
-            ("https://secure-server:443", "secure-server"), // Default HTTPS port is omitted
+            ("http://plain-server:80", "plain-server"),
+            ("http://plain-server", "plain-server"),
+            ("https://secure-server:443", "secure-server"),
             ("http://192.168.1.100:8080", "192.168.1.100:8080"),
-            ("https://secure-server", "secure-server"), // No port specified
+            ("https://secure-server", "secure-server"),
         ];
 
         for (url_str, expected_hostname) in test_cases {

crates/ecstore/src/core/sets.rs

@@ -37,7 +37,9 @@ use crate::{
     runtime::instance::{InstanceContext, bootstrap_ctx},
     runtime::sources as runtime_sources,
     set_disk::{PreparedGetObjectMetadata, SetDisks},
-    store::init_format::{check_format_erasure_values, get_format_erasure_in_quorum, load_format_erasure_all, save_format_file},
+    store::init_format::{
+        check_format_erasure_values, load_format_erasure_all, save_format_file, select_format_erasure_in_quorum,
+    },
 };
 use futures::{
     future::join_all,
@@ -947,7 +949,7 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
 
     #[tracing::instrument(skip(self))]
     async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
-        let (disks, _) = init_storage_disks_with_errors(
+        let (disks, init_errs) = init_storage_disks_with_errors(
             &self.endpoints.endpoints,
             &DiskOption {
                 cleanup: false,
@@ -955,15 +957,36 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
             },
         )
         .await;
-        let (formats, errs) = load_format_erasure_all(&disks, true).await;
+        let (formats, mut errs) = load_format_erasure_all(&disks, true).await;
+        for (err, init_err) in errs.iter_mut().zip(init_errs) {
+            if init_err.is_some() {
+                *err = init_err;
+            }
+        }
+        if errs.iter().any(|err| {
+            matches!(
+                err,
+                Some(DiskError::InconsistentDisk | DiskError::CorruptedFormat | DiskError::CorruptedBackend)
+            )
+        }) {
+            return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat)));
+        }
         if let Err(err) = check_format_erasure_values(&formats, self.set_drive_count) {
             info!("failed to check formats erasure values: {}", err);
             return Ok((HealResultItem::default(), Some(err)));
         }
-        let ref_format = match get_format_erasure_in_quorum(&formats) {
-            Ok(format) => format,
+        let (ref_format, quorum_members) = match select_format_erasure_in_quorum(&formats, 0) {
+            Ok((format, members)) if format.shared_identity() == self.format.shared_identity() => (format, members),
+            Ok(_) => return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat))),
             Err(err) => return Ok((HealResultItem::default(), Some(err))),
         };
+        if formats
+            .iter()
+            .zip(quorum_members)
+            .any(|(format, member)| format.is_some() && !member)
+        {
+            return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat)));
+        }
         let mut res = HealResultItem {
             heal_item_type: HealItemType::Metadata.to_string(),
             detail: "disk-format".to_string(),
@@ -985,11 +1008,6 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
             return Ok((res, Some(StorageError::NoHealRequired)));
         }
 
-        // if !self.format.eq(&ref_format) {
-        //     info!("format ({:?}) not eq ref_format ({:?})", self.format, ref_format);
-        //     return Ok((res, Some(Error::new(DiskError::CorruptedFormat))));
-        // }
-
         let (new_format_sets, _) = new_heal_format_sets(&ref_format, self.set_count, self.set_drive_count, &formats, &errs);
         if !dry_run {
             let mut tmp_new_formats = vec![None; self.set_count * self.set_drive_count];
@@ -1298,7 +1316,7 @@ mod tests {
         assert_eq!(result, (Some(3), Some(1), Some(0)));
     }
 
-    async fn multipart_listing_test_sets() -> (Vec<tempfile::TempDir>, Arc<Sets>) {
+    async fn two_set_test_sets() -> (Vec<tempfile::TempDir>, Arc<Sets>) {
         let format = FormatV3::new(2, 2);
         let mut temp_dirs = Vec::new();
         let mut all_endpoints = Vec::new();
@@ -1339,8 +1357,8 @@ mod tests {
                     Arc::new(RwLock::new(disks)),
                     2,
                     1,
-                    0,
                     set_index,
+                    0,
                     endpoints,
                     format.clone(),
                     vec![Arc::new(LocalClient::new()), Arc::new(LocalClient::new())],
@@ -1373,11 +1391,114 @@ mod tests {
         (temp_dirs, sets)
     }
 
+    #[tokio::test]
+    async fn set_format_heal_accepts_quorum_from_a_nonzero_set() {
+        let (_temp_dirs, sets) = two_set_test_sets().await;
+
+        let (result, err) = sets.disk_set[1]
+            .heal_format(false)
+            .await
+            .expect("the second erasure set should load its own format quorum");
+
+        assert!(matches!(err, Some(StorageError::NoHealRequired)), "unexpected heal result: {err:?}");
+        assert_eq!(result.disk_count, 2);
+        assert_eq!(result.set_count, 1);
+    }
+
+    #[tokio::test]
+    async fn format_heal_rejects_foreign_majorities_at_set_and_pool_scopes() {
+        let (_temp_dirs, _canonical_format, sets) = setup_heal_format_sets(2, true).await;
+        let set_disks = set_level_heal_view(&sets).await;
+
+        let (_, set_err) = set_disks
+            .heal_format(false)
+            .await
+            .expect("set format heal should report a typed mismatch");
+        assert!(
+            matches!(set_err, Some(StorageError::CorruptedFormat)),
+            "foreign set majority must not replace the cached format: {set_err:?}"
+        );
+
+        let (_, pool_err) = sets
+            .heal_format(false)
+            .await
+            .expect("pool format heal should report a typed mismatch");
+        assert!(
+            matches!(pool_err, Some(StorageError::CorruptedFormat)),
+            "foreign pool majority must not replace the cached format: {pool_err:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn pool_format_heal_rejects_a_wrong_slot_minority() {
+        let (_temp_dirs, canonical_format, sets) = setup_heal_format_sets(3, false).await;
+        let mut poisoned_format = canonical_format.clone();
+        poisoned_format.erasure.this = canonical_format.erasure.sets[0][0];
+        replace_heal_test_format(&sets, 2, &poisoned_format).await;
+        let probe_err = new_disk(
+            &sets.endpoints.endpoints.as_ref()[2],
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect_err("a wrong-slot local format must fail disk initialization");
+        assert_eq!(probe_err, DiskError::InconsistentDisk);
+
+        let (_, pool_err) = sets
+            .heal_format(false)
+            .await
+            .expect("pool format heal should report a typed slot mismatch");
+        assert!(
+            matches!(pool_err, Some(StorageError::CorruptedFormat)),
+            "a wrong-slot minority must not be reported as no-heal-required: {pool_err:?}"
+        );
+        assert_eq!(
+            read_heal_test_format(&sets, 2).await,
+            poisoned_format,
+            "format heal must not overwrite a wrong-slot disk"
+        );
+    }
+
+    #[tokio::test]
+    async fn format_heal_rejects_a_foreign_minority_at_set_and_pool_scopes() {
+        let (_temp_dirs, canonical_format, sets) = setup_heal_format_sets(3, false).await;
+        let mut poisoned_format = canonical_format.clone();
+        poisoned_format.id = Uuid::new_v4();
+        poisoned_format.erasure.this = poisoned_format.erasure.sets[0][2];
+        replace_heal_test_format(&sets, 2, &poisoned_format).await;
+        let set_disks = set_level_heal_view(&sets).await;
+
+        let (_, set_err) = set_disks
+            .heal_format(false)
+            .await
+            .expect("set format heal should report a typed identity mismatch");
+        assert!(
+            matches!(set_err, Some(StorageError::CorruptedFormat)),
+            "a foreign minority must not be reported as no-heal-required: {set_err:?}"
+        );
+
+        let (_, pool_err) = sets
+            .heal_format(false)
+            .await
+            .expect("pool format heal should report a typed identity mismatch");
+        assert!(
+            matches!(pool_err, Some(StorageError::CorruptedFormat)),
+            "a foreign minority must not be reported as no-heal-required: {pool_err:?}"
+        );
+        assert_eq!(
+            read_heal_test_format(&sets, 2).await,
+            poisoned_format,
+            "format heal must not overwrite a foreign disk"
+        );
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn list_multipart_uploads_merges_all_sets_without_pagination_loss() {
         let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::Erasure).await;
-        let (_temp_dirs, sets) = multipart_listing_test_sets().await;
+        let (_temp_dirs, sets) = two_set_test_sets().await;
         let bucket = format!("multipart-list-{}", Uuid::new_v4().simple());
         sets.make_bucket(&bucket, &MakeBucketOptions::default())
             .await
@@ -1616,11 +1737,15 @@ mod tests {
     // formatting the first `num_formatted` of them against a shared reference
     // format and leaving the rest unformatted. Returns the live TempDir handles
     // (must be kept alive), the reference format, and the assembled `Sets`.
-    // `disk_set` is intentionally empty: these tests only drive `heal_format`
-    // with `dry_run == true`, which never touches `disk_set`.
-    async fn setup_heal_format_sets(num_formatted: usize) -> (Vec<tempfile::TempDir>, FormatV3, Sets) {
+    // `disk_set` is intentionally empty: these tests only exercise paths that
+    // return before pool-level healing delegates into a set.
+    async fn setup_heal_format_sets(num_formatted: usize, foreign_identity: bool) -> (Vec<tempfile::TempDir>, FormatV3, Sets) {
         const SET_DRIVE_COUNT: usize = 3;
         let ref_format = FormatV3::new(1, SET_DRIVE_COUNT);
+        let mut stored_format = ref_format.clone();
+        if foreign_identity {
+            stored_format.id = Uuid::new_v4();
+        }
 
         let mut dirs = Vec::with_capacity(SET_DRIVE_COUNT);
         let mut endpoints = Vec::with_capacity(SET_DRIVE_COUNT);
@@ -1645,8 +1770,8 @@ mod tests {
             )
             .await
             .expect("disk should be created");
-            let mut disk_format = ref_format.clone();
-            disk_format.erasure.this = ref_format.erasure.sets[0][i];
+            let mut disk_format = stored_format.clone();
+            disk_format.erasure.this = stored_format.erasure.sets[0][i];
             save_format_file(&Some(disk), &Some(disk_format))
                 .await
                 .expect("format should be saved");
@@ -1677,6 +1802,60 @@ mod tests {
         (dirs, ref_format, sets)
     }
 
+    async fn set_level_heal_view(sets: &Sets) -> Arc<SetDisks> {
+        let endpoints = sets.endpoints.endpoints.as_ref().clone();
+        let mut disks = Vec::with_capacity(endpoints.len());
+        for endpoint in &endpoints {
+            disks.push(Some(
+                new_disk(
+                    endpoint,
+                    &DiskOption {
+                        cleanup: false,
+                        health_check: false,
+                    },
+                )
+                .await
+                .expect("fresh set-level disk handle should open"),
+            ));
+        }
+
+        SetDisks::new(
+            "test-owner".to_string(),
+            Arc::new(RwLock::new(disks)),
+            endpoints.len(),
+            1,
+            0,
+            0,
+            endpoints,
+            sets.format.clone(),
+            Vec::new(),
+        )
+        .await
+    }
+
+    async fn replace_heal_test_format(sets: &Sets, disk_index: usize, format: &FormatV3) {
+        let disk = new_disk(
+            &sets.endpoints.endpoints.as_ref()[disk_index],
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("heal test disk should open");
+        save_format_file(&Some(disk.clone()), &Some(format.clone()))
+            .await
+            .expect("poisoned test format should be written");
+    }
+
+    async fn read_heal_test_format(sets: &Sets, disk_index: usize) -> FormatV3 {
+        let path = std::path::Path::new(&sets.endpoints.endpoints.as_ref()[disk_index].get_file_path())
+            .join(crate::disk::RUSTFS_META_BUCKET)
+            .join(crate::disk::FORMAT_CONFIG_FILE);
+        let data = tokio::fs::read(path).await.expect("test format should be readable");
+        FormatV3::try_from(data.as_slice()).expect("test format should parse")
+    }
+
     // Regression for #956 (NoHealRequired path): with every disk already
     // formatted, `heal_format` reports exactly one drive record per disk
     // (N = set_count * set_drive_count), each carrying a real endpoint. Before
@@ -1685,7 +1864,7 @@ mod tests {
     #[tokio::test]
     #[serial]
     async fn heal_format_no_heal_required_reports_one_record_per_disk() {
-        let (_dirs, _ref_format, sets) = setup_heal_format_sets(3).await;
+        let (_dirs, _ref_format, sets) = setup_heal_format_sets(3, false).await;
 
         let (res, err) = sets.heal_format(true).await.expect("heal_format should succeed");
         // All disks formatted -> NoHealRequired early return, still returns `res`.
@@ -1715,7 +1894,7 @@ mod tests {
     #[serial]
     async fn heal_format_heal_path_reports_one_record_per_disk_aligned() {
         // Disks 0 and 1 formatted (quorum), disk 2 unformatted.
-        let (_dirs, _ref_format, sets) = setup_heal_format_sets(2).await;
+        let (_dirs, _ref_format, sets) = setup_heal_format_sets(2, false).await;
 
         let (res, err) = sets.heal_format(true).await.expect("heal_format should succeed");
         // Unformatted disk present -> heal path, not NoHealRequired.

crates/ecstore/src/layout/disks_layout.rs

@@ -203,7 +203,7 @@ fn get_all_sets<T: AsRef<str>>(set_drive_count: usize, is_ellipses: bool, args:
     for args in set_args.iter() {
         for arg in args {
             if unique_args.contains(arg) {
-                return Err(Error::other(format!("Input args {arg} has duplicate ellipses")));
+                return Err(Error::other("input arguments contain a duplicate endpoint after ellipsis expansion"));
             }
             unique_args.insert(arg);
         }
@@ -924,4 +924,15 @@ mod test {
             }
         }
     }
+
+    #[test]
+    fn layout_errors_do_not_echo_url_credentials() {
+        for volumes in [
+            vec!["http://:duplicate-secret@server/path", "http://:duplicate-secret@server/path"],
+            vec!["http://:ellipsis...secret@server/path"],
+        ] {
+            let err = DisksLayout::from_volumes(&volumes).unwrap_err();
+            assert!(!err.to_string().contains("secret"), "layout error leaked endpoint credentials: {err}");
+        }
+    }
 }

crates/ecstore/src/layout/endpoint.rs

@@ -88,6 +88,7 @@ impl TryFrom<&str> for Endpoint {
                 // - All field should be empty except Host and Path.
                 if !((url.scheme() == "http" || url.scheme() == "https")
                     && url.username().is_empty()
+                    && url.password().is_none()
                     && url.fragment().is_none()
                     && url.query().is_none())
                 {
@@ -366,6 +367,12 @@ mod test {
                 expected_type: None,
                 expected_err: Some(Error::other("invalid URL endpoint format")),
             },
+            TestCase {
+                arg: "http://:topsecret@server/path",
+                expected_endpoint: None,
+                expected_type: None,
+                expected_err: Some(Error::other("invalid URL endpoint format")),
+            },
             TestCase {
                 arg: "http://:/path",
                 expected_endpoint: None,
@@ -505,8 +512,18 @@ mod test {
         let endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap();
         assert_eq!(endpoint.host_port(), "example.com:9000");
 
-        let endpoint_no_port = Endpoint::try_from("https://example.com/path").unwrap();
-        assert_eq!(endpoint_no_port.host_port(), "example.com");
+        for endpoint in [
+            Endpoint::try_from("http://example.com/path").unwrap(),
+            Endpoint::try_from("http://example.com:80/path").unwrap(),
+        ] {
+            assert_eq!(endpoint.host_port(), "example.com");
+        }
+        for endpoint in [
+            Endpoint::try_from("https://example.com/path").unwrap(),
+            Endpoint::try_from("https://example.com:443/path").unwrap(),
+        ] {
+            assert_eq!(endpoint.host_port(), "example.com");
+        }
 
         let file_endpoint = Endpoint::try_from("/tmp/data").unwrap();
         assert_eq!(file_endpoint.host_port(), "");

crates/ecstore/src/layout/format.rs

@@ -18,7 +18,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::Error as JsonError;
 use uuid::Uuid;
 
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
 pub enum FormatMetaVersion {
     #[serde(rename = "1")]
     V1,
@@ -27,7 +27,7 @@ pub enum FormatMetaVersion {
     Unknown,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
 pub enum FormatBackend {
     #[serde(rename = "xl")]
     Erasure,
@@ -64,7 +64,7 @@ pub struct FormatErasureV3 {
     pub distribution_algo: DistributionAlgoVersion,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
 pub enum FormatErasureVersion {
     #[serde(rename = "1")]
     V1,
@@ -77,7 +77,7 @@ pub enum FormatErasureVersion {
     Unknown,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
 pub enum DistributionAlgoVersion {
     #[serde(rename = "CRCMOD")]
     V1,
@@ -121,6 +121,15 @@ pub struct FormatV3 {
     pub disk_info: Option<DiskInfo>,
 }
 
+pub(crate) type SharedFormatIdentity<'a> = (
+    &'a FormatMetaVersion,
+    &'a FormatBackend,
+    &'a Uuid,
+    &'a FormatErasureVersion,
+    &'a [Vec<Uuid>],
+    &'a DistributionAlgoVersion,
+);
+
 impl TryFrom<&[u8]> for FormatV3 {
     type Error = JsonError;
 
@@ -198,52 +207,24 @@ impl FormatV3 {
     }
 
     pub fn check_other(&self, other: &FormatV3) -> Result<()> {
-        let mut tmp = other.clone();
-        let this = tmp.erasure.this;
-        tmp.erasure.this = Uuid::nil();
-
-        if self.erasure.sets.len() != other.erasure.sets.len() {
-            return Err(Error::other(format!(
-                "Expected number of sets {}, got {}",
-                self.erasure.sets.len(),
-                other.erasure.sets.len()
-            )));
+        if self.shared_identity() != other.shared_identity() {
+            return Err(Error::other("storage formats do not match"));
         }
 
-        for i in 0..self.erasure.sets.len() {
-            if self.erasure.sets[i].len() != other.erasure.sets[i].len() {
-                return Err(Error::other(format!(
-                    "Each set should be of same size, expected {}, got {}",
-                    self.erasure.sets[i].len(),
-                    other.erasure.sets[i].len()
-                )));
-            }
+        self.find_disk_index_by_disk_id(other.erasure.this).map(|_| ())
+    }
 
-            for j in 0..self.erasure.sets[i].len() {
-                if self.erasure.sets[i][j] != other.erasure.sets[i][j] {
-                    return Err(Error::other(format!(
-                        "UUID on positions {}:{} do not match with, expected {:?} got {:?}: (%w)",
-                        i,
-                        j,
-                        self.erasure.sets[i][j].to_string(),
-                        other.erasure.sets[i][j].to_string(),
-                    )));
-                }
-            }
-        }
-
-        for i in 0..tmp.erasure.sets.len() {
-            for j in 0..tmp.erasure.sets[i].len() {
-                if this == tmp.erasure.sets[i][j] {
-                    return Ok(());
-                }
-            }
-        }
-
-        Err(Error::other(format!(
-            "DriveID {:?} not found in any drive sets {:?}",
-            this, other.erasure.sets
-        )))
+    /// Fields that must agree across every disk in one erasure format,
+    /// excluding the disk-specific `this` UUID and runtime-only `disk_info`.
+    pub(crate) fn shared_identity(&self) -> SharedFormatIdentity<'_> {
+        (
+            &self.version,
+            &self.format,
+            &self.id,
+            &self.erasure.version,
+            &self.erasure.sets,
+            &self.erasure.distribution_algo,
+        )
     }
 }
 
@@ -437,6 +418,30 @@ mod test {
         assert!(result.is_ok());
     }
 
+    #[test]
+    fn test_check_other_rejects_shared_identity_mismatches() {
+        type FormatMutation = (&'static str, fn(&mut FormatV3));
+
+        let format = FormatV3::new(1, 2);
+        let mutations: [FormatMutation; 5] = [
+            ("meta version", |other| other.version = FormatMetaVersion::Unknown),
+            ("backend", |other| other.format = FormatBackend::ErasureSingle),
+            ("deployment id", |other| other.id = Uuid::new_v4()),
+            ("erasure version", |other| other.erasure.version = FormatErasureVersion::V2),
+            ("distribution algorithm", |other| {
+                other.erasure.distribution_algo = DistributionAlgoVersion::V2
+            }),
+        ];
+
+        for (field, mutate) in mutations {
+            let mut other = format.clone();
+            other.erasure.this = format.erasure.sets[0][0];
+            mutate(&mut other);
+
+            assert!(format.check_other(&other).is_err(), "{field} mismatch must be rejected");
+        }
+    }
+
     #[test]
     fn test_check_other_different_set_count() {
         let format1 = FormatV3::new(2, 4);

crates/ecstore/src/services/notification_sys.rs

@@ -2044,15 +2044,10 @@ fn synthesized_disks(host: &str, endpoints: &EndpointServerPools, state: ItemSta
 /// Whether `peer_host` refers to the same node as an endpoint whose
 /// `host_port()` is `ep_host_port`.
 ///
-/// `PeerRestClient::host` is an `XHost`, which resolves names to an address on
-/// construction (`hosts_sorted` -> `XHost::try_from` -> `to_socket_addrs`), so
-/// `peer_host` is the resolved `IP:port`. An endpoint's `host_port()`, however,
-/// is `url.host():port` — still the raw `hostname:port` on hostname-based
-/// deployments. A plain string compare therefore misses on hostname clusters,
-/// leaving the synthesized/degraded drive list empty and `unknownDisks` at 0
-/// (rustfs/rustfs#4607 follow-up). Compare directly first (fast path / IP
-/// deployments), then canonicalize the endpoint side through the same `XHost`
-/// resolution and compare again.
+/// Current topology clients preserve the endpoint `hostname:port`, so the
+/// direct comparison is the normal path. The resolution fallback keeps
+/// compatibility with older or manually constructed clients whose `XHost`
+/// contains a resolved `IP:port` (rustfs/rustfs#4607 follow-up).
 fn endpoint_host_matches(peer_host: &str, ep_host_port: &str) -> bool {
     if peer_host == ep_host_port {
         return true;

crates/ecstore/src/services/tier/tier.rs

@@ -72,6 +72,7 @@ use crate::{
     cluster::rpc::peer_rest_client::{PeerRestClient, PeerTierMutationState},
     config::com::{CONFIG_PREFIX, read_config, read_config_with_metadata},
     disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET},
+    layout::endpoints::EndpointServerPools,
     object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader},
     runtime::sources as runtime_sources,
     set_disk::get_lock_acquire_timeout,
@@ -904,14 +905,17 @@ async fn remote_tier_mutation_peers() -> io::Result<Vec<Arc<dyn TierMutationPeer
     let Some(endpoints) = runtime_sources::endpoint_pools() else {
         return Err(tier_mutation_replay_error("cluster endpoint topology is not initialized"));
     };
-    let remote_host_count = endpoints.hosts_sorted().iter().flatten().count();
-    let (peers, _) = PeerRestClient::new_clients(endpoints).await;
+    remote_tier_mutation_peers_from_topology(endpoints).await
+}
+
+async fn remote_tier_mutation_peers_from_topology(endpoints: EndpointServerPools) -> io::Result<Vec<Arc<dyn TierMutationPeer>>> {
+    let (peers, _, remote_topology_hosts) = PeerRestClient::new_clients_with_topology(endpoints).await;
     let peers = peers
         .into_iter()
         .flatten()
         .map(|peer| Arc::new(peer) as Arc<dyn TierMutationPeer>)
         .collect::<Vec<_>>();
-    ensure_complete_tier_mutation_commit_peer_set(peers.len(), remote_host_count)?;
+    ensure_complete_tier_mutation_commit_peer_set(peers.len(), remote_topology_hosts.len())?;
     Ok(peers)
 }
 
@@ -4307,6 +4311,34 @@ fn tier_config_not_initialized_error(operation: &str) -> std::io::Error {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::layout::{
+        endpoint::Endpoint,
+        endpoints::{Endpoints, PoolEndpoints, SetupType},
+    };
+
+    struct SetupTypeGuard {
+        previous: SetupType,
+    }
+
+    impl SetupTypeGuard {
+        async fn switch_to(next: SetupType) -> Self {
+            let previous = runtime_sources::current_setup_type().await;
+            runtime_sources::set_setup_type(next).await;
+            Self { previous }
+        }
+    }
+
+    impl Drop for SetupTypeGuard {
+        fn drop(&mut self) {
+            let previous = self.previous.clone();
+            let handle = tokio::runtime::Handle::current();
+            tokio::task::block_in_place(|| {
+                handle.block_on(async move {
+                    runtime_sources::set_setup_type(previous).await;
+                });
+            });
+        }
+    }
 
     fn build_s3_tier(name: &str) -> TierConfig {
         TierConfig {
@@ -6354,6 +6386,42 @@ mod tests {
         assert!(err.to_string().contains("without peer commit clients"), "{err}");
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial_test::serial]
+    async fn tier_mutation_peer_composition_preserves_unresolved_topology_slots() {
+        let mut endpoints = Vec::new();
+        for disk_index in 0..4 {
+            let mut endpoint = Endpoint::try_from(format!("http://rustfs-{disk_index}.invalid:9000/data{disk_index}").as_str())
+                .expect("unresolved topology endpoint should parse without DNS");
+            endpoint.is_local = disk_index == 0;
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(disk_index);
+            endpoints.push(endpoint);
+        }
+        let topology = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 4,
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: "unresolved-tier-mutation-topology".to_string(),
+            platform: "test".to_string(),
+        }]);
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+
+        let peers = remote_tier_mutation_peers_from_topology(topology)
+            .await
+            .expect("every unresolved remote topology slot should retain a tier mutation client");
+        assert_eq!(
+            peers.iter().map(|peer| peer.peer_label()).collect::<Vec<_>>(),
+            vec![
+                "http://rustfs-1.invalid:9000".to_string(),
+                "http://rustfs-2.invalid:9000".to_string(),
+                "http://rustfs-3.invalid:9000".to_string(),
+            ]
+        );
+    }
+
     #[tokio::test]
     async fn coordinator_fanout_prepare_failure_aborts_prepared_peers_without_cas() {
         let manager = TierConfigMgr::new();

crates/ecstore/src/set_disk/mod.rs

@@ -110,7 +110,10 @@ use crate::{
     object_api::{GetObjectReader, ObjectInfo, PutObjReader},
     // event::name::EventName,
     services::event_notification::{EventArgs, send_event},
-    store::init_format::{get_format_erasure_in_quorum, load_format_erasure, load_format_erasure_all, save_format_file},
+    store::init_format::{
+        formats_match_reference_slots, get_format_erasure_in_quorum, load_format_erasure, load_format_erasure_all,
+        save_format_file,
+    },
 };
 use bytes::Bytes;
 use bytesize::ByteSize;

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

@@ -1373,11 +1373,24 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
     async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
         let disks = self.disks.read().await.clone();
         let (formats, errs) = load_format_erasure_all(&disks, true).await;
-        let ref_format = match get_format_erasure_in_quorum(&formats) {
-            Ok(format) => format,
+        if errs.iter().any(|err| {
+            matches!(
+                err,
+                Some(DiskError::InconsistentDisk | DiskError::CorruptedFormat | DiskError::CorruptedBackend)
+            )
+        }) {
+            return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat)));
+        }
+        let slot_offset = self
+            .set_index
+            .checked_mul(self.set_drive_count)
+            .ok_or_else(|| Error::other("erasure set slot offset overflow"))?;
+        let ref_format = match get_format_erasure_in_quorum(&formats, slot_offset) {
+            Ok(format) if format.shared_identity() == self.format.shared_identity() => format,
+            Ok(_) => return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat))),
             Err(err) => {
                 let can_use_cached_layout = count_errs(&errs, &DiskError::UnformattedDisk) > 0
-                    && formats.iter().flatten().all(|format| self.format.check_other(format).is_ok())
+                    && formats_match_reference_slots(&formats, &self.format, slot_offset)
                     && errs
                         .iter()
                         .all(|err| err.is_none() || matches!(err, Some(DiskError::UnformattedDisk)));
@@ -1388,6 +1401,9 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
                 }
             }
         };
+        if !formats_match_reference_slots(&formats, &ref_format, slot_offset) {
+            return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat)));
+        }
 
         let endpoints = crate::layout::endpoints::Endpoints::from(self.set_endpoints.clone());
         let before_drives = crate::layout::set_heal::formats_to_drives_info(&endpoints, &formats, &errs);
@@ -1543,11 +1559,16 @@ mod heal_result_report_tests {
     use crate::disk::error::DiskError;
     use crate::disk::format::FormatV3;
     use crate::disk::{DiskAPI as _, DiskOption, DiskStore, RUSTFS_META_TMP_BUCKET, ReadOptions, new_disk};
+    use crate::error::Error;
     use crate::object_api::{ObjectOptions, PutObjReader};
     use crate::set_disk::ops::object::hermetic_set_disks_support::hermetic_set_disks_isolated;
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+    use crate::storage_api_contracts::heal::HealOperations as _;
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
-    use crate::{config::storageclass, store::init_format::save_format_file};
+    use crate::{
+        config::storageclass,
+        store::init_format::{load_format_erasure, save_format_file},
+    };
     use rustfs_common::heal_channel::{DriveState, HealOpts, HealScanMode};
     use rustfs_filemeta::{BLOCK_SIZE_V2, FileInfo, ObjectPartInfo, TRANSITION_COMPLETE};
     use std::sync::Arc;
@@ -1863,6 +1884,44 @@ mod heal_result_report_tests {
         }
     }
 
+    #[tokio::test]
+    async fn format_heal_cached_layout_rejects_a_disk_from_another_slot() {
+        let mut _temp_dirs = Vec::new();
+        let mut endpoints = Vec::new();
+        let mut disks = Vec::new();
+        for disk_index in 0..3 {
+            let (temp_dir, mut endpoint, disk) = real_disk().await;
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(disk_index);
+            _temp_dirs.push(temp_dir);
+            endpoints.push(endpoint);
+            disks.push(Some(disk));
+        }
+        let set = set_disks_with(disks.clone(), endpoints, 1).await;
+        let mut wrong_slot = set.format.clone();
+        wrong_slot.erasure.this = set.format.erasure.sets[0][1];
+        save_format_file(&disks[0], &Some(wrong_slot))
+            .await
+            .expect("wrong-slot format fixture should be saved");
+        let mut correct_slot = set.format.clone();
+        correct_slot.erasure.this = set.format.erasure.sets[0][2];
+        save_format_file(&disks[2], &Some(correct_slot))
+            .await
+            .expect("correct format fixture should be saved");
+
+        let (_, heal_err) = set
+            .heal_format(false)
+            .await
+            .expect("format heal should report the quorum failure in its result");
+
+        assert!(matches!(heal_err, Some(Error::CorruptedFormat)));
+        let unformatted = load_format_erasure(disks[1].as_ref().expect("second disk should be online"), true)
+            .await
+            .expect_err("a rejected fallback must not format the missing slot");
+        assert_eq!(unformatted, DiskError::UnformattedDisk);
+    }
+
     // Regression for #955: an offline disk must contribute exactly one drive
     // record. Before the fix the offline branch fell through and pushed a second
     // (Corrupt) record for the same disk, so `before/after.drives` grew to

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

@@ -343,20 +343,23 @@ impl SetDisks {
             }
         };
 
-        // The drive's format may place it in a different erasure set than this
-        // one. Claiming a misplaced drive into `self.disks` would let two sets
-        // manage the same drive and degrade together, so reject it here
-        // (backlog#799 B19).
-        if set_idx != self.set_index {
+        // Claiming a misplaced drive into `self.disks` would let two slots or
+        // sets manage the same drive and degrade together (backlog#799 B19).
+        if set_idx != self.set_index || self.set_endpoints.get(disk_idx) != Some(ep) {
             warn!(
-                "renew_disk: drive {:?} belongs to set {} but is being renewed on set {}; skipping",
-                ep, set_idx, self.set_index
+                endpoint = %ep,
+                format_set_index = set_idx,
+                format_disk_index = disk_idx,
+                endpoint_pool_index = ep.pool_idx,
+                endpoint_set_index = ep.set_idx,
+                endpoint_disk_index = ep.disk_idx,
+                expected_pool_index = self.pool_index,
+                expected_set_index = self.set_index,
+                "renew_disk rejected a drive whose endpoint and format do not identify the same topology slot"
             );
             return;
         }
 
-        // Check that the endpoint matches
-
         let _ = new_disk.set_disk_id(Some(fm.erasure.this)).await;
         new_disk.enable_health_check();
 
@@ -715,6 +718,108 @@ mod tests {
         drop(temp_dirs);
     }
 
+    #[tokio::test]
+    async fn renew_disk_rejects_a_format_from_another_slot_or_cluster() {
+        let disk_count = 3;
+        let format = FormatV3::new(1, disk_count);
+        let mut temp_dirs = Vec::with_capacity(disk_count);
+        let mut endpoints = Vec::with_capacity(disk_count);
+        let mut fixture_disks = Vec::with_capacity(disk_count);
+
+        for disk_idx in 0..disk_count {
+            let (temp_dir, endpoint, disk) = make_formatted_local_disk(disk_idx, &format).await;
+            temp_dirs.push(temp_dir);
+            endpoints.push(endpoint);
+            fixture_disks.push(disk);
+        }
+
+        let set_disks = SetDisks::new(
+            "test-owner".to_string(),
+            Arc::new(RwLock::new(vec![Some(fixture_disks[0].clone()), None, None])),
+            disk_count,
+            disk_count / 2,
+            0,
+            0,
+            endpoints.clone(),
+            format.clone(),
+            Vec::new(),
+        )
+        .await;
+
+        let mut other_cluster_format = format.clone();
+        other_cluster_format.id = Uuid::new_v4();
+        other_cluster_format.erasure.this = format.erasure.sets[0][2];
+        save_format_file(&Some(fixture_disks[2].clone()), &Some(other_cluster_format))
+            .await
+            .expect("other-cluster format should be written for the rejection test");
+
+        set_disks.renew_disk(&endpoints[2]).await;
+
+        let disks = set_disks.get_disks_internal().await;
+        assert_eq!(
+            disks[0]
+                .as_ref()
+                .expect("the canonical first slot must remain attached")
+                .endpoint(),
+            endpoints[0]
+        );
+        assert!(
+            disks[2].is_none(),
+            "a disk from another deployment must remain detached even when its slot UUID matches"
+        );
+
+        let mut correct_format = format.clone();
+        correct_format.erasure.this = format.erasure.sets[0][2];
+        let replacement_disk = new_disk(
+            &endpoints[2],
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("third endpoint should reopen after other-cluster rejection");
+        save_format_file(&Some(replacement_disk), &Some(correct_format))
+            .await
+            .expect("correct slot format should be restored");
+
+        set_disks.renew_disk(&endpoints[2]).await;
+
+        let disks = set_disks.get_disks_internal().await;
+        assert_eq!(
+            disks[0]
+                .as_ref()
+                .expect("the canonical first slot must remain attached")
+                .endpoint(),
+            endpoints[0]
+        );
+        assert_eq!(disks[2].as_ref().expect("the restored third slot should attach").endpoint(), endpoints[2]);
+
+        let third_disk = disks[2].clone();
+        let mut wrong_slot_format = format.clone();
+        wrong_slot_format.erasure.this = format.erasure.sets[0][0];
+        save_format_file(&third_disk, &Some(wrong_slot_format))
+            .await
+            .expect("wrong-slot format should be written for the rejection test");
+        set_disks.disks.write().await[2] = None;
+
+        let mut misplaced_endpoint = endpoints[2].clone();
+        misplaced_endpoint.set_disk_index(0);
+        set_disks.renew_disk(&misplaced_endpoint).await;
+
+        let disks = set_disks.get_disks_internal().await;
+        assert_eq!(
+            disks[0]
+                .as_ref()
+                .expect("the canonical first slot must remain attached")
+                .endpoint(),
+            endpoints[0]
+        );
+        assert!(disks[2].is_none(), "a disk claiming another endpoint's slot must remain detached");
+
+        drop(temp_dirs);
+    }
+
     // SetDisks split P0 (#816): the borrow handle must mirror the core state and
     // the List operation family must run identically through it.
     #[tokio::test]

crates/ecstore/src/store/heal.rs

@@ -30,6 +30,7 @@ impl ECStore {
         };
 
         let mut count_no_heal = 0;
+        let mut first_error = None;
         for pool in self.pools.iter() {
             let (mut result, err) = pool.heal_format(dry_run).await?;
             if let Some(err) = err {
@@ -37,8 +38,8 @@ impl ECStore {
                     StorageError::NoHealRequired => {
                         count_no_heal += 1;
                     }
-                    _ => {
-                        continue;
+                    err => {
+                        first_error.get_or_insert(err);
                     }
                 }
             }
@@ -47,6 +48,9 @@ impl ECStore {
             r.before.drives.append(&mut result.before.drives);
             r.after.drives.append(&mut result.after.drives);
         }
+        if let Some(err) = first_error {
+            return Ok((r, Some(err)));
+        }
         if count_no_heal == self.pools.len() {
             info!(
                 event = EVENT_HEAL_FORMAT_COMPLETED,
@@ -165,3 +169,134 @@ impl ECStore {
         Err(StorageError::NotImplemented)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::disk::{DiskOption, format::FormatV3, new_disk};
+    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
+    use crate::store::init_format::{load_format_erasure, save_format_file};
+
+    #[tokio::test]
+    async fn handle_heal_format_continues_after_a_pool_error() {
+        let canonical_format = FormatV3::new(1, 3);
+        let mut foreign_format = canonical_format.clone();
+        foreign_format.id = Uuid::new_v4();
+        let mut temp_dirs = Vec::new();
+        let mut endpoints = Vec::new();
+        let mut disks = Vec::new();
+
+        for disk_index in 0..3 {
+            let temp_dir = tempfile::tempdir().expect("temporary disk root should be created");
+            let mut endpoint = Endpoint::try_from(temp_dir.path().to_str().expect("temporary path should be UTF-8"))
+                .expect("temporary endpoint should parse");
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(disk_index);
+            let disk = new_disk(
+                &endpoint,
+                &DiskOption {
+                    cleanup: false,
+                    health_check: false,
+                },
+            )
+            .await
+            .expect("temporary disk should open");
+            let mut disk_format = foreign_format.clone();
+            disk_format.erasure.this = foreign_format.erasure.sets[0][disk_index];
+            save_format_file(&Some(disk.clone()), &Some(disk_format))
+                .await
+                .expect("foreign format should be written");
+            temp_dirs.push(temp_dir);
+            endpoints.push(endpoint);
+            disks.push(Some(disk));
+        }
+
+        let pool_endpoints = PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 3,
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: "foreign-format-majority-test".to_string(),
+            platform: "test".to_string(),
+        };
+        let pool = Sets::new(disks, &pool_endpoints, &canonical_format, 0, 1)
+            .await
+            .expect("test pool should build around the cached canonical format");
+
+        let mut recoverable_format = FormatV3::new(1, 3);
+        recoverable_format.id = canonical_format.id;
+        let mut recoverable_temp_dirs = Vec::new();
+        let mut recoverable_endpoints = Vec::new();
+        let mut recoverable_disks = Vec::new();
+        let mut unformatted_disk = None;
+        for disk_index in 0..3 {
+            let temp_dir = tempfile::tempdir().expect("temporary disk root should be created");
+            let mut endpoint = Endpoint::try_from(temp_dir.path().to_str().expect("temporary path should be UTF-8"))
+                .expect("temporary endpoint should parse");
+            endpoint.set_pool_index(1);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(disk_index);
+            let disk = new_disk(
+                &endpoint,
+                &DiskOption {
+                    cleanup: false,
+                    health_check: false,
+                },
+            )
+            .await
+            .expect("temporary disk should open");
+            if disk_index < 2 {
+                let mut disk_format = recoverable_format.clone();
+                disk_format.erasure.this = recoverable_format.erasure.sets[0][disk_index];
+                save_format_file(&Some(disk.clone()), &Some(disk_format))
+                    .await
+                    .expect("recoverable format should be written");
+            } else {
+                unformatted_disk = Some(disk.clone());
+            }
+            recoverable_temp_dirs.push(temp_dir);
+            recoverable_endpoints.push(endpoint);
+            recoverable_disks.push(Some(disk));
+        }
+        let recoverable_pool_endpoints = PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 3,
+            endpoints: Endpoints::from(recoverable_endpoints),
+            cmd_line: "recoverable-format-test".to_string(),
+            platform: "test".to_string(),
+        };
+        let recoverable_pool = Sets::new(recoverable_disks, &recoverable_pool_endpoints, &recoverable_format, 1, 1)
+            .await
+            .expect("recoverable test pool should build");
+
+        let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints.clone(), recoverable_pool_endpoints.clone()]);
+        let store = ECStore {
+            id: canonical_format.id,
+            disk_map: HashMap::new(),
+            pools: vec![pool, recoverable_pool],
+            peer_sys: S3PeerSys::new(&endpoint_pools),
+            pool_meta: RwLock::new(PoolMeta::default()),
+            rebalance_meta: RwLock::new(None),
+            decommission_cancelers: RwLock::new(Vec::new()),
+            start_gate: Mutex::new(()),
+            pool_meta_save_gate: Mutex::new(()),
+            ctx: crate::runtime::instance::bootstrap_ctx(),
+        };
+
+        let (result, err) = store
+            .handle_heal_format(false)
+            .await
+            .expect("format heal should return the typed pool error");
+        assert!(
+            matches!(err, Some(StorageError::CorruptedFormat)),
+            "foreign format majority must not be downgraded to a successful heal: {err:?}"
+        );
+        assert_eq!(result.disk_count, 3, "the recoverable pool should still be inspected");
+        let healed = load_format_erasure(&unformatted_disk.expect("the unformatted disk handle should be retained"), true)
+            .await
+            .expect("the later pool should be healed despite the first pool error");
+        assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]);
+    }
+}

crates/ecstore/src/store/init.rs

@@ -101,6 +101,10 @@ fn should_retry_local_decommission_resume(err: &Error, attempt: usize) -> bool {
     matches!(err, Error::ConfigNotFound) && attempt < LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES
 }
 
+fn should_retry_format_load(err: &Error) -> bool {
+    !matches!(err, Error::CorruptedFormat)
+}
+
 fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_meta_loaded: bool) -> bool {
     rebalance_meta_loaded && !decommission_running
 }
@@ -294,7 +298,7 @@ impl ECStore {
             // periodic monitoring until format loading succeeds. Startup RPC
             // failures can still spawn recovery probes for peers that come up
             // after this node.
-            let (disks, errs) = init_format::init_disks(
+            let (mut disks, errs) = init_format::init_disks(
                 &pool_eps.endpoints,
                 &DiskOption {
                     cleanup: true,
@@ -311,7 +315,7 @@ impl ECStore {
                 loop {
                     match init_format::connect_load_init_formats(
                         pool_first_is_local,
-                        &disks,
+                        &mut disks,
                         pool_eps.set_count,
                         pool_eps.drives_per_set,
                         deployment_id,
@@ -319,6 +323,7 @@ impl ECStore {
                     .await
                     {
                         Ok(fm) => break Ok(fm),
+                        Err(e) if !should_retry_format_load(&e) => break Err(e),
                         // Wrap the final error if we are giving up
                         Err(e) if times >= 10 => {
                             break Err(Error::other(format!("store init failed to load formats after {times} retries: {e}")));
@@ -551,7 +556,7 @@ mod tests {
         LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, load_pool_meta_for_startup, pool_first_endpoint_is_local,
         pool_meta_has_active_decommission, preflight_startup_rpc_secret_with, resolve_startup_pool_defaults_with,
         resolve_store_init_stage_result, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init,
-        should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
+        should_retry_format_load, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
     };
     #[cfg(feature = "test-util")]
     use crate::{
@@ -773,6 +778,13 @@ mod tests {
         assert!(!should_retry_local_decommission_resume(&StorageError::SlowDown, 0));
     }
 
+    #[test]
+    fn test_should_retry_format_load_rejects_permanent_corruption() {
+        assert!(!should_retry_format_load(&StorageError::CorruptedFormat));
+        assert!(should_retry_format_load(&StorageError::ErasureReadQuorum));
+        assert!(should_retry_format_load(&StorageError::FirstDiskWait));
+    }
+
     #[test]
     fn test_should_auto_start_rebalance_after_init_allows_loaded_rebalance_without_decommission() {
         assert!(should_auto_start_rebalance_after_init(false, true));

crates/ecstore/src/store/mod.rs

@@ -989,7 +989,7 @@ mod tests {
 
         init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
 
-        let (disks, errs) = init_disks(
+        let (mut disks, errs) = init_disks(
             &endpoint_pools.as_ref().first().expect("pool endpoints").endpoints,
             &DiskOption {
                 cleanup: true,
@@ -999,7 +999,7 @@ mod tests {
         .await;
 
         assert!(errs.iter().all(|err| err.is_none()), "disk init should succeed: {errs:?}");
-        connect_load_init_formats(true, &disks, 1, 4, None)
+        connect_load_init_formats(true, &mut disks, 1, 4, None)
             .await
             .expect("initialize format metadata");
 

crates/utils/src/net.rs

@@ -92,7 +92,6 @@ fn resolve_domain(domain: &str) -> std::io::Result<HashSet<IpAddr>> {
     (domain, 0)
         .to_socket_addrs()
         .map(|v| v.map(|v| v.ip()).collect::<HashSet<_>>())
-        .map_err(Error::other)
 }
 
 #[cfg(test)]
@@ -284,7 +283,7 @@ pub async fn get_host_ip(host: Host<&str>) -> std::io::Result<HashSet<IpAddr>> {
                 }
                 Err(err) => {
                     error!("Failed to resolve domain {domain} using system resolver, err: {err}");
-                    Err(Error::other(err))
+                    Err(err)
                 }
             }
         }
@@ -608,6 +607,27 @@ mod test {
         assert!(get_host_ip(invalid_host).await.is_err());
     }
 
+    #[tokio::test]
+    async fn test_get_host_ip_preserves_resolver_error_provenance() {
+        let _resolver_guard = set_mock_dns_resolver(|_| Err(IoError::from_raw_os_error(-3)));
+
+        let err = get_host_ip(Host::Domain("temporarily-unavailable.example"))
+            .await
+            .unwrap_err();
+
+        assert_eq!(err.raw_os_error(), Some(-3));
+    }
+
+    #[test]
+    fn test_resolve_domain_preserves_system_resolver_error_provenance() {
+        let _resolver_lock = DNS_RESOLVER_TEST_LOCK.lock().unwrap();
+        reset_dns_resolver_inner();
+
+        let err = resolve_domain("rustfs-resolver-provenance.invalid").unwrap_err();
+
+        assert_ne!(err.kind(), std::io::ErrorKind::Other, "system resolver error was wrapped: {err}");
+    }
+
     #[test]
     fn test_dns_cache_entry_expiry() {
         let ips: HashSet<IpAddr> = [IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))].into_iter().collect();

crates/utils/src/string.rs

@@ -360,7 +360,7 @@ pub fn find_ellipses_patterns(arg: &str) -> Result<ArgPattern> {
         Some(caps) => caps,
         None => {
             return Err(Error::other(format!(
-                "Invalid ellipsis format in ({arg}), Ellipsis range must be provided in format {{N...M}} where N and M are decimal or hexadecimal positive integers, M must be greater than N, with a maximum expanded range size of {MAX_ELLIPSES_RANGE_SIZE}"
+                "Invalid ellipsis format. Ellipsis range must be provided in format {{N...M}} where N and M are decimal or hexadecimal positive integers, M must be greater than N, with a maximum expanded range size of {MAX_ELLIPSES_RANGE_SIZE}"
             )));
         }
     };
@@ -399,7 +399,7 @@ pub fn find_ellipses_patterns(arg: &str) -> Result<ArgPattern> {
             || p.suffix.contains(CLOSE_BRACES)
         {
             return Err(Error::other(format!(
-                "Invalid ellipsis format in ({arg}), Ellipsis range must be provided in format {{N...M}} where N and M are decimal or hexadecimal positive integers, M must be greater than N, with a maximum expanded range size of {MAX_ELLIPSES_RANGE_SIZE}"
+                "Invalid ellipsis format. Ellipsis range must be provided in format {{N...M}} where N and M are decimal or hexadecimal positive integers, M must be greater than N, with a maximum expanded range size of {MAX_ELLIPSES_RANGE_SIZE}"
             )));
         }
     }
@@ -938,6 +938,15 @@ mod tests {
         assert!(err.to_string().contains("decimal or hexadecimal"), "unexpected error message: {err}");
     }
 
+    #[test]
+    fn test_find_ellipses_patterns_leftover_brace_error_does_not_echo_input() {
+        let err = find_ellipses_patterns("http://:brace-secret@server/{1...2}}").unwrap_err();
+        assert!(
+            !err.to_string().contains("brace-secret"),
+            "ellipsis error leaked endpoint credentials: {err}"
+        );
+    }
+
     #[test]
     fn test_parse_ellipses_range_rejects_oversized_ranges() {
         let err = parse_ellipses_range("{0...10000}").unwrap_err();

docs/architecture/compat-cleanup-register.md

@@ -12,6 +12,8 @@ for later deletion.
 
 ## Open Items
 
+- `rustfs-5416` Helm distributed startup wait setting: charts that predate explicit local endpoint identity expose startupWaitTimeoutSeconds for their peer DNS/TCP init gate. The new chart keeps the value accepted but ignores it after moving startup convergence into RustFS. Remove the value and its documentation after the minimum supported direct-upgrade chart includes localEndpointHost.autoInject and no longer renders the peer gate.
+- `rustfs-5416-wait-mode` startup wait-mode validation: releases before explicit local endpoint identity treat an unknown RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE value as auto. New servers retain that fallback only when no explicit local endpoint host is configured; an anchor requires a recognized mode so a typo cannot bypass DNS locality. Remove the fallback and reject every unknown value after every supported direct-upgrade chart validates this setting before rollout.
 - `scanner-usage-v2` persisted scanner usage migration: pre-v2 scanners write `.usage.json`, so upgraded clusters read that primary/backup pair only while `.usage.v2.json` is absent and continue removing deleted buckets from legacy copies that still exist. The additive usage_snapshot_complete field in `.usage.v2.json` must remain optional while mixed-version clusters are supported; a missing field means the snapshot is not authoritative. Remove the legacy object fallback and cleanup only after every supported direct-upgrade source writes `.usage.v2.json`.
 - `ns-scanner-rpc-v3` namespace scanner capability and activity handshake: old peers and legacy internode transports lack the authenticated startup-epoch handshake. The oldest peers send an empty activity request and receive a field-empty protocol-0 response. Protocol v4 binds the challenge and response topology but cannot authenticate distributed dirty-usage state. Current protocol v5 binds the request version, acknowledgement target and generation, and the response dirty-usage state. Servers retain protocol-0 and protocol-v4 codecs for rolling upgrades, while the distributed scanner publishes usage only after every peer returns authenticated protocol v5 state. Scanner selection treats HTTP 404/405/426 and the legacy MethodNotAllowed default as an explicit lack of remote scanner v3 support and assigns those disks to coordinator-driven workers; transient capability failures remain incomplete and do not activate the fallback. Remove the coordinator fallback after the minimum supported RustFS peer version implements namespace scanner protocol v3, remove protocol-0 activity requests and responses after every supported peer implements authenticated scanner activity protocol v4, and remove the protocol-v4 activity codec after every supported peer implements protocol v5; future protocol revisions must keep the same dual-version server/codec window before changing the advertised version.
 - `#4648` walk-dir stream completion capability: old clients can append fallback output to an already-used metacache writer after a terminal body error, so servers emit terminal walk errors only to clients that sign the `walk_dir_stream_completion=error-v1` query capability and its request-body digest. Remove the legacy clean-EOF path after the minimum supported RustFS peer version always advertises this capability.

helm/README.md

@@ -47,6 +47,53 @@ without manual intervention:
 If you previously set `replicaCount=16` and now want a different topology,
 set both `replicaCount` and `drivesPerNode` explicitly.
 
+For distributed deployments that use the chart-generated
+`RUSTFS_VOLUMES`, `localEndpointHost.autoInject` defaults to automatic
+selection. Without `secret.existingSecret`, the chart injects a private
+Downward API variable, `RUSTFS_CHART_POD_NAME`, and uses it to build the pod's
+fully qualified hostname in
+`RUSTFS_LOCAL_ENDPOINT_HOST`. RustFS can then identify local drives without
+waiting for every peer's DNS record or TCP listener. A user-defined `POD_NAME`
+is preserved and does not interfere with the private chart variable. Setting
+`RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE` explicitly to `bounded`, `fail-fast`,
+`failfast`, or `strict` also keeps legacy DNS-based locality discovery. A
+dynamically sourced wait mode keeps the legacy path because its value cannot be
+validated while rendering. Otherwise, Kubernetes auto-detection selects
+orchestrated startup when RustFS consumes the generated anchor.
+
+An existing Secret is opaque to the chart and may historically contain more
+than credentials, so the chart does not inject an anchor whenever
+`secret.existingSecret` is set. In Kubernetes auto/orchestrated mode, RustFS
+then derives a DNS-free identity from the kernel hostname when exactly one
+domain endpoint at the server port has the same full hostname or first label.
+All-IP topologies retain direct IP locality detection. A domain topology with
+zero or multiple hostname matches fails before format initialization and must
+set `RUSTFS_LOCAL_ENDPOINT_HOST` explicitly. For a credentials-only Secret, set
+`localEndpointHost.autoInject=true` to add the chart anchor without changing
+the historical ConfigMap-then-Secret `envFrom` precedence. If injection is
+explicitly enabled with an incompatible hidden `RUSTFS_VOLUMES`,
+`RUSTFS_ADDRESS`, or `RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE`, RustFS also fails
+during endpoint construction; it does not silently fall back to a different
+topology.
+
+When `config.rustfs.volumes` is set explicitly, the chart does not infer a
+local endpoint identity. RustFS applies the same kernel-hostname inference to
+custom domain topologies in Kubernetes auto/orchestrated mode; aliases that do
+not match the Pod hostname must provide `RUSTFS_LOCAL_ENDPOINT_HOST` through
+`extraEnv`. An explicit `RUSTFS_VOLUMES`, explicit
+`RUSTFS_LOCAL_ENDPOINT_HOST`, bounded/dynamic or unrecognized startup mode, or
+`localEndpointHost.autoInject=false`, also disables chart injection. A
+`RUSTFS_ADDRESS` override alone does not disable it; the effective address and
+generated topology must agree on the endpoint port. Custom anchor-based
+configurations must resolve to `orchestrated` startup mode and must not receive
+a conflicting mode from an `envFrom` source.
+`startupWaitTimeoutSeconds` is retained for values-file compatibility but is
+deprecated and ignored.
+
+Upgrade the chart and RustFS image together. An older image that does not
+recognize `RUSTFS_LOCAL_ENDPOINT_HOST` retains its previous DNS-based startup
+behavior.
+
 ---
 
 # Parameters Overview
@@ -57,6 +104,7 @@ set both `replicaCount` and `drivesPerNode` explicitly.
 | affinity.podAntiAffinity.enabled | bool | `true` |  |
 | affinity.podAntiAffinity.topologyKey | string | `"kubernetes.io/hostname"` |  |
 | clusterDomain | string | `"cluster.local"` | Kubernetes cluster DNS domain used to build in-cluster FQDNs for `RUSTFS_VOLUMES` (distributed mode) and mTLS server certificate SANs. Override for clusters not using the default `cluster.local`. Provide the DNS root only, without a `svc.` prefix or leading/trailing dots. |
+| localEndpointHost.autoInject | bool or null | `null` | Automatically inject `RUSTFS_LOCAL_ENDPOINT_HOST` for chart-generated distributed topologies unless `secret.existingSecret` is set. Use `true` for a credentials-only existing Secret or `false` to preserve legacy DNS locality explicitly. |
 | commonLabels | object | `{}` | Labels to add to all deployed objects. |
 | config.rustfs.address | string | `":9000"` |  |
 | config.rustfs.console_address | string | `":9001"` |  |
@@ -66,7 +114,7 @@ set both `replicaCount` and `drivesPerNode` explicitly.
 | config.rustfs.obs_environment | string | `"development"` |  |
 | config.rustfs.obs_log_directory | string | `"/logs"` | Log directory inside the RustFS container. Set to `""` to disable log PVCs and mounts. |
 | config.rustfs.region | string | `"us-east-1"` |  |
-| config.rustfs.volumes | string | `""` |  |
+| config.rustfs.volumes | string | `""` | Explicit distributed volume topology. When empty, the chart generates the topology and normally injects `RUSTFS_LOCAL_ENDPOINT_HOST`; custom topologies must configure local endpoint identity explicitly when needed. |
 | config.rustfs.log_rotation.size | int | `"100"` | Default log rotation size mb for rustfs. |
 | config.rustfs.log_rotation.time | string | `"hour"` | Default log rotation time for rustfs. |
 | config.rustfs.log_rotation.keep_files | int | `"30"` | Default log keep files for rustfs.  |
@@ -107,7 +155,7 @@ set both `replicaCount` and `drivesPerNode` explicitly.
 | config.rustfs.kms.vault.vault_token | string | `""`| The vault token. Rendered into a dedicated Secret (`<fullname>-kms-secret`), never into the ConfigMap. |
 | config.rustfs.kms.vault.vault_mount_path | string | `"transit"`| The vault mount path, only works if `vault_backend` equals `vault-transit` . |
 | config.rustfs.kms.vault.default_key | string | `"transit"`| The master key id for RustFS. |
-| extraEnv | map | `[]` |  Extra environment variables for RustFS container. |
+| extraEnv | list | `[]` | Extra environment variables for the RustFS container. An explicit `RUSTFS_LOCAL_ENDPOINT_HOST` or `RUSTFS_VOLUMES`, or a bounded, dynamic, or unrecognized startup mode, disables generated anchor injection. `POD_NAME` and `RUSTFS_ADDRESS` remain independent overrides. |
 | extraVolumes | list | `[]` | Extra volumes to add to the pod spec. Supported in both standalone (Deployment) and distributed (StatefulSet) modes. |
 | extraVolumeMounts | list | `[]` | Extra volume mounts to add to the RustFS container. Supported in both standalone (Deployment) and distributed (StatefulSet) modes. |
 | containerSecurityContext.capabilities.drop[0] | string | `"ALL"` |  |
@@ -190,7 +238,7 @@ uer. `ClusterIssuer` or `Issuer`. |
 | resources.limits.memory | string | `"512Mi"` |  |
 | resources.requests.cpu | string | `"100m"` |  |
 | resources.requests.memory | string | `"128Mi"` |  |
-| secret.existingSecret | string | `""` | Use existing secret with a credentials. |
+| secret.existingSecret | string | `""` | Use an existing Secret. Automatic endpoint-anchor injection is disabled because the Secret is opaque; set `localEndpointHost.autoInject=true` only after confirming it contains credentials rather than runtime topology, address, or startup-mode overrides. |
 | secret.rustfs.access_key | string | `"rustfsadmin"` | RustFS Access Key ID |
 | secret.rustfs.secret_key | string | `"rustfsadmin"` | RustFS Secret Key ID |
 | service.type | string | `"ClusterIP"` |  |
@@ -202,6 +250,7 @@ uer. `ClusterIssuer` or `Issuer`. |
 | serviceAccount.automount | bool | `true` |  |
 | serviceAccount.create | bool | `true` |  |
 | serviceAccount.name | string | `""` |  |
+| startupWaitTimeoutSeconds | int | `300` | Deprecated and ignored; retained for values-file compatibility. |
 | storageclass.dataStorageSize | string | `"256Mi"` | The storage size for data PVC. |
 | storageclass.logStorageSize | string | `"256Mi"` | The storage size for logs PVC. |
 | storageclass.name | string | `"local-path"` | The name for StorageClass. |
@@ -291,20 +340,17 @@ Notes:
 * **Pools are append-only.** The list index determines the StatefulSet name —
   never remove or reorder entries. Retire a pool with
   `rc admin decommission` before removing it from the list.
-* During the expansion rollout, pods restart until every pod of every pool is
-  resolvable — the server refuses to start with unresolvable peers, so expect
-  a few crash/restart cycles before the cluster converges. This is harmless.
+* With chart-generated volumes, each pod receives an explicit local endpoint
+  identity. An unavailable peer no longer blocks a pod from reaching RustFS's
+  own startup and quorum checks.
 * After the cluster converges, run `rc admin rebalance start <alias>` to
   spread existing objects across the new pool.
 * Pod anti-affinity in pool mode is scoped per pool and **preferred**
   (soft), not required: two pools can share nodes, and each pool's own pods
-  spread across distinct nodes when capacity allows. Soft affinity is
-  load-bearing for in-place expansion — with required rules, the
-  not-yet-rolled pods of the existing pool block the new pool's pods from
-  their nodes while the rolled pods crash on the unresolvable (Pending)
-  peers, deadlocking the rollout on any cluster with fewer nodes than the
-  total pod count. Single-pool deployments (`pools.enabled=false`) keep the
-  chart's existing required anti-affinity unchanged.
+  spread across distinct nodes when capacity allows. Preferred affinity keeps
+  additional pools schedulable when the cluster has fewer nodes than total
+  pods. Single-pool deployments (`pools.enabled=false`) keep the chart's
+  existing required anti-affinity unchanged.
 * The PodDisruptionBudget spans all pools: with the default
   `pdb.maxUnavailable: 1`, at most one pod of the whole cluster may be
   evicted at a time. This is deliberately conservative — quorum safety
@@ -315,7 +361,8 @@ Notes:
 ## Requirement
 
 * Helm V3
-* RustFS >= 1.0.0-alpha.69
+* The RustFS image from the same release as the chart. If `image.rustfs.tag`
+  is overridden, that image must support `RUSTFS_LOCAL_ENDPOINT_HOST`.
 
 Due to the traefik and ingress has different session sticky/affinity annotations, and rustfs support both those two controller, you should specify parameter `ingress.className` to select the right one which suits for you.
 

helm/rustfs/templates/_helpers.tpl

@@ -322,21 +322,23 @@ Render RUSTFS_SERVER_DOMAINS
 {{- join "," $domains -}}
 {{- end -}}
 
-{{/* Render probe command for liveness and readiness
+{{/* Render an mTLS probe command
 */}}
 
 {{- define "rustfs.probeCommand" -}}
-{{- $endpoint_port := .Values.service.endpoint.port | default 9000 -}}
-{{- $console_port := .Values.service.console.port | default 9001 -}}
+{{- $root := .root -}}
+{{- $endpointPath := .endpointPath -}}
+{{- $endpoint_port := $root.Values.service.endpoint.port | default 9000 -}}
+{{- $console_port := $root.Values.service.console.port | default 9001 -}}
 {{- $args := "-skf" -}}
 
-{{- if and .Values.mtls.enabled -}}
-  {{- $args = printf "%s --cert %s --key %s" $args .Values.mtls.clientCertPath .Values.mtls.clientKeyPath -}}
+{{- if and $root.Values.mtls.enabled -}}
+  {{- $args = printf "%s --cert %s --key %s" $args $root.Values.mtls.clientCertPath $root.Values.mtls.clientKeyPath -}}
 {{- end -}}
 - /bin/sh
 - -c
 - |
-  curl {{ $args }} https://127.0.0.1:{{ $endpoint_port }}/health/ready && \
+  curl {{ $args }} https://127.0.0.1:{{ $endpoint_port }}{{ $endpointPath }} && \
   curl {{ $args }} https://127.0.0.1:{{ $console_port }}/rustfs/console/health
 {{- end -}}
 
@@ -350,7 +352,7 @@ livenessProbe:
   {{- if .Values.mtls.enabled }}
   exec:
     command:
-{{ include "rustfs.probeCommand" . | nindent 6 }}
+{{ include "rustfs.probeCommand" (dict "root" . "endpointPath" "/health") | nindent 6 }}
   {{- else }}
   httpGet:
     path: /health
@@ -369,7 +371,7 @@ readinessProbe:
   {{- if .Values.mtls.enabled }}
   exec:
     command:
-{{ include "rustfs.probeCommand" . | nindent 6 }}
+{{ include "rustfs.probeCommand" (dict "root" . "endpointPath" "/health/ready") | nindent 6 }}
   {{- else }}
   httpGet:
     path: /health/ready

helm/rustfs/templates/statefulset.yaml

@@ -1,12 +1,37 @@
 {{- $logDir := .Values.config.rustfs.obs_log_directory }}
 {{- $logDirEnabled := ne $logDir "" }}
 {{- $poolsEnabled := and .Values.pools .Values.pools.enabled }}
+{{- $generatedVolumes := empty .Values.config.rustfs.volumes }}
+{{- $localEndpointHostAutoInject := empty .Values.secret.existingSecret }}
+{{- with .Values.localEndpointHost }}
+{{- if and (hasKey . "autoInject") (ne .autoInject nil) }}
+{{- if not (kindIs "bool" .autoInject) }}
+{{- fail "localEndpointHost.autoInject must be a boolean or null" }}
+{{- end }}
+{{- $localEndpointHostAutoInject = .autoInject }}
+{{- end }}
+{{- end }}
+{{- $injectLocalEndpointHost := and $generatedVolumes $localEndpointHostAutoInject }}
+{{- range $env := .Values.extraEnv -}}
+{{- $name := default "" $env.name -}}
+{{- if eq $name "RUSTFS_VOLUMES" -}}
+{{- $injectLocalEndpointHost = false -}}
+{{- else if eq $name "RUSTFS_LOCAL_ENDPOINT_HOST" -}}
+{{- $injectLocalEndpointHost = false -}}
+{{- else if eq $name "RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" -}}
+{{- if hasKey $env "value" -}}
+{{- $mode := lower (trim (toString $env.value)) -}}
+{{- if and (ne $mode "") (ne $mode "auto") (ne $mode "orchestrated") -}}
+{{- $injectLocalEndpointHost = false -}}
+{{- end -}}
+{{- else -}}
+{{- $injectLocalEndpointHost = false -}}
+{{- end -}}
+{{- end -}}
+{{- end -}}
 {{- if and .Values.mode.distributed.enabled (le (int .Values.replicaCount) 1) -}}
 {{- fail "Distributed mode requires replicaCount >= 2" -}}
 {{- end -}}
-{{- if and .Values.mode.distributed.enabled (le (int .Values.startupWaitTimeoutSeconds) 0) -}}
-{{- fail "startupWaitTimeoutSeconds must be greater than 0 in distributed mode" -}}
-{{- end -}}
 
 {{- if .Values.mode.distributed.enabled }}
 {{- $pools := include "rustfs.pools" . | fromJsonArray }}
@@ -74,13 +99,10 @@ spec:
         {{- if $.Values.affinity.podAntiAffinity.enabled }}
         podAntiAffinity:
           {{- if $poolsEnabled }}
-          {{- /* Pool-scoped and PREFERRED, not required: during in-place
-                 expansion the not-yet-rolled pods' required rules block new
-                 pods from their nodes, while the new pods' DNS must resolve
-                 before the roll can proceed - required rules deadlock
-                 expansion whenever the cluster has fewer nodes than total
-                 pods. Soft anti-affinity converges (the scheduler still
-                 spreads when capacity allows). */}}
+          {{- /* Pool-scoped and PREFERRED, not required: additional pools
+                 must remain schedulable when the cluster has fewer nodes
+                 than total pods. The scheduler still spreads them when
+                 capacity allows. */}}
           preferredDuringSchedulingIgnoredDuringExecution:
             - weight: 100
               podAffinityTerm:
@@ -127,10 +149,6 @@ spec:
           env:
             - name: DRIVES_PER_NODE
               value: {{ $drivesPerNode | quote }}
-            - name: ENDPOINT_PORT
-              value: {{ $.Values.service.endpoint.port | quote }}
-            - name: STARTUP_WAIT_TIMEOUT_SECONDS
-              value: {{ $.Values.startupWaitTimeoutSeconds | quote }}
           command:
             - sh
             - -c
@@ -146,44 +164,6 @@ spec:
               mkdir -p /mnt/rustfs/logs
               chmod 755 /mnt/rustfs/logs
               {{- end }}
-              deadline=$(($(date +%s) + STARTUP_WAIT_TIMEOUT_SECONDS))
-              wait_for_peer() {
-                description=$1
-                shift
-                until "$@" >/dev/null 2>&1; do
-                  if [ "$(date +%s)" -ge "$deadline" ]; then
-                    echo "Timed out after ${STARTUP_WAIT_TIMEOUT_SECONDS}s waiting for ${description}" >&2
-                    exit 1
-                  fi
-                  sleep 2
-                done
-              }
-
-              echo "Waiting for distributed peer DNS records"
-              {{- range $peerPool := $pools }}
-              {{- range $i := until (int $peerPool.replicaCount) }}
-              wait_for_peer "DNS for {{ $peerPool.fullname }}-{{ $i }}" \
-                nslookup "{{ $peerPool.fullname }}-{{ $i }}.{{ include "rustfs.fullname" $ }}-headless.{{ $.Release.Namespace }}.svc.{{ include "rustfs.clusterDomain" $ }}"
-              {{- end }}
-              {{- end }}
-
-              ordinal=${HOSTNAME##*-}
-              echo "Waiting for earlier distributed peers to listen on port ${ENDPOINT_PORT}"
-              {{- range $peerPool := $pools }}
-              {{- if lt (int $peerPool.index) (int $pool.index) }}
-              {{- range $i := until (int $peerPool.replicaCount) }}
-              wait_for_peer "{{ $peerPool.fullname }}-{{ $i }}:${ENDPOINT_PORT}" \
-                nc -z -w 2 "{{ $peerPool.fullname }}-{{ $i }}.{{ include "rustfs.fullname" $ }}-headless.{{ $.Release.Namespace }}.svc.{{ include "rustfs.clusterDomain" $ }}" "$ENDPOINT_PORT"
-              {{- end }}
-              {{- else if eq (int $peerPool.index) (int $pool.index) }}
-              i=0
-              while [ "$i" -lt "$ordinal" ]; do
-                wait_for_peer "{{ $peerPool.fullname }}-${i}:${ENDPOINT_PORT}" \
-                  nc -z -w 2 "{{ $peerPool.fullname }}-${i}.{{ include "rustfs.fullname" $ }}-headless.{{ $.Release.Namespace }}.svc.{{ include "rustfs.clusterDomain" $ }}" "$ENDPOINT_PORT"
-                i=$((i + 1))
-              done
-              {{- end }}
-              {{- end }}
           volumeMounts:
             {{- if gt $drivesPerNode 1 }}
             {{- range $i := until $drivesPerNode }}
@@ -210,9 +190,19 @@ spec:
               containerPort: {{ $.Values.service.endpoint.port }}
             - name: console
               containerPort: {{ $.Values.service.console.port }}
-          {{- with $.Values.extraEnv }}
+          {{- if or $injectLocalEndpointHost (not (empty $.Values.extraEnv)) }}
           env:
+            {{- if $injectLocalEndpointHost }}
+            - name: RUSTFS_CHART_POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: RUSTFS_LOCAL_ENDPOINT_HOST
+              value: {{ printf "$(RUSTFS_CHART_POD_NAME).%s-headless.%s.svc.%s" (include "rustfs.fullname" $) $.Release.Namespace (include "rustfs.clusterDomain" $) | quote }}
+            {{- end }}
+            {{- with $.Values.extraEnv }}
             {{- toYaml . | nindent 12 }}
+            {{- end }}
           {{- end }}
           envFrom:
             - configMapRef:

helm/rustfs/values.yaml

@@ -30,11 +30,21 @@ drivesPerNode: null
 # Provide the DNS root only, without a `svc.` prefix or leading/trailing dots.
 clusterDomain: cluster.local
 
-# Maximum time an init container waits for distributed peers to publish DNS
-# and for earlier peers to start listening. The wait is shared across all
-# peers, not applied once per peer.
+# Deprecated and ignored. Retained so existing values files continue to
+# render after peer startup coordination moved into RustFS.
+# RUSTFS_COMPAT_TODO(rustfs-5416): Keep the removed init-gate setting visible to existing values files. Remove after the minimum supported direct-upgrade chart includes localEndpointHost.autoInject and no longer renders peer gates.
 startupWaitTimeoutSeconds: 300
 
+# Inject a pod-specific endpoint identity for chart-generated distributed
+# topologies. The default enables injection unless secret.existingSecret is
+# set, because the chart cannot inspect runtime overrides hidden in that
+# Secret. Without injection, the server infers a unique matching domain
+# endpoint from its kernel hostname in Kubernetes auto/orchestrated mode; custom
+# aliases must configure RUSTFS_LOCAL_ENDPOINT_HOST explicitly. Set true for a
+# credentials-only existing Secret or false to suppress chart injection.
+localEndpointHost:
+  autoInject: null
+
 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
 image:
   rustfs: # This sets the rustfs image repository and tag.
@@ -223,7 +233,18 @@ config:
         default_key: ""
 
 
-extraEnv: [] # This is for setting extra environment variables in the rustfs container. It should be a list of key value pairs. For example:
+# Extra environment variables for the RustFS container.
+# In distributed mode with chart-generated volumes (config.rustfs.volumes is
+# empty), the chart injects a private Downward API pod name and
+# RUSTFS_LOCAL_ENDPOINT_HOST to identify the pod's local endpoints. An explicit
+# RUSTFS_LOCAL_ENDPOINT_HOST or RUSTFS_VOLUMES entry disables automatic anchor
+# injection. An explicit bounded or fail-fast
+# RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE, or an unrecognized static value, also keeps
+# legacy DNS-based locality discovery. Otherwise Kubernetes auto-detection
+# selects orchestrated startup when RustFS consumes the generated anchor.
+# With custom volumes, the chart does not inject them and extraEnv may define
+# the local endpoint identity explicitly.
+extraEnv: []
 # extraEnv:
 #   - name: RUSTFS_ERASURE_SET_DRIVE_COUNT
 #     value: "16"

scripts/test_helm_templates.sh

@@ -36,6 +36,16 @@ render_distributed_statefulset() {
     '
 }
 
+statefulset_env_names() {
+  yq eval '[(.spec.template.spec.containers[0].env // [])[] | .name] | join(" ")' -
+}
+
+statefulset_env_value() {
+  local rustfs_test_env_name=$1
+  RUSTFS_TEST_ENV_NAME="$rustfs_test_env_name" \
+    yq eval '.spec.template.spec.containers[0].env[] | select(.name == strenv(RUSTFS_TEST_ENV_NAME)) | .value' -
+}
+
 render_distributed_configmap() {
   helm template rustfs "$CHART_DIR" \
     --namespace rustfs \
@@ -403,38 +413,308 @@ if grep -q "name: data$" <<<"$generic_eight_by_two"; then
   exit 1
 fi
 
-# Distributed startup coordination must use the configured endpoint port and
-# stay transport-neutral so the same wait works with and without mTLS.
-custom_port_startup=$(render_distributed_statefulset \
+# Issue #5416's exact three-node, one-drive topology must render all three
+# ordinal hosts while each StatefulSet pod receives its own explicit anchor.
+three_by_one_configmap=$(render_distributed_configmap --set replicaCount=3 --set drivesPerNode=1)
+if ! grep -Fq 'RUSTFS_VOLUMES: "http://rustfs-{0...2}.rustfs-headless.rustfs.svc.cluster.local:9000/data"' <<<"$three_by_one_configmap"; then
+  echo "Three-node topology must render ordinals 0 through 2 in RUSTFS_VOLUMES" >&2
+  exit 1
+fi
+# The distributed init container only prepares directories. Peer DNS or TCP
+# availability must not gate the RustFS process from applying its own quorum
+# checks. The deprecated timeout remains accepted but is not rendered.
+distributed_startup=$(render_distributed_statefulset \
+  --set replicaCount=3 \
+  --set drivesPerNode=1 \
+  --set startupWaitTimeoutSeconds=0)
+for removed_gate in nslookup 'nc -z' wait_for_peer STARTUP_WAIT_TIMEOUT_SECONDS ENDPOINT_PORT; do
+  if grep -Fq "$removed_gate" <<<"$distributed_startup"; then
+    echo "Distributed init must not contain the removed peer gate: $removed_gate" >&2
+    exit 1
+  fi
+done
+if ! grep -Fq 'mkdir -p /data/rustfs$i' <<<"$distributed_startup"; then
+  echo "Distributed init must keep per-drive directory initialization" >&2
+  exit 1
+fi
+if ! grep -Eq '^[[:space:]]+mkdir -p /data$' <<<"$distributed_startup"; then
+  echo "Distributed init must keep single-drive directory initialization" >&2
+  exit 1
+fi
+
+# Chart-generated volumes use the Downward API pod name to construct the exact
+# endpoint hostname. Order is load-bearing because Kubernetes only expands
+# environment references that were defined earlier in the env list.
+generated_env_names=$(statefulset_env_names <<<"$distributed_startup")
+if [[ "$generated_env_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST" ]]; then
+  echo "Unexpected generated topology env order: $generated_env_names" >&2
+  exit 1
+fi
+
+pod_name_field=$(yq eval '.spec.template.spec.containers[0].env[] | select(.name == "RUSTFS_CHART_POD_NAME") | .valueFrom.fieldRef.fieldPath' - <<<"$distributed_startup")
+local_endpoint_host=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$distributed_startup")
+if [[ "$pod_name_field" != "metadata.name" ]]; then
+  echo "Generated topology must source RUSTFS_CHART_POD_NAME from the Downward API" >&2
+  exit 1
+fi
+if [[ "$local_endpoint_host" != '$(RUSTFS_CHART_POD_NAME).rustfs-headless.rustfs.svc.cluster.local' ]]; then
+  echo "Unexpected generated RUSTFS_LOCAL_ENDPOINT_HOST: $local_endpoint_host" >&2
+  exit 1
+fi
+
+# Non-default release identity, namespace, cluster domain, mTLS, and endpoint
+# port must compose into one generated topology without restoring peer gates.
+nondefault_generated=$(helm template tenant "$CHART_DIR" \
+  --namespace object-storage \
+  --set secret.rustfs.access_key=test-access-key \
+  --set secret.rustfs.secret_key=test-secret-key \
+  --set clusterDomain=cluster.corp \
+  --set mtls.enabled=true \
   --set service.endpoint.port=9443 \
   --set config.rustfs.address=:9443)
-grep -q 'name: ENDPOINT_PORT' <<<"$custom_port_startup"
-grep -A1 'name: ENDPOINT_PORT' <<<"$custom_port_startup" | grep -q 'value: "9443"'
-grep -q 'nc -z -w 2' <<<"$custom_port_startup"
-if grep -q 'http.*9000/health' <<<"$custom_port_startup"; then
-  echo "Distributed startup wait must not hardcode the HTTP scheme or port 9000" >&2
+nondefault_statefulset=$(awk '
+  /^# Source: rustfs\/templates\/statefulset.yaml$/ { in_statefulset = 1 }
+  in_statefulset && /^---$/ { exit }
+  in_statefulset { print }
+' <<<"$nondefault_generated")
+nondefault_local_host=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$nondefault_statefulset")
+nondefault_namespace=$(yq eval '.metadata.namespace' - <<<"$nondefault_statefulset")
+nondefault_endpoint_port=$(yq eval '.spec.template.spec.containers[0].ports[] | select(.name == "endpoint") | .containerPort' - <<<"$nondefault_statefulset")
+mtls_liveness_command=$(yq eval '.spec.template.spec.containers[0].livenessProbe.exec.command[-1]' - <<<"$nondefault_statefulset")
+mtls_readiness_command=$(yq eval '.spec.template.spec.containers[0].readinessProbe.exec.command[-1]' - <<<"$nondefault_statefulset")
+if [[ "$nondefault_local_host" != '$(RUSTFS_CHART_POD_NAME).tenant-rustfs-headless.object-storage.svc.cluster.corp' ]]; then
+  echo "Combined generated topology produced an unexpected local endpoint host: $nondefault_local_host" >&2
+  exit 1
+fi
+if [[ "$nondefault_namespace" != "object-storage" || "$nondefault_endpoint_port" != "9443" ]]; then
+  echo "Combined generated topology must retain its namespace and custom endpoint port" >&2
+  exit 1
+fi
+if ! grep -Fq 'https://127.0.0.1:9443/health' <<<"$mtls_liveness_command" ||
+   grep -Fq '/health/ready' <<<"$mtls_liveness_command"; then
+  echo "mTLS liveness must use the process health endpoint, not readiness" >&2
+  exit 1
+fi
+if ! grep -Fq 'https://127.0.0.1:9443/health/ready' <<<"$mtls_readiness_command"; then
+  echo "mTLS readiness must use the ready endpoint" >&2
+  exit 1
+fi
+if ! grep -Fq 'RUSTFS_VOLUMES: "https://tenant-rustfs-{0...3}.tenant-rustfs-headless.object-storage.svc.cluster.corp:9443/data/rustfs{0...3}"' <<<"$nondefault_generated"; then
+  echo "Combined generated topology must use its release fullname, namespace, domain, mTLS scheme, and endpoint port" >&2
+  exit 1
+fi
+if grep -Eq 'nslookup|nc -z|wait_for_peer' <<<"$nondefault_generated"; then
+  echo "Combined generated topology must not render peer DNS or TCP gates" >&2
   exit 1
 fi
 
-mtls_startup=$(render_distributed_statefulset --set mtls.enabled=true)
-grep -q 'nc -z -w 2' <<<"$mtls_startup"
-if grep -q 'wget .*http.*health' <<<"$mtls_startup"; then
-  echo "mTLS startup wait must not use a plaintext HTTP health probe" >&2
+default_scheme_port_generated=$(render_distributed_configmap \
+  --set service.endpoint.port=80 \
+  --set config.rustfs.address=:80)
+if ! grep -Fq 'rustfs-headless.rustfs.svc.cluster.local:80/' <<<"$default_scheme_port_generated"; then
+  echo "Chart-generated topology must retain an explicit HTTP default port" >&2
   exit 1
 fi
 
-# The timeout is one global deadline and invalid non-positive values fail
-# during rendering instead of creating an unbounded init wait.
-grep -q 'deadline=$(($(date +%s) + STARTUP_WAIT_TIMEOUT_SECONDS))' <<<"$custom_port_startup"
-invalid_startup_timeout_status=0
-render_distributed_statefulset --set startupWaitTimeoutSeconds=0 >/dev/null 2>&1 || invalid_startup_timeout_status=$?
-if [[ $invalid_startup_timeout_status -eq 0 ]]; then
-  echo "Distributed mode with startupWaitTimeoutSeconds=0 must fail rendering" >&2
+# Unrelated extraEnv entries follow the generated entries so dependent
+# environment expansion keeps working.
+generated_with_extra_env=$(render_distributed_statefulset \
+  --set 'extraEnv[0].name=CUSTOM_ENV' \
+  --set 'extraEnv[0].value=custom-value')
+generated_with_extra_names=$(statefulset_env_names <<<"$generated_with_extra_env")
+if [[ "$generated_with_extra_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST CUSTOM_ENV" ]]; then
+  echo "extraEnv must be appended after generated topology env entries" >&2
   exit 1
 fi
 
-# Every pool waits for DNS across the complete normalized pool list, while
-# port availability follows the stable pool-index/ordinal startup order.
+# Explicit bounded/fail-fast modes retain their legacy DNS locality semantics,
+# so the generated anchor must not make the runtime reject the configuration.
+for legacy_wait_mode in bounded '  Strict  '; do
+  bounded_wait_mode=$(render_distributed_statefulset \
+    --set 'extraEnv[0].name=RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE' \
+    --set-string "extraEnv[0].value=$legacy_wait_mode")
+  bounded_wait_env_names=$(statefulset_env_names <<<"$bounded_wait_mode")
+  bounded_wait_value=$(yq eval '.spec.template.spec.containers[0].env[0].value' - <<<"$bounded_wait_mode")
+  if [[ "$bounded_wait_env_names" != "RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" || "$bounded_wait_value" != "$legacy_wait_mode" ]]; then
+    echo "Startup mode $legacy_wait_mode must opt out of the generated local endpoint anchor" >&2
+    exit 1
+  fi
+done
+
+unknown_wait_mode=$(render_distributed_statefulset \
+  --set 'extraEnv[0].name=RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE' \
+  --set 'extraEnv[0].value=invalid-mode')
+unknown_wait_env_names=$(statefulset_env_names <<<"$unknown_wait_mode")
+if [[ "$unknown_wait_env_names" != "RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" ]]; then
+  echo "An unknown explicit startup mode must not receive a generated local endpoint anchor" >&2
+  exit 1
+fi
+
+dynamic_wait_mode=$(render_distributed_statefulset \
+  --set 'extraEnv[0].name=RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE' \
+  --set 'extraEnv[0].valueFrom.configMapKeyRef.name=startup-policy' \
+  --set 'extraEnv[0].valueFrom.configMapKeyRef.key=mode')
+dynamic_wait_env_names=$(statefulset_env_names <<<"$dynamic_wait_mode")
+if [[ "$dynamic_wait_env_names" != "RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" ]]; then
+  echo "Dynamically sourced startup mode must opt out of the generated local endpoint anchor" >&2
+  exit 1
+fi
+
+orchestrated_wait_mode=$(render_distributed_statefulset \
+  --set 'extraEnv[0].name=RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE' \
+  --set 'extraEnv[0].value=orchestrated')
+orchestrated_wait_env_names=$(statefulset_env_names <<<"$orchestrated_wait_mode")
+if [[ "$orchestrated_wait_env_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" ]]; then
+  echo "Explicit orchestrated startup mode must retain one generated local endpoint anchor" >&2
+  exit 1
+fi
+
+# Existing POD_NAME remains independent from the chart-private Downward API
+# variable, while an explicit anchor disables automatic anchor injection.
+existing_pod_name=$(render_distributed_statefulset \
+  --set 'extraEnv[0].name=POD_NAME' \
+  --set 'extraEnv[0].valueFrom.fieldRef.fieldPath=metadata.name')
+existing_pod_name_names=$(statefulset_env_names <<<"$existing_pod_name")
+existing_pod_name_count=$(yq eval '[.spec.template.spec.containers[0].env[] | select(.name == "POD_NAME")] | length' - <<<"$existing_pod_name")
+existing_pod_name_field=$(yq eval '.spec.template.spec.containers[0].env[] | select(.name == "POD_NAME") | .valueFrom.fieldRef.fieldPath' - <<<"$existing_pod_name")
+existing_chart_pod_name_field=$(yq eval '.spec.template.spec.containers[0].env[] | select(.name == "RUSTFS_CHART_POD_NAME") | .valueFrom.fieldRef.fieldPath' - <<<"$existing_pod_name")
+existing_pod_name_anchor=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$existing_pod_name")
+if [[ "$existing_pod_name_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST POD_NAME" ||
+      "$existing_pod_name_count" != "1" ||
+      "$existing_pod_name_field" != "metadata.name" ||
+      "$existing_chart_pod_name_field" != "metadata.name" ||
+      "$existing_pod_name_anchor" != '$(RUSTFS_CHART_POD_NAME).rustfs-headless.rustfs.svc.cluster.local' ]]; then
+  echo "An existing extraEnv POD_NAME must be preserved without interfering with the automatic anchor" >&2
+  exit 1
+fi
+
+existing_local_anchor=$(render_distributed_statefulset \
+  --set 'extraEnv[0].name=RUSTFS_LOCAL_ENDPOINT_HOST' \
+  --set-string 'extraEnv[0].value=rustfs-0.rustfs-headless.rustfs.svc.cluster.local')
+existing_local_anchor_names=$(statefulset_env_names <<<"$existing_local_anchor")
+existing_local_anchor_value=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$existing_local_anchor")
+if [[ "$existing_local_anchor_names" != "RUSTFS_LOCAL_ENDPOINT_HOST" || "$existing_local_anchor_value" != "rustfs-0.rustfs-headless.rustfs.svc.cluster.local" ]]; then
+  echo "An existing explicit local endpoint anchor must be preserved without a generated duplicate" >&2
+  exit 1
+fi
+
+# `helm upgrade --reuse-values` can omit keys introduced by the new chart.
+# A missing localEndpointHost map keeps the new safe default instead of
+# failing template evaluation.
+missing_local_endpoint_host_values=$(render_distributed_statefulset \
+  --set-json 'localEndpointHost=null')
+missing_local_endpoint_host_names=$(statefulset_env_names <<<"$missing_local_endpoint_host_values")
+if [[ "$missing_local_endpoint_host_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST" ]]; then
+  echo "Missing localEndpointHost values from a reused release must default to automatic anchor injection" >&2
+  exit 1
+fi
+
+# An opaque existing Secret keeps the historical DNS path by default because
+# it may hide runtime topology or startup-mode overrides. A credentials-only
+# Secret can explicitly opt in without changing envFrom precedence.
+existing_secret_statefulset=$(render_distributed_statefulset \
+  --set secret.existingSecret=legacy-rustfs-config)
+existing_secret_env_names=$(statefulset_env_names <<<"$existing_secret_statefulset")
+existing_secret_env_from=$(yq eval '[.spec.template.spec.containers[0].envFrom[] | (.configMapRef.name // .secretRef.name)] | join(" ")' - <<<"$existing_secret_statefulset")
+if [[ -n "$existing_secret_env_names" || "$existing_secret_env_from" != "rustfs-config legacy-rustfs-config" ]]; then
+  echo "An opaque existingSecret must default to the legacy environment path" >&2
+  exit 1
+fi
+
+existing_secret_opt_in=$(render_distributed_statefulset \
+  --set secret.existingSecret=legacy-rustfs-config \
+  --set localEndpointHost.autoInject=true)
+existing_secret_opt_in_names=$(statefulset_env_names <<<"$existing_secret_opt_in")
+existing_secret_opt_in_env_from=$(yq eval '[.spec.template.spec.containers[0].envFrom[] | (.configMapRef.name // .secretRef.name)] | join(" ")' - <<<"$existing_secret_opt_in")
+if [[ "$existing_secret_opt_in_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST" ||
+      "$existing_secret_opt_in_env_from" != "rustfs-config legacy-rustfs-config" ]]; then
+  echo "A credentials-only existingSecret must support explicit automatic anchor injection" >&2
+  exit 1
+fi
+
+# A Secret that hides a custom topology can explicitly preserve the historical
+# DNS path. No explicit env entries are added, and envFrom ordering is unchanged.
+existing_secret_opt_out=$(render_distributed_statefulset \
+  --set secret.existingSecret=legacy-rustfs-config \
+  --set localEndpointHost.autoInject=false)
+existing_secret_opt_out_env_names=$(statefulset_env_names <<<"$existing_secret_opt_out")
+existing_secret_opt_out_env_from=$(yq eval '[.spec.template.spec.containers[0].envFrom[] | (.configMapRef.name // .secretRef.name)] | join(" ")' - <<<"$existing_secret_opt_out")
+if [[ -n "$existing_secret_opt_out_env_names" || "$existing_secret_opt_out_env_from" != "rustfs-config legacy-rustfs-config" ]]; then
+  echo "localEndpointHost.autoInject=false must preserve the legacy existingSecret environment path" >&2
+  exit 1
+fi
+
+quoted_local_endpoint_host_status=0
+quoted_local_endpoint_host_error=$(render_distributed_statefulset \
+  --set secret.existingSecret=legacy-rustfs-config \
+  --set-string localEndpointHost.autoInject=false 2>&1) || quoted_local_endpoint_host_status=$?
+if [[ $quoted_local_endpoint_host_status -eq 0 ||
+      "$quoted_local_endpoint_host_error" != *"localEndpointHost.autoInject must be a boolean or null"* ]]; then
+  echo "A quoted localEndpointHost.autoInject value must fail closed instead of becoming truthy" >&2
+  exit 1
+fi
+
+missing_local_endpoint_host_with_secret=$(render_distributed_statefulset \
+  --set secret.existingSecret=legacy-rustfs-config \
+  --set-json 'localEndpointHost=null')
+if [[ -n "$(statefulset_env_names <<<"$missing_local_endpoint_host_with_secret")" ]]; then
+  echo "A reused release without localEndpointHost values must not override an opaque existingSecret" >&2
+  exit 1
+fi
+
+# An explicit topology disables inference, but an address override alone keeps
+# the generated anchor so RustFS can validate the effective port before format
+# initialization.
+volumes_override_statefulset=$(render_distributed_statefulset \
+  --set 'extraEnv[0].name=RUSTFS_VOLUMES' \
+  --set-string 'extraEnv[0].value=http://legacy.example/data')
+volumes_override_names=$(statefulset_env_names <<<"$volumes_override_statefulset")
+volumes_override_value=$(yq eval '.spec.template.spec.containers[0].env[0].value' - <<<"$volumes_override_statefulset")
+if [[ "$volumes_override_names" != "RUSTFS_VOLUMES" || "$volumes_override_value" != "http://legacy.example/data" ]]; then
+  echo "extraEnv RUSTFS_VOLUMES must retain precedence without a generated anchor" >&2
+  exit 1
+fi
+
+address_override_statefulset=$(render_distributed_statefulset \
+  --set 'extraEnv[0].name=RUSTFS_ADDRESS' \
+  --set-string 'extraEnv[0].value=:9443')
+address_override_names=$(statefulset_env_names <<<"$address_override_statefulset")
+address_override_value=$(statefulset_env_value RUSTFS_ADDRESS <<<"$address_override_statefulset")
+address_override_anchor_count=$(yq eval '[.spec.template.spec.containers[0].env[] | select(.name == "RUSTFS_LOCAL_ENDPOINT_HOST")] | length' - <<<"$address_override_statefulset")
+if [[ "$address_override_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST RUSTFS_ADDRESS" ||
+      "$address_override_value" != ":9443" ||
+      "$address_override_anchor_count" != "1" ]]; then
+  echo "extraEnv RUSTFS_ADDRESS must retain one generated local endpoint anchor" >&2
+  exit 1
+fi
+
+# Explicit volumes remain authoritative: the chart must not infer any identity,
+# while a user-supplied local endpoint anchor remains allowed.
+custom_volumes_statefulset=$(render_distributed_statefulset \
+  --set config.rustfs.volumes=http://example.test/data)
+custom_volumes_env_names=$(statefulset_env_names <<<"$custom_volumes_statefulset")
+if [[ -n "$custom_volumes_env_names" ]]; then
+  echo "Custom volumes must not receive generated topology env entries" >&2
+  exit 1
+fi
+
+custom_volumes_with_anchor=$(render_distributed_statefulset \
+  --set config.rustfs.volumes=http://example.test/data \
+  --set 'extraEnv[0].name=RUSTFS_LOCAL_ENDPOINT_HOST' \
+  --set 'extraEnv[0].value=example.test')
+custom_anchor=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$custom_volumes_with_anchor")
+if [[ "$custom_anchor" != "example.test" ]]; then
+  echo "Custom volumes must allow a user-supplied local endpoint anchor" >&2
+  exit 1
+fi
+
+if grep -q 'RUSTFS_LOCAL_ENDPOINT_HOST' <<<"$standalone_default_output"; then
+  echo "Standalone mode must not receive a distributed local endpoint anchor" >&2
+  exit 1
+fi
+
+# Every generated server pool uses its runtime pod name with the shared root
+# headless service. No pool waits for another pool's DNS or TCP listener.
 multi_pool_startup=$(helm template rustfs "$CHART_DIR" \
   --namespace rustfs \
   --set secret.rustfs.access_key=test-access-key \
@@ -442,14 +722,19 @@ multi_pool_startup=$(helm template rustfs "$CHART_DIR" \
   --set pools.enabled=true \
   --set 'pools.list[0].replicaCount=2' \
   --set 'pools.list[1].replicaCount=2')
-grep -q 'nslookup "rustfs-0.rustfs-headless.rustfs.svc.cluster.local"' <<<"$multi_pool_startup"
-grep -q 'nslookup "rustfs-pool1-1.rustfs-headless.rustfs.svc.cluster.local"' <<<"$multi_pool_startup"
-if grep -q 'rustfs-pool1-headless' <<<"$multi_pool_startup"; then
-  echo "Multi-pool startup wait must use the shared root headless service" >&2
+multi_pool_anchor_count=$(grep -c 'name: RUSTFS_LOCAL_ENDPOINT_HOST' <<<"$multi_pool_startup" || true)
+if [[ $multi_pool_anchor_count -ne 2 ]]; then
+  echo "Every generated server pool must receive RUSTFS_LOCAL_ENDPOINT_HOST" >&2
+  exit 1
+fi
+if grep -Eq 'nslookup|nc -z|wait_for_peer' <<<"$multi_pool_startup"; then
+  echo "Multi-pool init must not wait for peer DNS or TCP" >&2
+  exit 1
+fi
+if grep -q 'rustfs-pool1-headless' <<<"$multi_pool_startup"; then
+  echo "Multi-pool local endpoint identity must use the shared root headless service" >&2
   exit 1
 fi
-grep -q 'wait_for_peer "rustfs-1:${ENDPOINT_PORT}"' <<<"$multi_pool_startup"
-grep -q 'wait_for_peer "rustfs-pool1-${i}:${ENDPOINT_PORT}"' <<<"$multi_pool_startup"
 
 # volumeClaimTemplates must not contain empty annotations when pvcAnnotations are unset,
 # because Kubernetes treats annotations: {} as a mutation of the immutable field.