fix(ecstore): isolate unavailable local drives during startup (#9353)

* fix(ecstore): isolate unavailable local drives during startup

Keep missing local endpoints in their configured slots and initialize healthy
local drives before applying the existing format and service quorums.
Protect unformatted Linux URL endpoints from writes to the system filesystem,
validate legacy admission, and retry missing local slots during format waits.

Refs rustfs/rustfs#9350
Refs rustfs/backlog#3174

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

* refactor(ecstore): prepare local drives before registry publication

Prepare local disks with bounded concurrency outside inventory locks and publish validated batches under per-instance writer coordination. Preserve existing registrations on failed or cancelled full initialization, reuse current-root handles, and withdraw stale reopened slots.

Unify initial-format and reconnect publication, retain raw and explicit-context constructor contracts, and move filesystem queries into owned native jobs. Centralize format validation and reject special format files before startup writes.

Refs rustfs/rustfs#9350
Refs rustfs/backlog#3174

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

* chore(ecstore): tighten formatted-error ratchet after refactor

Drop the three removed runtime registration call sites from the shrink-only baseline.

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

* test: align drive fixtures with startup admission

Expect the disk-scoped missing-root error and enable the existing directory test mode for the synthetic URL-based receiver fixture.

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

* refactor(ecstore): unify local drive admission and publication

Share bounded preparation and pre-I/O membership snapshots across local
startup and retry paths, with coherent registry and topology publication.
Coalesce native startup preflight and preserve root and buffer ownership.
Keep reconnect ID caches instance-scoped and retain admitted health monitors.

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

* fix(ecstore): pin multipart index queue aliases to owner lifetime

Derive the Linux index alias from the existing identity descriptor so a
retired caller descriptor cannot redirect rebuild or reconnect queue lookup.
Retain the original descriptor assertion and cover cross-disk alias reuse.

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

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/core/sets.rs

@@ -270,15 +270,18 @@ impl Sets {
                 if disk.as_ref().unwrap().is_local() && instance_ctx.is_dist_erasure().await {
                     let local_disk = runtime_sources::local_disk_set_drive(&instance_ctx, pool_idx, i, j).await;
 
-                    if local_disk.is_none() {
+                    let Some(local_disk) = local_disk else {
                         warn!("sets new set_drive {}-{} local_disk is none", i, j);
                         set_drive.push(None);
                         continue;
+                    };
+                    if let Some(previous) = disk.as_ref()
+                        && !Arc::ptr_eq(previous, &local_disk)
+                    {
+                        let _ = previous.close().await;
                     }
 
-                    let _ = disk.as_ref().unwrap().close().await;
-
-                    disk = local_disk;
+                    disk = Some(local_disk);
                 }
 
                 let has_disk_id = disk.as_ref().unwrap().get_disk_id().await.unwrap_or_else(|err| {
@@ -2388,6 +2391,83 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn distributed_sets_preserve_the_admitted_disk_health_monitor() {
+        use crate::layout::endpoints::EndpointServerPools;
+
+        for reuse_registered in [true, false] {
+            let directory = tempfile::tempdir().expect("distributed set root");
+            let metadata = directory.path().join(crate::disk::RUSTFS_META_BUCKET);
+            tokio::fs::create_dir_all(&metadata).await.expect("metadata directory");
+            let mut format = FormatV3::new(1, 1);
+            format.erasure.this = format.erasure.sets[0][0];
+            tokio::fs::write(
+                metadata.join(crate::disk::FORMAT_CONFIG_FILE),
+                format.to_json().expect("disk format bytes"),
+            )
+            .await
+            .expect("persist distributed set format");
+            let mut endpoint =
+                Endpoint::try_from(directory.path().to_str().expect("UTF-8 disk root")).expect("distributed endpoint");
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(0);
+            let pool = PoolEndpoints {
+                legacy: false,
+                set_count: 1,
+                drives_per_set: 1,
+                endpoints: vec![endpoint.clone()].into(),
+                cmd_line: "distributed-health-monitor".to_string(),
+                platform: "test".to_string(),
+            };
+            let context = Arc::new(InstanceContext::new());
+            context.update_erasure_type(SetupType::DistErasure).await;
+            let options = DiskOption {
+                cleanup: false,
+                health_check: true,
+            };
+            runtime_sources::initialize_local_disk_maps(&context, EndpointServerPools(vec![pool.clone()]), &options)
+                .await
+                .expect("register admitted disk");
+            let admitted = runtime_sources::local_disk_set_drive(&context, 0, 0, 0)
+                .await
+                .expect("registered physical handle");
+            let input = if reuse_registered {
+                admitted.clone()
+            } else {
+                new_disk(&endpoint, &options).await.expect("duplicate startup handle")
+            };
+            if !reuse_registered {
+                input.enable_health_check();
+            }
+            admitted.enable_health_check();
+
+            let sets = Sets::new_with_instance_ctx(vec![Some(input.clone())], &pool, &format, 0, 0, context)
+                .await
+                .expect("construct distributed set");
+
+            let active = sets.disk_set[0].disk_inventory().await[0]
+                .clone()
+                .expect("admitted runtime handle");
+            assert!(Arc::ptr_eq(&active, &admitted));
+            let crate::disk::Disk::Local(active_wrapper) = active.as_ref() else {
+                panic!("fixture must retain a local handle");
+            };
+            assert!(active_wrapper.health_check_enabled_for_test());
+            assert!(!active_wrapper.health_monitor_cancelled_for_test());
+            let crate::disk::Disk::Local(input_wrapper) = input.as_ref() else {
+                panic!("fixture input must be local");
+            };
+            assert_eq!(input_wrapper.health_monitor_cancelled_for_test(), !reuse_registered);
+            assert_eq!(active.get_disk_id().await.expect("active handle identity"), Some(format.erasure.this));
+            active
+                .disk_info(&crate::disk::DiskInfoOptions::default())
+                .await
+                .expect("admitted handle remains usable");
+            active.close().await.expect("stop test health monitor");
+        }
+    }
+
     // The legacy constructor keeps single-instance behavior byte-for-byte: it
     // still adopts the process bootstrap context.
     #[tokio::test]

crates/ecstore/src/disk/disk_store.rs

@@ -1950,6 +1950,11 @@ impl LocalDiskWrapper {
         self.health_check
     }
 
+    #[cfg(test)]
+    pub(crate) fn health_monitor_cancelled_for_test(&self) -> bool {
+        self.cancel_token.is_cancelled()
+    }
+
     #[cfg(test)]
     pub(crate) fn override_operation_timeout_for_test(&self, timeout: Duration) -> ScopedDiskOperationTimeout {
         assert!(!timeout.is_zero(), "fault injection needs a positive timeout");

crates/ecstore/src/disk/local.rs

@@ -62,6 +62,7 @@ use crate::disk::HEALING_MARKER_PATH;
 use crate::disk::RUSTFS_META_MULTIPART_BUCKET;
 use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout};
 use crate::disk::io_admission::ResourceIoUnit;
+use crate::disk::io_admission::memory::BackgroundMemoryBudget;
 use crate::disk::io_admission::{DiskIoPermit, current_resource_unit, spawn_blocking_with_disk_io_permit};
 use crate::disk::{
     BUCKET_META_PREFIX, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN,
@@ -1767,6 +1768,11 @@ async fn remove_empty_directory_tree(root: &Path) -> std::io::Result<()> {
     fs::remove_dir(root).await
 }
 
+// Only bounds the legacy exception for an unformatted system-filesystem root.
+// Data filesystems retain the existing topology-derived migration limit.
+#[cfg(target_os = "linux")]
+const ROOT_LEGACY_FORMAT_MAX_BYTES: usize = 1024 * 1024;
+
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_DISK_LOCAL: &str = "disk_local";
 const EVENT_DISK_LOCAL_STARTUP_CLEANUP: &str = "disk_local_startup_cleanup";
@@ -3830,6 +3836,13 @@ pub struct FormatInfo {
     pub last_check: Option<OffsetDateTime>,
 }
 
+struct LocalDiskStartup {
+    publication_root: os::PublicationRoot,
+    format_info: FormatInfo,
+    #[cfg(target_os = "linux")]
+    mount_lease_mount_id: Option<u64>,
+}
+
 /// Durability mode for `write_all_internal`.
 ///
 /// `FileOnly` is reserved for tmp files the caller immediately renames away.
@@ -8244,7 +8257,7 @@ fn resolve_local_disk_root(ep_path: &str) -> Result<PathBuf> {
                 let absolute = match crate::disk::endpoint::windows_fallback_local_path(ep_path, &err, "local disk root") {
                     Ok(path) => path,
                     Err(_) => {
-                        return Err(DiskError::VolumeNotFound);
+                        return Err(DiskError::DiskNotFound);
                     }
                 };
 
@@ -8257,7 +8270,7 @@ fn resolve_local_disk_root(ep_path: &str) -> Result<PathBuf> {
                     }
                     Err(meta_err) => {
                         if meta_err.kind() == ErrorKind::NotFound {
-                            return Err(DiskError::VolumeNotFound);
+                            return Err(DiskError::DiskNotFound);
                         }
                         return Err(to_file_error(meta_err).into());
                     }
@@ -8266,7 +8279,7 @@ fn resolve_local_disk_root(ep_path: &str) -> Result<PathBuf> {
 
             #[cfg(not(windows))]
             {
-                Err(DiskError::VolumeNotFound)
+                Err(DiskError::DiskNotFound)
             }
         }
     }
@@ -8358,11 +8371,6 @@ impl LocalDisk {
             .insert((self.io_format_path.clone(), read), Box::new(hook));
     }
 
-    #[cfg(not(target_os = "linux"))]
-    fn open_mount_lease(root: &Path) -> Result<PathBuf> {
-        Ok(root.to_path_buf())
-    }
-
     fn io_root(&self) -> &Path {
         &self.io_root
     }
@@ -8424,6 +8432,25 @@ impl LocalDisk {
         false
     }
 
+    /// Check the configured root against this handle's captured physical identity.
+    pub(crate) async fn has_current_root(self: &Arc<Self>) -> bool {
+        let disk = Arc::clone(self);
+        spawn_blocking_with_disk_io_permit(None, move || {
+            #[cfg(target_os = "linux")]
+            {
+                disk.has_replacement_mount_lease()
+            }
+            #[cfg(not(target_os = "linux"))]
+            {
+                disk.publication_history
+                    .as_ref()
+                    .is_some_and(|history| history.matches_current_root(Path::new(&disk.endpoint.get_file_path())))
+            }
+        })
+        .await
+        .unwrap_or(false)
+    }
+
     /// Return the descriptor-rooted path for the mount instance admitted for
     /// automatic replacement. Callers must not derive destructive identity
     /// from the mutable endpoint pathname.
@@ -8438,16 +8465,51 @@ impl LocalDisk {
         replacement_lease::acquire(root).await
     }
 
-    pub async fn new(ep: &Endpoint, cleanup: bool) -> Result<Self> {
-        debug!(
-            event = EVENT_DISK_LOCAL_STARTUP_CLEANUP,
-            component = LOG_COMPONENT_ECSTORE,
-            subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
-            endpoint = %ep,
-            state = "create_started",
-            cleanup,
-            "Local disk creation started"
-        );
+    #[cfg(target_os = "linux")]
+    fn unformatted_on_system_filesystem(publication_root: &os::PublicationRoot, ep: &Endpoint) -> Result<bool> {
+        let legacy_format_path = publication_root
+            .io_path()
+            .join(super::MIGRATING_META_BUCKET)
+            .join(super::FORMAT_CONFIG_FILE);
+        let (device, _) = publication_root.physical_identity()?;
+        if device != rustix::fs::stat(SLASH_SEPARATOR).map_err(IoError::from)?.st_dev {
+            return Ok(false);
+        }
+        let file = match publication_root.open_readonly(&legacy_format_path) {
+            Ok(file) => file,
+            Err(err) if err.kind() == ErrorKind::NotFound => return Ok(true),
+            Err(err) => return Err(to_file_error(err).into()),
+        };
+        let metadata = file.metadata()?;
+        if !metadata.is_file() || metadata.len() > u64::try_from(ROOT_LEGACY_FORMAT_MAX_BYTES).map_err(DiskError::other)? {
+            return Err(DiskError::CorruptedFormat);
+        }
+        let mut data = Vec::new();
+        let mut reader = std::io::Read::take(file, u64::try_from(ROOT_LEGACY_FORMAT_MAX_BYTES + 1).map_err(DiskError::other)?);
+        std::io::Read::read_to_end(&mut reader, &mut data)?;
+        if data.is_empty() {
+            return Ok(true);
+        }
+        if data.len() > ROOT_LEGACY_FORMAT_MAX_BYTES {
+            return Err(DiskError::CorruptedFormat);
+        }
+        let format = FormatV3::try_from(data.as_slice()).map_err(|_| DiskError::CorruptedFormat)?;
+        crate::layout::format::validate_format_erasure_layout(&format).map_err(|_| DiskError::CorruptedFormat)?;
+        let max_bytes =
+            crate::layout::format::legacy_format_max_bytes(format.drives()).map_err(|_| DiskError::CorruptedFormat)?;
+        if data.len() > max_bytes || format.erasure.this.is_nil() || format.erasure.this == Uuid::max() {
+            return Err(DiskError::CorruptedFormat);
+        }
+        let (set_index, disk_index) = format
+            .find_disk_index_by_disk_id(format.erasure.this)
+            .map_err(|_| DiskError::CorruptedFormat)?;
+        if i32::try_from(set_index).ok() != Some(ep.set_idx) || i32::try_from(disk_index).ok() != Some(ep.disk_idx) {
+            return Err(DiskError::InconsistentDisk);
+        }
+        Ok(false)
+    }
+
+    fn prepare_startup(ep: &Endpoint, memory: Option<Arc<BackgroundMemoryBudget>>) -> Result<LocalDiskStartup> {
         let endpoint_path = ep.get_file_path();
         let root = resolve_local_disk_root(&endpoint_path).inspect_err(|err| {
             log_startup_disk_error("resolve_local_disk_root", Path::new(&endpoint_path), err);
@@ -8463,7 +8525,13 @@ impl LocalDisk {
         #[cfg(not(windows))]
         let publication_root_path = root.as_path();
         let publication_root = os::PublicationRoot::new(publication_root_path)
-            .map_err(DiskError::from)
+            .map_err(|err| {
+                if err.kind() == ErrorKind::NotFound {
+                    DiskError::DiskNotFound
+                } else {
+                    DiskError::from(err)
+                }
+            })
             .inspect_err(|err| {
                 log_startup_disk_error("open_publication_root", publication_root_path, err);
             })?;
@@ -8474,19 +8542,136 @@ impl LocalDisk {
         let root = publication_root.path().to_path_buf();
 
         #[cfg(target_os = "linux")]
-        let (mount_lease, io_root, mount_lease_mount_id) = (
-            publication_root.directory().clone(),
-            publication_root.io_path().to_path_buf(),
-            mount_id_for_fd(publication_root.directory()),
-        );
+        let mount_lease_mount_id = mount_id_for_fd(publication_root.directory());
+        #[cfg(target_os = "linux")]
+        let io_root = publication_root.io_path();
         #[cfg(not(target_os = "linux"))]
-        let io_root = Self::open_mount_lease(&root)?;
+        let io_root = root.as_path();
+
+        // Use optimized path resolution instead of absolutize_virtually
+        let format_path = root.join(RUSTFS_META_BUCKET).join(super::FORMAT_CONFIG_FILE);
+        let io_format_path = io_root.join(RUSTFS_META_BUCKET).join(super::FORMAT_CONFIG_FILE);
+        debug!(
+            event = EVENT_DISK_LOCAL_STARTUP_CLEANUP,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
+            state = "format_path_resolved",
+            root = ?root,
+            format_path = ?format_path,
+            "Local disk format path resolved"
+        );
+        let (format_data, format_meta) =
+            read_file_exists_native(&io_format_path, &publication_root, memory).inspect_err(|err| {
+                log_startup_disk_error("read_format_json", &io_format_path, err);
+            })?;
+
+        let mut id = None;
+        // let mut format_legacy = false;
+        let mut format_last_check = None;
+
+        if !format_data.is_empty() {
+            let s = format_data.as_ref();
+            let fm = FormatV3::try_from(s).map_err(Error::other)?;
+            if fm.erasure.this.is_nil() {
+                return Err(DiskError::CorruptedFormat);
+            }
+            let (set_idx, disk_idx) = fm.find_disk_index_by_disk_id(fm.erasure.this)?;
+
+            if i32::try_from(set_idx).ok() != Some(ep.set_idx) || i32::try_from(disk_idx).ok() != Some(ep.disk_idx) {
+                return Err(DiskError::InconsistentDisk);
+            }
+
+            id = Some(fm.erasure.this);
+            // format_legacy = fm.erasure.distribution_algo == DistributionAlgoVersion::V1;
+            format_last_check = Some(OffsetDateTime::now_utc());
+        }
+
+        let format_info = FormatInfo {
+            id,
+            data: format_data,
+            file_info: format_meta,
+            last_check: format_last_check,
+        };
+        #[cfg(target_os = "linux")]
+        if format_info.data.is_empty()
+            && ep.get_type() == crate::disk::endpoint::EndpointType::Url
+            && !rustfs_utils::get_env_bool_with_aliases(
+                rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK,
+                &[rustfs_config::ENV_MINIO_CI],
+                rustfs_config::DEFAULT_UNSAFE_BYPASS_DISK_CHECK,
+            )
+        {
+            let unformatted_root_disk = Self::unformatted_on_system_filesystem(&publication_root, ep)?;
+            if unformatted_root_disk {
+                warn!(
+                    event = EVENT_DISK_LOCAL_STARTUP_CLEANUP,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
+                    state = "offline",
+                    endpoint = %ep,
+                    reason = "unformatted_root_filesystem",
+                    "Local endpoint has no format on the system filesystem; restore its data mount"
+                );
+                return Err(DiskError::DiskNotFound);
+            }
+        }
+
+        Ok(LocalDiskStartup {
+            publication_root,
+            format_info,
+            #[cfg(target_os = "linux")]
+            mount_lease_mount_id,
+        })
+    }
+
+    /// Open an existing endpoint root and initialize its internal directories.
+    ///
+    /// # Errors
+    /// Returns [`DiskError::DiskNotFound`] when the root is missing. On Linux,
+    /// an unformatted URL endpoint on the system filesystem also stays offline
+    /// unless an explicit directory-mode override applies. Existing formats,
+    /// including the bounded legacy exception, retain identity validation.
+    /// Permission, format, and filesystem errors propagate to the caller.
+    pub async fn new(ep: &Endpoint, cleanup: bool) -> Result<Self> {
+        debug!(
+            event = EVENT_DISK_LOCAL_STARTUP_CLEANUP,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
+            endpoint = %ep,
+            state = "create_started",
+            cleanup,
+            "Local disk creation started"
+        );
+        let endpoint = ep.clone();
+        let memory = current_resource_unit().and_then(|resources| resources.memory_budget());
+        let LocalDiskStartup {
+            publication_root,
+            format_info,
+            #[cfg(target_os = "linux")]
+            mount_lease_mount_id,
+        } = spawn_blocking_with_disk_io_permit(None, move || Self::prepare_startup(&endpoint, memory))
+            .await
+            .map_err(DiskError::other)??;
+        let root = publication_root.path().to_path_buf();
+        #[cfg(target_os = "linux")]
+        let (mount_lease, io_root) = (publication_root.directory().clone(), publication_root.io_path().to_path_buf());
+        #[cfg(not(target_os = "linux"))]
+        let io_root = root.clone();
+        let format_path = root.join(RUSTFS_META_BUCKET).join(super::FORMAT_CONFIG_FILE);
+        let io_format_path = io_root.join(RUSTFS_META_BUCKET).join(super::FORMAT_CONFIG_FILE);
+        let id = format_info.id;
 
         Self::check_startup_writable(&publication_root, &root).await?;
         multipart_source_state::invalidate_active_multipart_sources_under(&io_root);
         multipart_source_state::invalidate_active_multipart_sources_under(&root);
 
-        let publication_history = os::register_native_publication_root(&root, &io_root);
+        let history_root = publication_root.clone();
+        let history_io_root = io_root.clone();
+        let publication_history = spawn_blocking_with_disk_io_permit(None, move || {
+            os::register_native_publication_root(history_root.path(), &history_io_root)
+        })
+        .await
+        .map_err(DiskError::other)?;
 
         #[cfg(all(test, target_os = "linux"))]
         os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::BeforeDataUsageLayout, &root);
@@ -8523,48 +8708,6 @@ impl LocalDisk {
             );
         }
 
-        // Use optimized path resolution instead of absolutize_virtually
-        let format_path = root.join(RUSTFS_META_BUCKET).join(super::FORMAT_CONFIG_FILE);
-        let io_format_path = io_root.join(RUSTFS_META_BUCKET).join(super::FORMAT_CONFIG_FILE);
-        debug!(
-            event = EVENT_DISK_LOCAL_STARTUP_CLEANUP,
-            component = LOG_COMPONENT_ECSTORE,
-            subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
-            root = ?root,
-            format_path = ?format_path,
-            state = "format_path_resolved",
-            "Local disk format path resolved"
-        );
-        let (format_data, format_meta) = read_file_exists(&io_format_path, &publication_root)
-            .await
-            .inspect_err(|err| {
-                log_startup_disk_error("read_format_json", &io_format_path, err);
-            })?;
-
-        let mut id = None;
-        // let mut format_legacy = false;
-        let mut format_last_check = None;
-
-        if !format_data.is_empty() {
-            let s = format_data.as_ref();
-            let fm = FormatV3::try_from(s).map_err(Error::other)?;
-            let (set_idx, disk_idx) = fm.find_disk_index_by_disk_id(fm.erasure.this)?;
-
-            if set_idx as i32 != ep.set_idx || disk_idx as i32 != ep.disk_idx {
-                return Err(DiskError::InconsistentDisk);
-            }
-
-            id = Some(fm.erasure.this);
-            // format_legacy = fm.erasure.distribution_algo == DistributionAlgoVersion::V1;
-            format_last_check = Some(OffsetDateTime::now_utc());
-        }
-
-        let format_info = FormatInfo {
-            id,
-            data: format_data,
-            file_info: format_meta,
-            last_check: format_last_check,
-        };
         let root_clone = root.clone();
         let update_fn: UpdateFn<DiskInfo> = Box::new(move || {
             let disk_id = id;
@@ -13347,69 +13490,81 @@ fn file_meta_counts_toward_limit(meta: &FileMeta) -> bool {
         .map_or_else(|_| !meta.all_hidden(true), |latest| !latest.deleted && !latest.tier_free_version())
 }
 
-// Filter std::io::ErrorKind::NotFound
 async fn read_file_exists(path: impl AsRef<Path>, root: &os::PublicationRoot) -> Result<(Bytes, Option<Metadata>)> {
-    let p = path.as_ref();
-    let (data, meta) = match read_file_all(&p, root).await {
-        Ok((data, meta)) => (data, Some(meta)),
-        Err(e) => {
-            if e == Error::FileNotFound {
-                (Bytes::new(), None)
-            } else {
-                return Err(e);
-            }
-        }
-    };
+    let path = path.as_ref().to_path_buf();
+    let root = root.clone();
+    let memory = current_resource_unit().and_then(|resources| resources.memory_budget());
+    spawn_blocking_with_disk_io_permit(None, move || read_file_exists_native(&path, &root, memory))
+        .await
+        .map_err(DiskError::from)?
+}
 
-    // let mut data = Vec::new();
-    // if meta.is_some() {
-    //     data = fs::read(&p).await?;
-    // }
-
-    Ok((data, meta))
+fn read_file_exists_native(
+    path: &Path,
+    root: &os::PublicationRoot,
+    memory: Option<Arc<BackgroundMemoryBudget>>,
+) -> Result<(Bytes, Option<Metadata>)> {
+    match read_file_all_native(path, root, memory) {
+        Ok((data, metadata)) => Ok((data, Some(metadata))),
+        Err(Error::FileNotFound) => Ok((Bytes::new(), None)),
+        Err(error) => Err(error),
+    }
 }
 
 async fn read_file_all(path: impl AsRef<Path>, root: &os::PublicationRoot) -> Result<(Bytes, Metadata)> {
     let path = path.as_ref().to_path_buf();
     let root = root.clone();
     let memory = current_resource_unit().and_then(|resources| resources.memory_budget());
-    spawn_blocking_with_disk_io_permit(None, move || {
-        let result = (|| {
-            #[cfg(test)]
-            run_owned_file_before_open(&path);
-            #[cfg(all(any(test, feature = "test-util"), not(windows)))]
-            os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::NativeReadOpen, &path);
-            let file = root.open_readonly(&path)?;
-            let metadata = file.metadata()?;
-            let mut file = std::io::Read::take(file, metadata.len());
-            if let Some(memory) = memory {
-                let hint = usize::try_from(metadata.len()).map_err(|_| std::io::Error::from(ErrorKind::OutOfMemory))?;
-                let data = super::io_admission::memory::BackgroundBuffer::new(Some(memory)).read_to_end(&mut file, hint)?;
-                return Ok((data, metadata));
-            }
-            let size = usize::try_from(metadata.len()).map_err(std::io::Error::other)?;
-            let mut data = Vec::with_capacity(size.saturating_add(1));
-            std::io::Read::read_to_end(&mut file, &mut data)?;
-            Ok::<_, std::io::Error>((Bytes::from(data), metadata))
-        })();
+    spawn_blocking_with_disk_io_permit(None, move || read_file_all_native(&path, &root, memory))
+        .await
+        .map_err(DiskError::from)?
+}
+
+fn read_file_all_native(
+    path: &Path,
+    root: &os::PublicationRoot,
+    memory: Option<Arc<BackgroundMemoryBudget>>,
+) -> Result<(Bytes, Metadata)> {
+    let result = (|| {
         #[cfg(test)]
-        if result.is_ok() {
-            let hook = OWNED_FILE_AFTER_READ.lock().expect("whole-file result hook").remove(&path);
-            if let Some(hook) = hook {
-                hook();
-            }
+        run_owned_file_before_open(path);
+        #[cfg(all(any(test, feature = "test-util"), not(windows)))]
+        os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::NativeReadOpen, path);
+        #[cfg(all(test, not(windows)))]
+        if path.ends_with(Path::new(RUSTFS_META_BUCKET).join(super::FORMAT_CONFIG_FILE)) {
+            os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::FormatReadOpen, root.path());
         }
-        result
-            .inspect_err(|err| {
-                if !matches!(err.kind(), ErrorKind::NotFound | ErrorKind::WouldBlock) {
-                    log_startup_disk_io_error("read_file_all", &path, err);
-                }
-            })
-            .map_err(to_file_error)
-            .map_err(DiskError::from)
-    })
-    .await
-    .map_err(DiskError::from)?
+        let file = root.open_readonly(path)?;
+        let metadata = file.metadata()?;
+        if !metadata.is_file() {
+            return Err(IoError::from(DiskError::IsNotRegular));
+        }
+        let mut file = std::io::Read::take(file, metadata.len());
+        if let Some(memory) = memory {
+            let hint = usize::try_from(metadata.len()).map_err(|_| std::io::Error::from(ErrorKind::OutOfMemory))?;
+            let data = super::io_admission::memory::BackgroundBuffer::new(Some(memory)).read_to_end(&mut file, hint)?;
+            return Ok((data, metadata));
+        }
+        let size = usize::try_from(metadata.len()).map_err(std::io::Error::other)?;
+        let mut data = Vec::with_capacity(size.saturating_add(1));
+        std::io::Read::read_to_end(&mut file, &mut data)?;
+        Ok::<_, std::io::Error>((Bytes::from(data), metadata))
+    })();
+    #[cfg(test)]
+    if result.is_ok() {
+        let hook = OWNED_FILE_AFTER_READ.lock().expect("whole-file result hook").remove(path);
+        if let Some(hook) = hook {
+            hook();
+        }
+    }
+    result
+        .inspect_err(|err| {
+            if !matches!(err.kind(), ErrorKind::NotFound | ErrorKind::WouldBlock) {
+                log_startup_disk_io_error("read_file_all", path, err);
+            }
+        })
+        .map_err(to_file_error)
+        .map_err(DiskError::from)
 }
 
 fn skip_access_checks(p: impl AsRef<str>) -> bool {
@@ -17398,23 +17553,32 @@ fn win32_volume_path(path: &Path) -> std::io::Result<std::borrow::Cow<'_, Path>>
 async fn get_disk_info(drive_path: PathBuf) -> Result<(rustfs_utils::os::DiskInfo, bool)> {
     let drive_path = drive_path.to_string_lossy().to_string();
     check_path_length(&drive_path)?;
+    let root_disk_threshold = runtime_sources::root_disk_threshold_for_erasure_disk().await;
 
-    let disk_info = win32_volume_path(Path::new(&drive_path))
-        .and_then(|volume_path| get_info(&volume_path))
-        .inspect_err(|err| {
-            log_startup_disk_io_error("get_disk_info_stat", Path::new(&drive_path), err);
-        })?;
-    let root_drive = if let Some(root_disk_threshold) = runtime_sources::root_disk_threshold_for_erasure_disk().await {
-        if root_disk_threshold > 0 {
-            disk_info.total <= root_disk_threshold
+    spawn_blocking_with_disk_io_permit(None, move || {
+        #[cfg(all(test, not(windows)))]
+        os::prepared_publication_test_hooks::run(
+            os::prepared_publication_test_hooks::Stage::DiskInfoRead,
+            Path::new(&drive_path),
+        );
+        let disk_info = win32_volume_path(Path::new(&drive_path))
+            .and_then(|volume_path| get_info(&volume_path))
+            .inspect_err(|err| {
+                log_startup_disk_io_error("get_disk_info_stat", Path::new(&drive_path), err);
+            })?;
+        let root_drive = if let Some(threshold) = root_disk_threshold {
+            if threshold > 0 {
+                disk_info.total <= threshold
+            } else {
+                is_root_disk(&drive_path, SLASH_SEPARATOR).unwrap_or_default()
+            }
         } else {
-            is_root_disk(&drive_path, SLASH_SEPARATOR).unwrap_or_default()
-        }
-    } else {
-        false
-    };
-
-    Ok((disk_info, root_drive))
+            false
+        };
+        Ok((disk_info, root_drive))
+    })
+    .await
+    .map_err(DiskError::other)?
 }
 
 #[cfg(test)]
@@ -32824,14 +32988,335 @@ mod test {
     }
 
     #[test]
-    fn resolve_local_disk_root_reports_missing_path_as_volume_not_found() {
+    fn resolve_local_disk_root_reports_missing_path_as_disk_not_found() {
         let dir = tempfile::tempdir().expect("temp dir should be created");
         let missing = dir.path().join("missing");
 
         let err = resolve_local_disk_root(missing.to_str().expect("temp path should be utf8"))
             .expect_err("missing disk root must be rejected");
 
-        assert!(matches!(err, DiskError::VolumeNotFound));
+        assert!(matches!(err, DiskError::DiskNotFound));
+    }
+
+    #[tokio::test]
+    async fn startup_corrupt_format_is_rejected_before_directory_writes() {
+        let dir = tempfile::tempdir().expect("startup disk directory");
+        let meta = dir.path().join(RUSTFS_META_BUCKET);
+        std::fs::create_dir(&meta).expect("existing metadata directory");
+        let format_path = meta.join(super::super::FORMAT_CONFIG_FILE);
+        std::fs::write(&format_path, b"invalid-format").expect("corrupt format fixture");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
+
+        LocalDisk::new(&endpoint, true).await.expect_err("corrupt format must fail");
+
+        assert_eq!(std::fs::read(&format_path).expect("original format"), b"invalid-format");
+        assert_eq!(std::fs::read_dir(dir.path()).expect("disk inventory").count(), 1);
+        assert_eq!(std::fs::read_dir(meta).expect("metadata inventory").count(), 1);
+    }
+
+    #[test]
+    fn startup_preflight_preserves_current_format_budget_and_bytes() {
+        let directory = tempfile::tempdir().expect("current format budget fixture");
+        let metadata = directory.path().join(RUSTFS_META_BUCKET);
+        std::fs::create_dir(&metadata).expect("current metadata directory");
+        let mut format = FormatV3::new(1, 4);
+        format.erasure.this = format.erasure.sets[0][0];
+        let mut data = format.to_json().expect("current format bytes").into_bytes();
+        // The system-filesystem legacy exception must not cap current formats.
+        data.resize(1024 * 1024 + 1, b' ');
+        std::fs::write(metadata.join(super::super::FORMAT_CONFIG_FILE), &data).expect("persist current format");
+        let mut endpoint = Endpoint::try_from(directory.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        let memory = BackgroundMemoryBudget::new();
+
+        assert!(LocalDisk::prepare_startup(&endpoint, Some(memory.clone())).is_err());
+        assert_eq!(memory.snapshot().used, 0, "a rejected read retains no payload allocation");
+        memory.set_sampled_limit(u64::try_from(data.len()).expect("format size"), Duration::from_secs(30));
+        let startup = LocalDisk::prepare_startup(&endpoint, Some(memory.clone())).expect("budgeted startup preflight");
+
+        assert_eq!(startup.format_info.id, Some(format.erasure.this));
+        assert_eq!(startup.format_info.data.as_ref(), data);
+        assert!(startup.format_info.file_info.is_some());
+        assert_eq!(memory.snapshot().used, u64::try_from(data.len()).expect("retained format size"));
+        assert_eq!(std::fs::read_dir(directory.path()).expect("disk inventory").count(), 1);
+        assert_eq!(std::fs::read_dir(metadata).expect("metadata inventory").count(), 1);
+        drop(startup);
+        assert_eq!(memory.snapshot().used, 0, "retiring the format releases its budget");
+    }
+
+    #[cfg(target_os = "linux")]
+    fn system_filesystem_fixture() -> tempfile::TempDir {
+        let root_device = rustix::fs::stat(SLASH_SEPARATOR).expect("system filesystem identity").st_dev;
+        for parent in [
+            std::env::temp_dir(),
+            PathBuf::from("/var/tmp"),
+            PathBuf::from(SLASH_SEPARATOR),
+        ] {
+            if rustix::fs::stat(&parent).is_ok_and(|metadata| metadata.st_dev == root_device)
+                && let Ok(directory) = tempfile::Builder::new()
+                    .prefix("rustfs-startup-admission-")
+                    .tempdir_in(parent)
+            {
+                return directory;
+            }
+        }
+        panic!("startup admission tests require a writable directory on the system filesystem");
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn startup_formats_reject_nonregular_before_writes() {
+        use std::process::{Command, Stdio};
+
+        const CHILD_ROOT: &str = "RUSTFS_TEST_NONREGULAR_FORMAT_ROOT";
+        const CHILD_BUCKET: &str = "RUSTFS_TEST_NONREGULAR_FORMAT_BUCKET";
+        if let Some(root) = std::env::var_os(CHILD_ROOT) {
+            let root = PathBuf::from(root);
+            let bucket = std::env::var(CHILD_BUCKET).expect("child metadata bucket");
+            let endpoint = Endpoint::try_from(format!("http://127.0.0.1:9000{}", root.display()).as_str()).expect("URL endpoint");
+            let result = LocalDisk::new(&endpoint, true).await;
+            let expected = if bucket == RUSTFS_META_BUCKET {
+                DiskError::IsNotRegular
+            } else {
+                assert_eq!(bucket, super::super::MIGRATING_META_BUCKET);
+                DiskError::CorruptedFormat
+            };
+            assert!(
+                matches!(&result, Err(error) if *error == expected),
+                "nonregular format result: {result:?}"
+            );
+            return;
+        }
+
+        for bucket in [RUSTFS_META_BUCKET, super::super::MIGRATING_META_BUCKET] {
+            for fifo in [false, true] {
+                let directory = system_filesystem_fixture();
+                let metadata = directory.path().join(bucket);
+                std::fs::create_dir(&metadata).expect("metadata directory");
+                let format = metadata.join(super::super::FORMAT_CONFIG_FILE);
+                if fifo {
+                    rustix::fs::mkfifoat(rustix::fs::CWD, &format, rustix::fs::Mode::RWXU).expect("format FIFO");
+                } else {
+                    std::fs::create_dir(&format).expect("format directory");
+                }
+                let mut child = Command::new(std::env::current_exe().expect("current test executable"))
+                    .arg("--exact")
+                    .arg("disk::local::test::startup_formats_reject_nonregular_before_writes")
+                    .arg("--nocapture")
+                    .env(CHILD_ROOT, directory.path())
+                    .env(CHILD_BUCKET, bucket)
+                    .env_remove(rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK)
+                    .env_remove(rustfs_config::ENV_MINIO_CI)
+                    .stdout(Stdio::piped())
+                    .stderr(Stdio::piped())
+                    .spawn()
+                    .expect("start isolated format-read check");
+                let completed = tokio::time::timeout(Duration::from_secs(10), async {
+                    let mut poll = tokio::time::interval(Duration::from_millis(10));
+                    loop {
+                        poll.tick().await;
+                        if child.try_wait().expect("poll format-read check").is_some() {
+                            break;
+                        }
+                    }
+                })
+                .await;
+                if completed.is_err() {
+                    child.kill().expect("stop blocked format-read check");
+                }
+                let output = child.wait_with_output().expect("collect format-read check");
+                assert!(completed.is_ok(), "a nonregular format must not block disk initialization");
+                assert!(
+                    output.status.success(),
+                    "format-read check failed: stdout={} stderr={}",
+                    String::from_utf8_lossy(&output.stdout),
+                    String::from_utf8_lossy(&output.stderr)
+                );
+                assert_eq!(std::fs::read_dir(directory.path()).expect("disk inventory").count(), 1);
+                assert_eq!(std::fs::read_dir(metadata).expect("metadata inventory").count(), 1);
+            }
+        }
+    }
+
+    #[cfg(any(unix, windows))]
+    #[tokio::test]
+    async fn current_root_check_accepts_the_opened_endpoint() {
+        let directory = tempfile::tempdir().expect("current-root fixture");
+        let endpoint = Endpoint::try_from(directory.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("open disk root"));
+
+        assert!(disk.has_current_root().await);
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn current_root_check_rejects_a_replaced_endpoint_directory() {
+        let directory = tempfile::tempdir().expect("current-root replacement fixture");
+        let root = directory.path().join("disk");
+        std::fs::create_dir(&root).expect("original disk root");
+        let endpoint = Endpoint::try_from(root.to_str().expect("UTF-8 disk path")).expect("disk endpoint");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("open original root"));
+        assert!(disk.has_current_root().await);
+        std::fs::rename(&root, directory.path().join("original")).expect("retire original root path");
+        std::fs::create_dir(&root).expect("replacement root directory");
+
+        assert!(!disk.has_current_root().await);
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn startup_url_root_guard_isolates_unformatted_system_filesystem_without_writes() {
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>),
+                (rustfs_config::ENV_MINIO_CI, None),
+            ],
+            async {
+                let dir = system_filesystem_fixture();
+                let endpoint = Endpoint::try_from(format!("http://127.0.0.1:9000{}", dir.path().display()).as_str())
+                    .expect("URL disk endpoint");
+                let result = LocalDisk::new(&endpoint, true).await;
+
+                assert!(matches!(result, Err(DiskError::DiskNotFound)));
+                assert_eq!(std::fs::read_dir(dir.path()).expect("offline path inventory").count(), 0);
+            },
+        )
+        .await;
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn startup_url_root_guard_preserves_existing_and_legacy_formats() {
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>),
+                (rustfs_config::ENV_MINIO_CI, None),
+            ],
+            async {
+                for bucket in [RUSTFS_META_BUCKET, super::super::MIGRATING_META_BUCKET] {
+                    let dir = system_filesystem_fixture();
+                    let meta = dir.path().join(bucket);
+                    std::fs::create_dir(&meta).expect("existing metadata directory");
+                    let mut format = FormatV3::new(1, 4);
+                    format.erasure.this = format.erasure.sets[0][0];
+                    let mut bytes = format.to_json().expect("existing format bytes");
+                    if bucket == super::super::MIGRATING_META_BUCKET {
+                        let limit = crate::layout::format::legacy_format_max_bytes(format.drives()).expect("legacy limit");
+                        bytes.extend(std::iter::repeat_n(' ', limit - bytes.len()));
+                    }
+                    let path = meta.join(super::super::FORMAT_CONFIG_FILE);
+                    std::fs::write(&path, &bytes).expect("seed existing format");
+                    let mut endpoint = Endpoint::try_from(format!("http://127.0.0.1:9000{}", dir.path().display()).as_str())
+                        .expect("URL disk endpoint");
+
+                    endpoint.set_pool_index(0);
+                    endpoint.set_set_index(0);
+                    endpoint.set_disk_index(0);
+                    LocalDisk::new(&endpoint, false)
+                        .await
+                        .expect("existing format remains available for quorum validation");
+
+                    assert_eq!(std::fs::read(path).expect("preserved format"), bytes.as_bytes());
+                }
+            },
+        )
+        .await;
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn startup_root_guard_rejects_invalid_legacy_candidates_before_writes() {
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>),
+                (rustfs_config::ENV_MINIO_CI, None),
+            ],
+            async {
+                let mut valid = FormatV3::new(1, 4);
+                valid.erasure.this = valid.erasure.sets[0][0];
+                let valid_bytes = valid.to_json().expect("legacy format bytes");
+                let max_bytes = crate::layout::format::legacy_format_max_bytes(valid.drives()).expect("legacy limit");
+                let mut padded = valid_bytes.into_bytes();
+                padded.resize(max_bytes + 1, b' ');
+                let mut nil = valid.clone();
+                nil.erasure.this = Uuid::nil();
+                for bytes in [
+                    b"invalid-format".to_vec(),
+                    vec![b' '; ROOT_LEGACY_FORMAT_MAX_BYTES + 1],
+                    padded,
+                    nil.to_json().expect("nil identity format").into_bytes(),
+                ] {
+                    let dir = system_filesystem_fixture();
+                    let metadata = dir.path().join(super::super::MIGRATING_META_BUCKET);
+                    std::fs::create_dir(&metadata).expect("legacy metadata directory");
+                    let path = metadata.join(super::super::FORMAT_CONFIG_FILE);
+                    std::fs::write(&path, &bytes).expect("legacy candidate");
+                    let mut endpoint = Endpoint::try_from(format!("http://127.0.0.1:9000{}", dir.path().display()).as_str())
+                        .expect("URL endpoint");
+                    endpoint.set_pool_index(0);
+                    endpoint.set_set_index(0);
+                    endpoint.set_disk_index(0);
+                    assert!(matches!(LocalDisk::new(&endpoint, true).await, Err(DiskError::CorruptedFormat)));
+                    assert_eq!(std::fs::read_dir(dir.path()).expect("disk inventory").count(), 1);
+                    assert_eq!(std::fs::read_dir(&metadata).expect("metadata inventory").count(), 1);
+                    assert_eq!(std::fs::read(&path).expect("preserved candidate"), bytes);
+                }
+            },
+        )
+        .await;
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn startup_url_root_guard_honors_explicit_directory_modes() {
+        for (canonical, legacy) in [(Some("true"), None), (None, Some("true"))] {
+            temp_env::async_with_vars(
+                [
+                    (rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, canonical),
+                    (rustfs_config::ENV_MINIO_CI, legacy),
+                ],
+                async {
+                    let dir = system_filesystem_fixture();
+                    let endpoint = Endpoint::try_from(format!("http://127.0.0.1:9000{}", dir.path().display()).as_str())
+                        .expect("URL disk endpoint");
+                    let disk = LocalDisk::new(&endpoint, false).await.expect("explicit directory deployment");
+                    assert!(disk.path().join(RUSTFS_META_BUCKET).is_dir());
+                },
+            )
+            .await;
+        }
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn startup_root_guard_keeps_the_opened_directory_identity_after_retarget() {
+        let dir = system_filesystem_fixture();
+        let path = dir.path().join("disk");
+        std::fs::create_dir(&path).expect("initial disk directory");
+        let root = os::PublicationRoot::new(&path).expect("pin original disk");
+        let mut endpoint = Endpoint::try_from(format!("http://127.0.0.1:9000{}", path.display()).as_str()).expect("URL endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        let before = LocalDisk::unformatted_on_system_filesystem(&root, &endpoint).expect("initial admission");
+        assert!(before, "the fixture exercises system-filesystem admission");
+        let retained = dir.path().join("retained");
+        std::fs::rename(&path, &retained).expect("retarget original endpoint");
+        std::fs::create_dir_all(path.join(super::super::MIGRATING_META_BUCKET)).expect("replacement metadata");
+        std::fs::write(
+            path.join(super::super::MIGRATING_META_BUCKET)
+                .join(super::super::FORMAT_CONFIG_FILE),
+            b"replacement",
+        )
+        .expect("replacement legacy candidate");
+
+        assert_eq!(
+            LocalDisk::unformatted_on_system_filesystem(&root, &endpoint).expect("pinned admission"),
+            before
+        );
+        assert_eq!(std::fs::read_dir(retained).expect("original disk inventory").count(), 0);
     }
 
     #[tokio::test]

crates/ecstore/src/disk/local/multipart_index.rs

@@ -40,16 +40,15 @@ static OWNERS: LazyLock<parking_lot::Mutex<HashMap<PhysicalRoot, Weak<IndexOwner
     LazyLock::new(|| parking_lot::Mutex::new(HashMap::new()));
 
 pub(super) struct IndexOwner {
-    // Only the path: holding the disk's pinned root handle here would keep the
-    // mount alive past the last disk owner, which the disk lifecycle forbids.
+    // Revalidate the configured path against this owner's physical identity.
     root_path: Arc<PathBuf>,
     physical: PhysicalRoot,
     #[cfg(unix)]
     _identity_pin: std::fs::File,
+    // Linux names _identity_pin so retiring a caller's root cannot retarget it.
     base: PathBuf,
     epoch: Uuid,
-    // Keep the original pinned-fd alias registered while a detached bootstrap
-    // or reconnect still owns this physical disk's index.
+    // Retain this physical owner's queue across bootstrap and disk-handle gaps.
     blocking_queue: Arc<os::DiskBlockingQueue>,
     // No persisted readiness is trusted: an old binary may have written while
     // this process was stopped. Same-process handles share the continuous owner.
@@ -264,14 +263,22 @@ impl IndexOwner {
             }
             pin
         };
+        #[cfg(target_os = "linux")]
+        let base = {
+            use std::os::fd::AsRawFd;
+            PathBuf::from(format!("/proc/self/fd/{}/.", identity_pin.as_raw_fd()))
+        };
+        #[cfg(not(target_os = "linux"))]
+        let base = base.to_owned();
+        let blocking_queue = os::register_disk_blocking_queue(logical_root, &base);
         let owner = Arc::new(Self {
             root_path: Arc::new(root.path().to_path_buf()),
             physical: key,
             #[cfg(unix)]
             _identity_pin: identity_pin,
-            base: base.to_owned(),
+            base,
             epoch: Uuid::new_v4(),
-            blocking_queue: os::register_disk_blocking_queue(logical_root, base),
+            blocking_queue,
             initialized: OnceLock::new(),
             driving: AtomicBool::new(false),
             disabled: AtomicBool::new(false),
@@ -1634,6 +1641,97 @@ mod tests {
         assert!(Arc::ptr_eq(&new.blocking_queue_for_test(), &reconnected.blocking_queue));
     }
 
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn multipart_index_owner_queue_survives_reassigned_caller_alias() {
+        const CHILD: &str = "RUSTFS_TEST_INDEX_OWNER_QUEUE_ALIAS_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            // Alias reassignment mutates a process-wide registry. Isolate it
+            // from concurrent tests that may reuse the retired fd number.
+            let mut child = std::process::Command::new(std::env::current_exe().expect("current test binary"))
+                .arg("--exact")
+                .arg("disk::local::multipart_index::tests::multipart_index_owner_queue_survives_reassigned_caller_alias")
+                .arg("--nocapture")
+                .env(CHILD, "1")
+                .stdout(std::process::Stdio::piped())
+                .stderr(std::process::Stdio::piped())
+                .spawn()
+                .expect("start isolated queue-alias regression");
+            let completed = tokio::time::timeout(std::time::Duration::from_secs(10), async {
+                let mut poll = tokio::time::interval(std::time::Duration::from_millis(10));
+                loop {
+                    poll.tick().await;
+                    if child.try_wait().expect("poll queue-alias regression").is_some() {
+                        break;
+                    }
+                }
+            })
+            .await;
+            if completed.is_err() {
+                child.kill().expect("stop blocked queue-alias regression");
+            }
+            let output = child.wait_with_output().expect("collect queue-alias regression");
+            let stdout = String::from_utf8_lossy(&output.stdout);
+            let stderr = String::from_utf8_lossy(&output.stderr);
+            assert!(completed.is_ok(), "queue-alias regression must finish within its deadline");
+            assert!(output.status.success(), "queue-alias regression failed: {stdout} {stderr}");
+            assert!(stdout.contains("1 passed"), "the child must execute the queue-alias regression: {stdout}");
+            return;
+        }
+
+        let root = tempfile::tempdir().expect("original disk root");
+        let (path, info) = upload("bucket", "legacy");
+        seed_legacy(root.path(), &path, info);
+        let publication = os::PublicationRoot::new(root.path()).expect("original caller root");
+        let caller_alias = publication.io_path().to_path_buf();
+        let owner = IndexOwner::acquire(&publication, root.path(), &caller_alias).expect("original physical owner");
+        let original_queue = owner.blocking_queue.clone();
+        let other_root = tempfile::tempdir().expect("other disk root");
+        let other_publication = os::PublicationRoot::new(other_root.path()).expect("other pinned root");
+        drop(publication);
+
+        // A retired caller's fd number can be reused by another disk. Assign
+        // its alias explicitly so the test does not depend on fd allocation.
+        let other_queue = os::register_disk_blocking_queue(other_root.path(), &caller_alias);
+        let other_attached = os::register_disk_blocking_queue(other_root.path(), other_publication.io_path());
+        assert!(Arc::ptr_eq(&other_queue, &other_attached));
+        assert!(!Arc::ptr_eq(&original_queue, &other_queue));
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), owner.initialize())
+            .await
+            .expect("bootstrap completes after the caller alias is reassigned");
+        assert!(Arc::ptr_eq(
+            owner.observed_rebuild_queue.lock().as_ref().expect("actual rebuild queue"),
+            &original_queue,
+        ));
+        assert_eq!(
+            owner
+                .page("bucket", None, Uuid::new_v4())
+                .await
+                .expect("original disk index coverage")
+                .roots
+                .len(),
+            1,
+        );
+
+        let reconnect = os::PublicationRoot::new(root.path()).expect("reconnected caller root");
+        let attached = owner.attach_blocking_queue(root.path(), reconnect.io_path());
+        assert!(Arc::ptr_eq(&attached, &original_queue));
+        for (path, expected_queue) in [
+            (owner.base.join(INDEX), &original_queue),
+            (reconnect.io_path().join(INDEX), &original_queue),
+            (root.path().join(INDEX), &original_queue),
+            (other_publication.io_path().join(INDEX), &other_queue),
+            (other_root.path().join(INDEX), &other_queue),
+        ] {
+            let lease = os::acquire_namespace_mutation_lease_with_owner(&path, None).await;
+            assert!(
+                Arc::ptr_eq(&lease.blocking_queue_for_test(), expected_queue),
+                "queue routing for {path:?}"
+            );
+        }
+    }
+
     #[test]
     fn multipart_index_rebuild_counts_empty_and_invalid_entries_toward_job_budget() {
         let root = tempfile::tempdir().expect("disk root");

crates/ecstore/src/disk/os.rs

@@ -344,6 +344,10 @@ pub(crate) mod prepared_publication_test_hooks {
         Rename,
         Remove,
         NativeReadOpen,
+        #[cfg(all(test, not(windows)))]
+        FormatReadOpen,
+        #[cfg(all(test, not(windows)))]
+        DiskInfoRead,
         MetadataReadAfterOpen,
         MetadataReadAfterRead,
         #[cfg(test)]
@@ -2332,6 +2336,11 @@ pub(crate) fn record_native_publication(path: &Path) {
 }
 
 impl NativePublicationRegistration {
+    #[cfg(not(target_os = "linux"))]
+    pub(crate) fn matches_current_root(&self, configured_root: &Path) -> bool {
+        native_root_identity(configured_root) == Some(self.identity)
+    }
+
     pub(crate) fn invalidate(&self) {
         if let Some(root) = NATIVE_PUBLICATION_HISTORY.lock().roots.get_mut(&self.identity) {
             root.unknown = true;
@@ -5826,7 +5835,7 @@ impl PublicationRoot {
 
     pub(crate) fn open_readonly(&self, path: &Path) -> io::Result<std::fs::File> {
         #[cfg(target_os = "linux")]
-        return self.open_relative(path, rustix::fs::OFlags::RDONLY, rustix::fs::Mode::empty());
+        return self.open_relative(path, rustix::fs::OFlags::RDONLY | rustix::fs::OFlags::NONBLOCK, rustix::fs::Mode::empty());
         #[cfg(windows)]
         return open_windows_readonly(path);
         #[cfg(not(any(target_os = "linux", windows)))]

crates/ecstore/src/layout/format.rs

@@ -16,8 +16,71 @@ use crate::disk::error::{Error, Result};
 use crate::disk::{DiskInfo, error::DiskError};
 use serde::{Deserialize, Serialize};
 use serde_json::Error as JsonError;
+use std::collections::HashSet;
 use uuid::Uuid;
 
+const LEGACY_FORMAT_BASE_MAX_BYTES: usize = 16 * 1024;
+const LEGACY_FORMAT_MAX_BYTES_PER_DISK: usize = 64;
+
+pub(crate) fn legacy_format_max_bytes(disk_count: usize) -> crate::error::Result<usize> {
+    disk_count
+        .checked_mul(LEGACY_FORMAT_MAX_BYTES_PER_DISK)
+        .and_then(|size| size.checked_add(LEGACY_FORMAT_BASE_MAX_BYTES))
+        .ok_or_else(|| crate::error::Error::other("legacy format size limit overflow"))
+}
+
+fn check_format_erasure_value(format: &FormatV3) -> crate::error::Result<()> {
+    if format.version != FormatMetaVersion::V1 {
+        return Err(crate::error::Error::other("invalid FormatMetaVersion"));
+    }
+
+    if !matches!(format.format, FormatBackend::Erasure | FormatBackend::ErasureSingle) {
+        return Err(crate::error::Error::other("invalid FormatBackend"));
+    }
+
+    if format.id.is_nil() || format.id == Uuid::max() {
+        return Err(crate::error::Error::other("invalid deployment ID"));
+    }
+
+    if format.erasure.version != FormatErasureVersion::V3 {
+        return Err(crate::error::Error::other("invalid FormatErasureVersion"));
+    }
+    Ok(())
+}
+
+pub(crate) fn validate_format_erasure_layout(format: &FormatV3) -> crate::error::Result<usize> {
+    check_format_erasure_value(format)?;
+
+    let set_drive_count = format
+        .erasure
+        .sets
+        .first()
+        .map(Vec::len)
+        .filter(|count| *count > 0)
+        .ok_or_else(|| crate::error::Error::other("erasure.sets must contain at least one drive"))?;
+    if format.erasure.sets.iter().any(|set| set.len() != set_drive_count) {
+        return Err(crate::error::Error::other("erasure.sets must be rectangular"));
+    }
+    let drive_count = format
+        .erasure
+        .sets
+        .len()
+        .checked_mul(set_drive_count)
+        .ok_or_else(|| crate::error::Error::other("erasure set drive count overflow"))?;
+    let mut disk_ids = HashSet::with_capacity(drive_count);
+
+    for disk_id in format.erasure.sets.iter().flatten() {
+        if disk_id.is_nil() || *disk_id == Uuid::max() {
+            return Err(crate::error::Error::other("erasure.sets contains an invalid disk UUID"));
+        }
+        if !disk_ids.insert(*disk_id) {
+            return Err(crate::error::Error::other("erasure.sets contains a duplicate disk UUID"));
+        }
+    }
+
+    Ok(set_drive_count)
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
 pub enum FormatMetaVersion {
     #[serde(rename = "1")]

crates/ecstore/src/runtime/instance.rs

@@ -309,6 +309,9 @@ pub struct InstanceContext {
     /// reads share one registry. Replaces the three `GLOBAL_LOCAL_DISK_*`
     /// statics: the path→disk map, the disk-id→endpoint map, and the
     /// pool/set/drive layout.
+    // Registration writers serialize preparation and publication independently
+    // of the short-lived inventory locks used by RPC readers.
+    local_disk_registration: Mutex<()>,
     local_disk_map: Arc<RwLock<HashMap<String, Option<DiskStore>>>>,
     local_disk_id_map: Arc<RwLock<HashMap<Uuid, String>>>,
     local_disk_set_drives: Arc<RwLock<TypeLocalDiskSetDrives>>,
@@ -478,6 +481,7 @@ impl InstanceContext {
             bucket_monitor: OnceLock::new(),
             replication_stats: OnceCell::new(),
             replication_pool: OnceCell::new(),
+            local_disk_registration: Mutex::new(()),
             local_disk_map: Arc::new(RwLock::new(HashMap::new())),
             local_disk_id_map: Arc::new(RwLock::new(HashMap::new())),
             local_disk_set_drives: Arc::new(RwLock::new(Vec::new())),
@@ -957,11 +961,12 @@ impl InstanceContext {
         Ok(stored.1.clone())
     }
 
+    pub(crate) fn bound_multipart_proof_verification_context(&self) -> Option<ProofVerificationContext> {
+        self.multipart_proof_topology.get().map(|(_, authority)| authority.clone())
+    }
+
     pub(crate) fn multipart_proof_verification_context(&self) -> ProofVerificationContext {
-        self.multipart_proof_topology
-            .get()
-            .map(|(_, authority)| authority.clone())
-            .unwrap_or_default()
+        self.bound_multipart_proof_verification_context().unwrap_or_default()
     }
 
     /// This instance's tier configuration manager, materializing it on first
@@ -1028,6 +1033,10 @@ impl InstanceContext {
     }
 
     /// Handle to this instance's local path→disk map.
+    pub(crate) fn local_disk_registration_lock(&self) -> &Mutex<()> {
+        &self.local_disk_registration
+    }
+
     pub(crate) fn local_disk_map(&self) -> Arc<RwLock<HashMap<String, Option<DiskStore>>>> {
         self.local_disk_map.clone()
     }

crates/ecstore/src/runtime/local_disk_startup_tests.rs

@@ -0,0 +1,414 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![cfg(not(windows))]
+
+use super::initialize_local_disk_maps;
+use crate::disk::format::FormatV3;
+use crate::disk::os::prepared_publication_test_hooks as hooks;
+use crate::disk::{DiskAPI, DiskOption, DiskStore, FORMAT_CONFIG_FILE, RUSTFS_META_BUCKET};
+use crate::layout::endpoint::Endpoint;
+use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
+use crate::runtime::global::TypeLocalDiskSetDrives;
+use crate::runtime::instance::InstanceContext;
+use std::collections::HashMap;
+use std::future::{Future, poll_fn};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Condvar, Mutex};
+use std::task::Poll;
+use std::time::Duration;
+use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel};
+
+const TEST_TIMEOUT: Duration = Duration::from_secs(10);
+
+fn startup_pools(directory: &Path, label: &str, count: usize, width: usize) -> EndpointServerPools {
+    assert!(count > 0 && count.is_multiple_of(width));
+    let format = FormatV3::new(count / width, width);
+    let mut endpoints = Vec::with_capacity(count);
+    for index in 0..count {
+        let path = directory.join(format!("{label}-disk-{index}"));
+        let metadata = path.join(RUSTFS_META_BUCKET);
+        std::fs::create_dir_all(&metadata).expect("startup metadata directory");
+        let mut disk_format = format.clone();
+        disk_format.erasure.this = format.erasure.sets[index / width][index % width];
+        std::fs::write(metadata.join(FORMAT_CONFIG_FILE), disk_format.to_json().expect("startup format bytes"))
+            .expect("persist startup format");
+        let mut endpoint = Endpoint::try_from(path.to_str().expect("UTF-8 startup path")).expect("startup endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(index / width);
+        endpoint.set_disk_index(index % width);
+        endpoints.push(endpoint);
+    }
+    EndpointServerPools(vec![PoolEndpoints {
+        legacy: false,
+        set_count: count / width,
+        drives_per_set: width,
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: "startup-concurrency-test".to_string(),
+        platform: "test".to_string(),
+    }])
+}
+
+fn roots(pools: &EndpointServerPools) -> Vec<PathBuf> {
+    pools.0[0]
+        .endpoints
+        .as_ref()
+        .iter()
+        .map(|endpoint| std::fs::canonicalize(endpoint.get_file_path()).expect("canonical startup root"))
+        .collect()
+}
+
+fn open_options(cleanup: bool) -> DiskOption {
+    DiskOption {
+        cleanup,
+        health_check: false,
+    }
+}
+
+#[derive(Default)]
+struct ReadGate {
+    released: Mutex<bool>,
+    changed: Condvar,
+}
+
+impl ReadGate {
+    fn wait(&self) {
+        let released = self.released.lock().expect("native read gate");
+        drop(
+            self.changed
+                .wait_while(released, |released| !*released)
+                .expect("native read gate wait"),
+        );
+    }
+
+    fn release(&self) {
+        *self.released.lock().expect("release native read gate") = true;
+        self.changed.notify_all();
+    }
+}
+
+#[derive(Default)]
+struct ReadStats {
+    entered: AtomicUsize,
+    active: AtomicUsize,
+    peak: AtomicUsize,
+}
+
+struct ActiveRead(Arc<ReadStats>);
+
+impl Drop for ActiveRead {
+    fn drop(&mut self) {
+        self.0.active.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
+struct NativeReadBarrier {
+    gate: Arc<ReadGate>,
+    stats: Arc<ReadStats>,
+    entries: UnboundedReceiver<usize>,
+    _hooks: Vec<hooks::Guard>,
+}
+
+impl NativeReadBarrier {
+    fn install(paths: &[PathBuf]) -> Self {
+        Self::install_at(hooks::Stage::FormatReadOpen, paths)
+    }
+
+    fn install_at(stage: hooks::Stage, paths: &[PathBuf]) -> Self {
+        let gate = Arc::new(ReadGate::default());
+        let stats = Arc::new(ReadStats::default());
+        let (entered, entries) = unbounded_channel();
+        let mut installed = Vec::with_capacity(paths.len());
+        for (index, path) in paths.iter().enumerate() {
+            let gate = gate.clone();
+            let stats = stats.clone();
+            let entered = entered.clone();
+            installed.push(hooks::install_at(stage, path, move || {
+                let active = stats.active.fetch_add(1, Ordering::SeqCst) + 1;
+                let _active = ActiveRead(stats.clone());
+                stats.peak.fetch_max(active, Ordering::SeqCst);
+                stats.entered.fetch_add(1, Ordering::SeqCst);
+                let _ = entered.send(index);
+                gate.wait();
+            }));
+        }
+        Self {
+            gate,
+            stats,
+            entries,
+            _hooks: installed,
+        }
+    }
+
+    async fn wait_for_entries(&mut self, count: usize) {
+        for _ in 0..count {
+            tokio::time::timeout(TEST_TIMEOUT, self.entries.recv())
+                .await
+                .expect("expected independent native reads to start")
+                .expect("native read entry channel");
+        }
+    }
+
+    fn release(&self) {
+        self.gate.release();
+    }
+}
+
+impl Drop for NativeReadBarrier {
+    fn drop(&mut self) {
+        self.release();
+    }
+}
+
+struct RegistrySnapshot {
+    by_path: HashMap<String, Option<DiskStore>>,
+    by_slot: TypeLocalDiskSetDrives,
+}
+
+fn snapshot_without_waiting(context: &InstanceContext) -> RegistrySnapshot {
+    // Observe both inventories in the publisher's set-to-map lock order.
+    let slots_handle = context.local_disk_set_drives();
+    let slots = slots_handle
+        .try_read()
+        .expect("native disk preparation must not hold the slot inventory lock");
+    let map_handle = context.local_disk_map();
+    let map = map_handle
+        .try_read()
+        .expect("native disk preparation must not hold the RPC disk map lock");
+    RegistrySnapshot {
+        by_path: map.clone(),
+        by_slot: slots.clone(),
+    }
+}
+
+fn assert_same_disk(expected: &Option<DiskStore>, actual: &Option<DiskStore>) {
+    match (expected, actual) {
+        (Some(expected), Some(actual)) => assert!(Arc::ptr_eq(expected, actual), "registered handle must be preserved"),
+        (None, None) => {}
+        _ => panic!("registered slot availability changed"),
+    }
+}
+
+fn assert_same_registry(expected: &RegistrySnapshot, actual: &RegistrySnapshot) {
+    assert_eq!(actual.by_path.len(), expected.by_path.len());
+    for (path, disk) in &expected.by_path {
+        assert_same_disk(disk, actual.by_path.get(path).expect("original endpoint remains registered"));
+    }
+    assert_eq!(actual.by_slot.len(), expected.by_slot.len());
+    for (expected_pool, actual_pool) in expected.by_slot.iter().zip(&actual.by_slot) {
+        assert_eq!(actual_pool.len(), expected_pool.len());
+        for (expected_set, actual_set) in expected_pool.iter().zip(actual_pool) {
+            assert_eq!(actual_set.len(), expected_set.len());
+            for (expected_disk, actual_disk) in expected_set.iter().zip(actual_set) {
+                assert_same_disk(expected_disk, actual_disk);
+            }
+        }
+    }
+}
+
+fn assert_complete_registry(context: &InstanceContext, pools: &EndpointServerPools) {
+    let snapshot = snapshot_without_waiting(context);
+    assert_eq!(snapshot.by_path.len(), pools.0[0].endpoints.as_ref().len());
+    assert_eq!(snapshot.by_slot.len(), 1);
+    assert_eq!(snapshot.by_slot[0].len(), pools.0[0].set_count);
+    for endpoint in pools.0[0].endpoints.as_ref() {
+        let disk = snapshot.by_path[&endpoint.to_string()]
+            .as_ref()
+            .expect("healthy registered endpoint");
+        assert_eq!(disk.endpoint(), *endpoint);
+        let set = usize::try_from(endpoint.set_idx).expect("valid set index");
+        let index = usize::try_from(endpoint.disk_idx).expect("valid disk index");
+        let by_slot = snapshot.by_slot[0][set][index].as_ref().expect("healthy original slot");
+        assert!(Arc::ptr_eq(disk, by_slot), "path and slot inventories must publish the same handle");
+    }
+}
+
+#[tokio::test]
+async fn startup_registration_keeps_inventory_readers_responsive_during_native_io() {
+    let directory = tempfile::tempdir().expect("responsive inventory fixture");
+    let context = Arc::new(InstanceContext::new());
+    let previous = startup_pools(directory.path(), "previous", 4, 4);
+    initialize_local_disk_maps(&context, previous, &open_options(false))
+        .await
+        .expect("initial registry");
+    let before = snapshot_without_waiting(&context);
+    let next = startup_pools(directory.path(), "next", 4, 4);
+    let mut barrier = NativeReadBarrier::install(&roots(&next)[..1]);
+    let task_context = context.clone();
+    let task_pools = next.clone();
+    let task = tokio::spawn(async move { initialize_local_disk_maps(&task_context, task_pools, &open_options(false)).await });
+    barrier.wait_for_entries(1).await;
+
+    assert_same_registry(&before, &snapshot_without_waiting(&context));
+
+    barrier.release();
+    tokio::time::timeout(TEST_TIMEOUT, task)
+        .await
+        .expect("registration completes after native I/O resumes")
+        .expect("registration task")
+        .expect("new registry publishes");
+    assert_complete_registry(&context, &next);
+}
+
+#[tokio::test]
+async fn startup_registration_bounds_parallel_native_opens_across_sets() {
+    let directory = tempfile::tempdir().expect("bounded startup fixture");
+    let context = Arc::new(InstanceContext::new());
+    let pools = startup_pools(directory.path(), "parallel", 32, 16);
+    let mut barrier = NativeReadBarrier::install(&roots(&pools));
+    let task_context = context.clone();
+    let task_pools = pools.clone();
+    let task = tokio::spawn(async move { initialize_local_disk_maps(&task_context, task_pools, &open_options(false)).await });
+    barrier.wait_for_entries(super::LOCAL_DISK_OPEN_CONCURRENCY).await;
+
+    assert_eq!(barrier.stats.active.load(Ordering::SeqCst), super::LOCAL_DISK_OPEN_CONCURRENCY);
+    assert!(snapshot_without_waiting(&context).by_path.is_empty());
+    barrier.release();
+    tokio::time::timeout(TEST_TIMEOUT, task)
+        .await
+        .expect("all startup waves complete")
+        .expect("bounded registration task")
+        .expect("bounded registration succeeds");
+    assert_eq!(barrier.stats.entered.load(Ordering::SeqCst), 32);
+    assert_eq!(barrier.stats.peak.load(Ordering::SeqCst), super::LOCAL_DISK_OPEN_CONCURRENCY);
+    assert_complete_registry(&context, &pools);
+}
+
+#[tokio::test]
+async fn startup_registration_parallelizes_native_disk_info_queries() {
+    let directory = tempfile::tempdir().expect("native disk-info fixture");
+    let context = Arc::new(InstanceContext::new());
+    let pools = startup_pools(directory.path(), "disk-info", 4, 4);
+    let mut barrier = NativeReadBarrier::install_at(hooks::Stage::DiskInfoRead, &roots(&pools));
+    let task_context = context.clone();
+    let task_pools = pools.clone();
+    let task = tokio::spawn(async move { initialize_local_disk_maps(&task_context, task_pools, &open_options(false)).await });
+    barrier.wait_for_entries(4).await;
+
+    assert_eq!(barrier.stats.active.load(Ordering::SeqCst), 4);
+    assert!(snapshot_without_waiting(&context).by_path.is_empty());
+    barrier.release();
+    tokio::time::timeout(TEST_TIMEOUT, task)
+        .await
+        .expect("disk-info preparation completes")
+        .expect("disk-info registration task")
+        .expect("disk-info registration succeeds");
+    assert_complete_registry(&context, &pools);
+}
+
+#[tokio::test]
+async fn failed_startup_preparation_preserves_the_previous_registry() {
+    let directory = tempfile::tempdir().expect("failed startup fixture");
+    let context = Arc::new(InstanceContext::new());
+    initialize_local_disk_maps(&context, startup_pools(directory.path(), "previous", 4, 4), &open_options(false))
+        .await
+        .expect("initial registry");
+    let before = snapshot_without_waiting(&context);
+    let next = startup_pools(directory.path(), "invalid", 4, 4);
+    let corrupt = PathBuf::from(next.0[0].endpoints.as_ref()[1].get_file_path())
+        .join(RUSTFS_META_BUCKET)
+        .join(FORMAT_CONFIG_FILE);
+    std::fs::write(corrupt, b"invalid-format").expect("corrupt new format fixture");
+
+    initialize_local_disk_maps(&context, next, &open_options(false))
+        .await
+        .expect_err("fatal disk preparation must reject the new registry");
+
+    assert_same_registry(&before, &snapshot_without_waiting(&context));
+}
+
+#[tokio::test]
+async fn cancelled_startup_preparation_preserves_the_previous_registry() {
+    let directory = tempfile::tempdir().expect("cancelled startup fixture");
+    let context = Arc::new(InstanceContext::new());
+    initialize_local_disk_maps(&context, startup_pools(directory.path(), "previous", 4, 4), &open_options(false))
+        .await
+        .expect("initial registry");
+    let before = snapshot_without_waiting(&context);
+    let next = startup_pools(directory.path(), "cancelled", 4, 4);
+    let mut barrier = NativeReadBarrier::install(&roots(&next));
+    let task_context = context.clone();
+    let task = tokio::spawn(async move { initialize_local_disk_maps(&task_context, next, &open_options(false)).await });
+    barrier.wait_for_entries(1).await;
+
+    task.abort();
+    assert!(task.await.expect_err("cancelled registration task").is_cancelled());
+    assert_same_registry(&before, &snapshot_without_waiting(&context));
+    barrier.release();
+}
+
+#[tokio::test]
+async fn repeated_registration_reuses_live_handles_and_staging() {
+    let directory = tempfile::tempdir().expect("reused startup fixture");
+    let context = Arc::new(InstanceContext::new());
+    let pools = startup_pools(directory.path(), "reused", 4, 4);
+    initialize_local_disk_maps(&context, pools.clone(), &open_options(false))
+        .await
+        .expect("initial registry");
+    let before = snapshot_without_waiting(&context);
+    let staging = roots(&pools)[0].join(RUSTFS_META_BUCKET).join("tmp/live-startup-writer");
+    std::fs::write(&staging, b"live staging").expect("live startup staging");
+    let barrier = NativeReadBarrier::install(&roots(&pools));
+
+    tokio::time::timeout(TEST_TIMEOUT, initialize_local_disk_maps(&context, pools.clone(), &open_options(true)))
+        .await
+        .expect("healthy registration reuses handles without native format opens")
+        .expect("repeated registration");
+
+    assert_eq!(barrier.stats.entered.load(Ordering::SeqCst), 0);
+    assert_same_registry(&before, &snapshot_without_waiting(&context));
+    assert_eq!(std::fs::read(staging).expect("preserved live staging"), b"live staging");
+}
+
+#[tokio::test]
+async fn concurrent_registration_writers_reuse_the_first_published_batch() {
+    let directory = tempfile::tempdir().expect("serialized registration fixture");
+    let context = Arc::new(InstanceContext::new());
+    initialize_local_disk_maps(&context, startup_pools(directory.path(), "previous", 4, 4), &open_options(false))
+        .await
+        .expect("initial registry");
+    let next = startup_pools(directory.path(), "next", 4, 4);
+    let root = roots(&next)[0].clone();
+    let mut first_barrier = NativeReadBarrier::install(std::slice::from_ref(&root));
+    let task_context = context.clone();
+    let task_pools = next.clone();
+    let first = tokio::spawn(async move { initialize_local_disk_maps(&task_context, task_pools, &open_options(false)).await });
+    first_barrier.wait_for_entries(1).await;
+    let second_barrier = NativeReadBarrier::install(std::slice::from_ref(&root));
+    let options = open_options(true);
+    let mut second = Box::pin(initialize_local_disk_maps(&context, next.clone(), &options));
+    poll_fn(|cx| {
+        assert!(second.as_mut().poll(cx).is_pending(), "a second writer waits for the preparing writer");
+        Poll::Ready(())
+    })
+    .await;
+
+    first_barrier.release();
+    tokio::time::timeout(TEST_TIMEOUT, first)
+        .await
+        .expect("first registration completes")
+        .expect("first registration task")
+        .expect("first registry publishes");
+    let first_snapshot = snapshot_without_waiting(&context);
+    second_barrier.release();
+    tokio::time::timeout(TEST_TIMEOUT, second)
+        .await
+        .expect("second registration completes")
+        .expect("second registry publishes");
+
+    assert_same_registry(&first_snapshot, &snapshot_without_waiting(&context));
+    assert_eq!(second_barrier.stats.entered.load(Ordering::SeqCst), 0);
+    assert_complete_registry(&context, &next);
+}

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

@@ -7079,71 +7079,16 @@ mod heal_result_report_tests {
     #[tokio::test]
     #[serial_test::serial]
     async fn heal_object_with_queued_disk_renewal() {
-        use crate::layout::endpoints::SetupType;
-        use crate::runtime::instance::InstanceContext;
         use crate::set_disk::core::io_primitives::disk_call_counters;
-        use std::collections::HashMap;
         use std::future::Future;
         use std::task::Poll;
         use std::time::Duration;
         use tokio::io::AsyncReadExt;
 
-        // renew_disk still registers local disks on the ambient context. Match
-        // the default serial group used by its other setup/registry fixtures,
-        // and restore only this temporary endpoint, including on a failed join.
-        struct RenewDiskTestState {
-            ctx: Arc<InstanceContext>,
-            was_dist_erasure: bool,
-            map: Arc<RwLock<HashMap<String, Option<DiskStore>>>>,
-            endpoint: String,
-            previous_disk: Option<Option<DiskStore>>,
-        }
-
-        impl Drop for RenewDiskTestState {
-            fn drop(&mut self) {
-                let ctx = self.ctx.clone();
-                let was_dist_erasure = self.was_dist_erasure;
-                let map = self.map.clone();
-                let endpoint = self.endpoint.clone();
-                let previous_disk = self.previous_disk.take();
-                let handle = tokio::runtime::Handle::current();
-                std::thread::spawn(move || {
-                    handle.block_on(async move {
-                        let mut map = map.write().await;
-                        match previous_disk {
-                            Some(disk) => {
-                                map.insert(endpoint, disk);
-                            }
-                            None => {
-                                map.remove(&endpoint);
-                            }
-                        }
-                        drop(map);
-                        if was_dist_erasure {
-                            ctx.update_erasure_type(SetupType::DistErasure).await;
-                        }
-                    });
-                })
-                .join()
-                .expect("renew fixture state restoration should finish");
-            }
-        }
-
         let (_temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
         let endpoint = set.set_endpoints[0].clone();
-        let ctx = crate::runtime::global::current_ctx();
-        let map = ctx.local_disk_map();
-        let restore = RenewDiskTestState {
-            ctx: ctx.clone(),
-            was_dist_erasure: ctx.is_dist_erasure().await,
-            map: map.clone(),
-            endpoint: endpoint.to_string(),
-            previous_disk: map.read().await.get(&endpoint.to_string()).cloned(),
-        };
-        // Only distributed erasure needs an override to avoid the ambient slot array.
-        if restore.was_dist_erasure {
-            ctx.update_erasure_type(SetupType::Erasure).await;
-        }
+        crate::runtime::sources::record_local_disks(set.instance_ctx(), disks.clone()).await;
+        let map = set.instance_ctx().local_disk_map();
 
         let bucket = "heal-disk-renewal-bucket";
         let object = "heal-disk-renewal-object";
@@ -7215,8 +7160,12 @@ mod heal_result_report_tests {
             .get(&endpoint.to_string())
             .cloned()
             .flatten()
-            .expect("renewal must register the connected disk before its inventory write");
-        assert!(!Arc::ptr_eq(&registered, &disks[0]), "renewal must construct a new disk handle");
+            .expect("original RPC route while renewal waits for its inventory write");
+        assert!(Arc::ptr_eq(&registered, &disks[0]));
+        assert!(Arc::ptr_eq(
+            read_gate[0].as_ref().expect("original set slot while renewal waits"),
+            &registered,
+        ));
         tokio::time::timeout(Duration::from_secs(5), async {
             while calls.total(disk_call_counters::KIND_READ_VERSION) < 4 {
                 tokio::task::yield_now().await;
@@ -7240,7 +7189,12 @@ mod heal_result_report_tests {
         let renewed = set.get_disks_internal().await[0]
             .clone()
             .expect("the renewed slot must remain online");
-        assert!(Arc::ptr_eq(&renewed, &registered), "the set must publish the newly connected handle");
+        let registered = map.read().await[&endpoint.to_string()].clone().expect("renewed RPC route");
+        assert!(!Arc::ptr_eq(&renewed, &disks[0]));
+        assert!(
+            Arc::ptr_eq(&renewed, &registered),
+            "the set and RPC route must publish the same renewed handle"
+        );
         assert_eq!(renewed.endpoint(), endpoint);
         let format = load_format_erasure(&renewed, false)
             .await

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

@@ -20,8 +20,9 @@
 //! bounds are unchanged and helper access is via inherent calls.
 
 use super::super::{
-    Arc, DiskError, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, Endpoint, Error, FormatV3, HealChannelPriority, LockResult,
-    NamespaceLock, NamespaceLockWrapper, ObjectKey, Result, SetDisks, StorageError, debug, disk, load_format_erasure, warn,
+    Arc, DiskError, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, Endpoint, Error, FormatV3, HealChannelPriority,
+    LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_SET_DISK, LockResult, NamespaceLock, NamespaceLockWrapper, ObjectKey, Result, SetDisks,
+    StorageError, debug, disk, load_format_erasure, warn,
 };
 use crate::disk::health_state::DriveMembershipSnapshot;
 use crate::disk::{DiskAPI, new_disk};
@@ -31,11 +32,17 @@ use futures::{StreamExt, stream, stream::FuturesUnordered};
 use rand::prelude::SliceRandom;
 use rustfs_heal_contracts::heal_channel::{create_auto_replacement_disks_request, send_heal_request};
 use rustfs_utils::path::object_name_disk_identity;
-#[cfg(test)]
 use uuid::Uuid;
 
 // Bound batching when another reconnect is still pending.
 const REPLACEMENT_BATCH_WAIT: std::time::Duration = std::time::Duration::from_secs(1);
+const EVENT_SET_DISK_RECONNECT: &str = "set_disk_reconnect";
+
+#[derive(Clone, Copy)]
+enum LocalReconnectFormat {
+    Formatted(Uuid),
+    Unformatted,
+}
 
 #[async_trait::async_trait]
 impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
@@ -532,6 +539,25 @@ impl SetDisks {
                 .find(|disk| disk.endpoint() == *ep)
                 .cloned()
         };
+        let previous_registration = if ep.is_local {
+            match runtime_sources::snapshot_local_disk_registration(&self.ctx, ep).await {
+                Ok(registration) => Some(registration),
+                Err(error) => {
+                    warn!(
+                        event = EVENT_SET_DISK_RECONNECT,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        state = "registration_rejected",
+                        endpoint = %ep,
+                        error = %error,
+                        "Local reconnect registration rejected"
+                    );
+                    return false;
+                }
+            }
+        } else {
+            None
+        };
         let previous_health = previous_disk
             .as_ref()
             .and_then(|disk| disk.local_health_tracker_epoch_for_reconnect());
@@ -580,7 +606,7 @@ impl SetDisks {
             return false;
         }
 
-        let _ = new_disk.set_disk_id(Some(fm.erasure.this)).await;
+        let _ = new_disk.set_disk_id_state(Some(fm.erasure.this)).await;
 
         if !new_disk.is_local() {
             let displaced = {
@@ -607,28 +633,79 @@ impl SetDisks {
             return false;
         }
 
-        new_disk.enable_health_check();
-
-        if new_disk.is_local() {
-            let local_disk_map = runtime_sources::local_disk_map_handle();
-            let mut global_local_disk_map = local_disk_map.write().await;
-            let path = new_disk.endpoint().to_string();
-            global_local_disk_map.insert(path, Some(new_disk.clone()));
-
-            if runtime_sources::setup_is_dist_erasure().await {
-                let local_disk_set_drives = runtime_sources::local_disk_set_drives_handle();
-                let mut local_set_drives = local_disk_set_drives.write().await;
-                local_set_drives[self.pool_index][set_idx][disk_idx] = Some(new_disk.clone());
-            }
+        let Some(previous_registration) = previous_registration else {
+            return false;
+        };
+        if let Err(error) = self
+            .publish_local_reconnect(
+                disk_idx,
+                previous_disk.as_ref(),
+                previous_registration,
+                new_disk,
+                LocalReconnectFormat::Formatted(fm.erasure.this),
+            )
+            .await
+        {
+            warn!(
+                event = EVENT_SET_DISK_RECONNECT,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_SET_DISK,
+                state = "publication_failed",
+                endpoint = %ep,
+                error = %error,
+                "Local reconnect publication failed"
+            );
         }
-
-        debug!("renew_disk: update {:?}", fm.erasure.this);
-
-        let mut disk_lock = self.disks.write().await;
-        disk_lock[disk_idx] = Some(new_disk);
         false
     }
 
+    async fn publish_local_reconnect(
+        &self,
+        disk_index: usize,
+        expected_set_disk: Option<&DiskStore>,
+        expected_registration: runtime_sources::LocalDiskRegistration,
+        candidate: DiskStore,
+        format: LocalReconnectFormat,
+    ) -> disk::error::Result<bool> {
+        let endpoint = candidate.endpoint();
+        let registration = runtime_sources::PreparedLocalDisk::new(expected_registration, Some(candidate))
+            .map_err(|error| error.narrow_to_disk().unwrap_or_else(DiskError::other))?;
+        // Registration writers precede the set inventory and global inventories.
+        // Filesystem preparation completes before these publication locks.
+        let _registration = self.ctx.local_disk_registration_lock().lock().await;
+        let mut disks = self.disks.write().await;
+        let Some(slot) = disks.get_mut(disk_index) else {
+            return Err(DiskError::InconsistentDisk);
+        };
+        if slot.as_ref().is_some_and(|current| current.endpoint() != endpoint) {
+            return Err(DiskError::InconsistentDisk);
+        }
+        if !runtime_sources::same_local_disk(slot.as_ref(), expected_set_disk) {
+            return Ok(false);
+        }
+        let published = runtime_sources::publish_local_disks(&self.ctx, vec![registration])
+            .await
+            .map_err(|error| error.narrow_to_disk().unwrap_or_else(DiskError::other))?;
+        let Some(disk) = published.into_iter().next().flatten() else {
+            return Ok(false);
+        };
+        if matches!(format, LocalReconnectFormat::Formatted(_)) {
+            disk.enable_health_check();
+        }
+        *slot = Some(disk);
+        drop(disks);
+        if let LocalReconnectFormat::Formatted(disk_id) = format {
+            // Registry publication is complete. A cancelled cache insert is
+            // recoverable through the RPC resolver's cold identity lookup.
+            self.ctx
+                .local_disk_id_map()
+                .write()
+                .await
+                .insert(disk_id, endpoint.to_string());
+        }
+        Ok(true)
+    }
+
     /// Attach a replacement target only after proving that the exact local slot
     /// is present and unformatted. A health-checked reconnect may reject a
     /// blank target before it reaches the format-heal path; that target still
@@ -645,6 +722,10 @@ impl SetDisks {
             return Ok(false);
         };
 
+        let previous_disk = self.disks.read().await.get(disk_idx).cloned().flatten();
+        let previous_registration = runtime_sources::snapshot_local_disk_registration(&self.ctx, ep)
+            .await
+            .map_err(|error| error.narrow_to_disk().unwrap_or_else(DiskError::other))?;
         let replacement = new_disk(
             ep,
             &DiskOption {
@@ -659,28 +740,14 @@ impl SetDisks {
             Err(err) => return Err(err),
         }
 
-        {
-            let mut disks = self.disks.write().await;
-            if disks[disk_idx].as_ref().is_some_and(|existing| existing.endpoint() != *ep) {
-                warn!(endpoint = %ep, disk_idx, "renew_disk rejected unformatted replacement for an occupied foreign slot");
-                return Ok(false);
-            }
-            disks[disk_idx] = Some(replacement.clone());
-        }
-
-        let local_disk_map = runtime_sources::local_disk_map_handle();
-        local_disk_map
-            .write()
-            .await
-            .insert(replacement.endpoint().to_string(), Some(replacement.clone()));
-
-        if runtime_sources::setup_is_dist_erasure().await {
-            let local_disk_set_drives = runtime_sources::local_disk_set_drives_handle();
-            let mut local_set_drives = local_disk_set_drives.write().await;
-            local_set_drives[self.pool_index][self.set_index][disk_idx] = Some(replacement);
-        }
-
-        Ok(true)
+        self.publish_local_reconnect(
+            disk_idx,
+            previous_disk.as_ref(),
+            previous_registration,
+            replacement,
+            LocalReconnectFormat::Unformatted,
+        )
+        .await
     }
 
     pub(in crate::set_disk) fn find_disk_index(&self, fm: &FormatV3) -> Result<(usize, usize)> {
@@ -828,6 +895,191 @@ mod tests {
         (dir, endpoint, disk)
     }
 
+    async fn isolated_reconnect_set() -> (TempDir, Arc<SetDisks>, DiskStore) {
+        use crate::layout::endpoints::{EndpointServerPools, PoolEndpoints, SetupType};
+        use crate::runtime::instance::InstanceContext;
+
+        let directory = tempfile::tempdir().expect("isolated reconnect root");
+        let metadata = directory.path().join(disk::RUSTFS_META_BUCKET);
+        tokio::fs::create_dir_all(&metadata)
+            .await
+            .expect("reconnect metadata directory");
+        let mut format = FormatV3::new(1, 1);
+        format.erasure.this = format.erasure.sets[0][0];
+        tokio::fs::write(metadata.join(disk::FORMAT_CONFIG_FILE), format.to_json().expect("reconnect format bytes"))
+            .await
+            .expect("persist reconnect format without touching an ambient cache");
+        let mut endpoint =
+            Endpoint::try_from(directory.path().to_str().expect("UTF-8 reconnect root")).expect("isolated reconnect endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        let pool = PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 1,
+            endpoints: vec![endpoint.clone()].into(),
+            cmd_line: "isolated-reconnect".to_string(),
+            platform: "test".to_string(),
+        };
+        let context = Arc::new(InstanceContext::new());
+        context.update_erasure_type(SetupType::Erasure).await;
+        runtime_sources::initialize_local_disk_maps(
+            &context,
+            EndpointServerPools(vec![pool]),
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("initialize isolated reconnect registry");
+        let original = context.local_disk_map().read().await[&endpoint.to_string()]
+            .clone()
+            .expect("registered original disk");
+        let set = SetDisks::new_with_instance_ctx(
+            "isolated-reconnect".to_string(),
+            Arc::new(RwLock::new(vec![Some(original.clone())])),
+            1,
+            0,
+            0,
+            0,
+            vec![endpoint],
+            format,
+            Vec::new(),
+            context,
+        )
+        .await;
+        (directory, set, original)
+    }
+
+    #[tokio::test]
+    async fn local_reconnect_updates_only_the_owning_instance_id_cache() {
+        let (_directory, set, original) = isolated_reconnect_set().await;
+        let endpoint = original.endpoint();
+        let disk_id = set.format.erasure.sets[0][0];
+        let ambient_cache = crate::runtime::global::current_ctx().local_disk_id_map();
+        let before = ambient_cache.read().await.get(&disk_id).cloned();
+
+        set.renew_disk(&endpoint).await;
+
+        let ambient_after = ambient_cache.read().await.get(&disk_id).cloned();
+        assert_eq!(
+            ambient_after, before,
+            "an explicit instance reconnect must leave the ambient cache intact"
+        );
+        assert_eq!(
+            set.instance_ctx().local_disk_id_map().read().await.get(&disk_id),
+            Some(&endpoint.to_string()),
+        );
+        let renewed = set.disk_inventory().await[0].clone().expect("renewed local slot");
+        assert!(!Arc::ptr_eq(&renewed, &original));
+        let registered = set.instance_ctx().local_disk_map().read().await[&endpoint.to_string()]
+            .clone()
+            .expect("renewed local RPC route");
+        assert!(Arc::ptr_eq(&renewed, &registered));
+        renewed.close().await.expect("stop renewed health monitor");
+    }
+
+    #[tokio::test]
+    async fn superseded_local_reconnect_preserves_the_winning_handle_and_id_cache() {
+        let (_directory, set, original) = isolated_reconnect_set().await;
+        let endpoint = original.endpoint();
+        let disk_id = set.format.erasure.sets[0][0];
+        let registration = runtime_sources::snapshot_local_disk_registration(set.instance_ctx(), &endpoint)
+            .await
+            .expect("capture original registration");
+        let (candidate, _) = set
+            .connect_endpoint(&endpoint, None)
+            .await
+            .expect("prepare competing candidate");
+        candidate
+            .set_disk_id_state(Some(disk_id))
+            .await
+            .expect("set candidate identity privately");
+
+        set.renew_disk(&endpoint).await;
+        let winner = set.disk_inventory().await[0].clone().expect("winning renewed disk");
+        let cache = set.instance_ctx().local_disk_id_map();
+        cache.write().await.remove(&disk_id);
+        let accepted = set
+            .publish_local_reconnect(0, Some(&original), registration, candidate, LocalReconnectFormat::Formatted(disk_id))
+            .await
+            .expect("superseded publication is a normal CAS loss");
+
+        assert!(!accepted);
+        assert!(
+            !cache.read().await.contains_key(&disk_id),
+            "a losing candidate must not publish an ID-cache entry"
+        );
+        assert!(Arc::ptr_eq(&winner, set.disk_inventory().await[0].as_ref().expect("winning set slot")));
+        let registered = set.instance_ctx().local_disk_map().read().await[&endpoint.to_string()]
+            .clone()
+            .expect("winning RPC route");
+        assert!(Arc::ptr_eq(&winner, &registered));
+        let slots = set.instance_ctx().local_disk_set_drives();
+        assert!(Arc::ptr_eq(&winner, slots.read().await[0][0][0].as_ref().expect("winning topology slot")));
+        winner.close().await.expect("stop winning health monitor");
+    }
+
+    #[tokio::test]
+    async fn local_reconnect_registry_cas_loss_preserves_the_existing_set_and_cache() {
+        let (_directory, set, original) = isolated_reconnect_set().await;
+        let endpoint = original.endpoint();
+        let disk_id = set.format.erasure.sets[0][0];
+        let stale_registration = runtime_sources::snapshot_local_disk_registration(set.instance_ctx(), &endpoint)
+            .await
+            .expect("capture registration before native preparation");
+        let (candidate, _) = set.connect_endpoint(&endpoint, None).await.expect("prepare losing reconnect");
+        candidate
+            .set_disk_id_state(Some(disk_id))
+            .await
+            .expect("private candidate identity");
+        let (winner, _) = set
+            .connect_endpoint(&endpoint, None)
+            .await
+            .expect("prepare registry replacement");
+        {
+            let _registration = set.instance_ctx().local_disk_registration_lock().lock().await;
+            let registration = runtime_sources::snapshot_local_disk_registration(set.instance_ctx(), &endpoint)
+                .await
+                .expect("capture registry writer's current member");
+            let prepared = runtime_sources::PreparedLocalDisk::new(registration, Some(winner.clone()))
+                .expect("prepare replacement publication");
+            let published = runtime_sources::publish_local_disks(set.instance_ctx(), vec![prepared])
+                .await
+                .expect("a registry writer publishes before the set reconnect");
+            assert!(published[0].is_some());
+        }
+
+        let accepted = set
+            .publish_local_reconnect(
+                0,
+                Some(&original),
+                stale_registration,
+                candidate,
+                LocalReconnectFormat::Formatted(disk_id),
+            )
+            .await
+            .expect("a stale registry token is a normal CAS loss");
+
+        assert!(!accepted);
+        assert!(Arc::ptr_eq(
+            &original,
+            set.disk_inventory().await[0].as_ref().expect("existing set handle")
+        ));
+        assert!(!set.instance_ctx().local_disk_id_map().read().await.contains_key(&disk_id));
+        let registered = set.instance_ctx().local_disk_map().read().await[&endpoint.to_string()]
+            .clone()
+            .expect("registry writer's replacement");
+        assert!(Arc::ptr_eq(&winner, &registered));
+        let slots = set.instance_ctx().local_disk_set_drives();
+        assert!(Arc::ptr_eq(
+            &winner,
+            slots.read().await[0][0][0].as_ref().expect("replacement topology slot")
+        ));
+    }
+
     #[tokio::test]
     async fn quiet_monitor_skips_idle_format_stat_but_request_detects_removal() {
         let format = FormatV3::new(1, 1);
@@ -1464,6 +1716,24 @@ mod tests {
         )
         .await;
 
+        let foreign = set_disks.get_disks_internal().await[0]
+            .as_ref()
+            .expect("existing disk in another slot")
+            .clone();
+        set_disks.disks.write().await[disk_count - 1] = Some(foreign.clone());
+        let result = set_disks.attach_unformatted_replacement_disk(&replacement_endpoint).await;
+        assert!(
+            matches!(result, Err(DiskError::InconsistentDisk)),
+            "an occupied foreign slot must be rejected"
+        );
+        assert!(Arc::ptr_eq(
+            set_disks.get_disks_internal().await[disk_count - 1]
+                .as_ref()
+                .expect("foreign slot remains occupied"),
+            &foreign,
+        ));
+        set_disks.disks.write().await[disk_count - 1] = None;
+
         assert!(
             set_disks
                 .attach_unformatted_replacement_disk(&replacement_endpoint)

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

@@ -29201,13 +29201,7 @@ mod delete_recovery_inventory_tests {
             .await
             .expect("seed actual version");
         let endpoint = set.set_endpoints[0].clone();
-        let ctx = crate::runtime::global::current_ctx();
-        let was_dist = ctx.is_dist_erasure().await;
-        if was_dist {
-            ctx.update_erasure_type(crate::layout::endpoints::SetupType::Erasure).await;
-        }
-        let map = ctx.local_disk_map();
-        let previous = map.read().await.get(&endpoint.to_string()).cloned();
+        let map = set.instance_ctx().local_disk_map();
 
         let barrier = DeleteCleanupBarrier::install(bucket);
         let delete_store = store.clone();
@@ -29250,8 +29244,12 @@ mod delete_recovery_inventory_tests {
             .get(&endpoint.to_string())
             .cloned()
             .flatten()
-            .expect("renewal registered new handle");
-        assert!(!Arc::ptr_eq(&connected, &disks[0]));
+            .expect("registered original handle while renewal waits");
+        assert!(Arc::ptr_eq(&connected, &disks[0]));
+        assert!(Arc::ptr_eq(
+            extra_gate[0].as_ref().expect("original set slot while renewal waits"),
+            &connected,
+        ));
         drop(extra_gate);
         barrier.release();
         let outcome = tokio::time::timeout(Duration::from_secs(3), async { tokio::join!(renewal.as_mut(), &mut deletion) }).await;
@@ -29262,25 +29260,14 @@ mod delete_recovery_inventory_tests {
         if outcome.is_err() {
             let _ = deletion.await;
         }
-        {
-            let mut map = map.write().await;
-            match previous {
-                Some(old) => {
-                    map.insert(endpoint.to_string(), old);
-                }
-                None => {
-                    map.remove(&endpoint.to_string());
-                }
-            }
-        }
-        if was_dist {
-            ctx.update_erasure_type(crate::layout::endpoints::SetupType::DistErasure)
-                .await;
-        }
         let (_, result) =
             outcome.expect("batch DELETE and real queued disk renewal must finish without nested inventory-read deadlock");
         let (_, errors) = result.expect("delete task must finish");
         assert!(errors.iter().all(Option::is_none), "DELETE result must remain successful: {errors:?}");
+        let renewed = set.disk_inventory().await[0].clone().expect("renewed set slot");
+        let registered = map.read().await[&endpoint.to_string()].clone().expect("renewed RPC route");
+        assert!(!Arc::ptr_eq(&renewed, &disks[0]));
+        assert!(Arc::ptr_eq(&renewed, &registered));
     }
 }
 

crates/ecstore/src/store/init.rs

@@ -488,6 +488,7 @@ impl ECStore {
 
         preflight_startup_rpc_secret(&endpoint_pools)?;
         instance_ctx.bind_multipart_proof_topology(&endpoint_pools)?;
+        runtime_sources::ensure_local_disk_maps(&instance_ctx, endpoint_pools.clone()).await?;
 
         let mut deployment_id = None;
         let mut pool_meta_bootstrap_authorities = Vec::new();
@@ -497,8 +498,6 @@ impl ECStore {
         let mut pools = Vec::with_capacity(endpoint_pools.as_ref().len());
         let mut disk_map = HashMap::with_capacity(endpoint_pools.as_ref().len());
 
-        let mut local_disks = Vec::new();
-
         debug!(
             event = EVENT_ECSTORE_INIT_STATUS,
             component = LOG_COMPONENT_ECSTORE,
@@ -611,6 +610,19 @@ impl ECStore {
                     for disk in disks.iter().flatten() {
                         disk.reset_health_for_store_init_retry();
                     }
+                    tokio::select! {
+                        _ = ctx.cancelled() => {
+                            return Err(Error::other(std::io::Error::new(
+                                std::io::ErrorKind::Interrupted,
+                                "storage format initialization cancelled",
+                            )));
+                        }
+                        refreshed = init_format::refresh_missing_local_disks(
+                            &instance_ctx,
+                            &pool_eps.endpoints,
+                            &mut disks,
+                        ) => refreshed?,
+                    }
                 }
             }?;
             // First-hand authority for this pool only: `Fresh` when this process
@@ -636,23 +648,12 @@ impl ECStore {
                 deployment_id = Some(Uuid::new_v4());
             }
 
-            for disk in disks.iter() {
-                if disk.is_some() && disk.as_ref().expect("operation should succeed").is_local() {
-                    local_disks.push(disk.as_ref().expect("operation should succeed").clone());
-                }
-            }
-
             let sets = Sets::new_with_instance_ctx(disks.clone(), pool_eps, &fm, i, parity_drives, instance_ctx.clone()).await?;
             pools.push(sets);
 
             disk_map.insert(i, disks);
         }
 
-        // Replace the local disk
-        if !instance_ctx.is_dist_erasure().await {
-            runtime_sources::record_local_disks(&instance_ctx, local_disks).await;
-        }
-
         let deployment_id = deployment_id.ok_or_else(|| Error::other("store init failed: deployment id is not initialized"))?;
         let peer_sys = S3PeerSys::new_with_instance_ctx(&endpoint_pools, instance_ctx.clone());
         let mut pool_meta = PoolMeta::new(&pools, &PoolMeta::default())?;
@@ -3482,9 +3483,7 @@ mod tests {
         (instance_ctx, store, shutdown)
     }
 
-    #[tokio::test]
-    #[serial_test::serial(storage_class_env)]
-    async fn direct_local_constructor_multipart_binds_complete_topology_without_registry_preinit() {
+    async fn assert_direct_local_constructor_multipart(configure_endpoints: bool) {
         use crate::disk::DiskAPI;
         use crate::disk::multipart_source::ProofVerificationContext;
         use crate::layout::endpoints::SetupType;
@@ -3510,6 +3509,9 @@ mod tests {
         }]);
         let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
         ctx.update_erasure_type(SetupType::Erasure).await;
+        if configure_endpoints {
+            ctx.set_endpoints(topology.clone());
+        }
         assert!(ctx.local_disk_map().read().await.is_empty());
         assert!(matches!(ctx.multipart_proof_verification_context(), ProofVerificationContext::RpcOnly));
         let shutdown = CancellationToken::new();
@@ -3561,11 +3563,28 @@ mod tests {
             .expect("the direct local constructor must support native multipart completion");
         assert!(completed.is_multipart());
         let disks = set.disks.read().await.clone();
+        let registered = ctx.local_disk_map().read().await.clone();
+        let layout = ctx.local_disk_set_drives().read().await.clone();
+        assert_eq!(disks.len(), 4, "direct construction keeps the configured set width");
+        assert_eq!(registered.len(), disks.len(), "direct construction publishes every local drive");
+        assert_eq!((layout.len(), layout[0].len(), layout[0][0].len()), (1, 1, 4));
         for (slot, disk) in disks.iter().enumerate() {
+            let disk = disk.as_ref().expect("direct constructor populates the runtime slot");
             assert_eq!(
-                disk.as_ref().unwrap().get_disk_id().await.unwrap(),
+                disk.get_disk_id().await.expect("runtime slot disk identity"),
                 Some(set.format.erasure.sets[0][slot])
             );
+            let route = registered
+                .get(&disk.endpoint().to_string())
+                .and_then(Option::as_ref)
+                .expect("runtime slot has a local RPC route");
+            assert!(Arc::ptr_eq(disk, route), "runtime slot and RPC registry share their admitted handle");
+            assert!(Arc::ptr_eq(
+                disk,
+                layout[0][0][slot]
+                    .as_ref()
+                    .expect("direct construction publishes the original topology slot"),
+            ));
         }
         let mut reader = set
             .get_object_reader(bucket, object, None, http::HeaderMap::new(), &opts)
@@ -3577,6 +3596,18 @@ mod tests {
         shutdown.cancel();
     }
 
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn direct_local_constructor_multipart_binds_complete_topology_without_registry_preinit() {
+        assert_direct_local_constructor_multipart(false).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn direct_configured_local_constructor_preserves_registry_and_native_multipart() {
+        assert_direct_local_constructor_multipart(true).await;
+    }
+
     /// CopyObject destinations and CreateMultipartUpload create keys, so they
     /// must reject exactly the keys PutObject rejects, with the same error.
     /// The copy path reaches the pools' put_object directly and used to admit

crates/ecstore/src/store/init_format.rs

@@ -16,17 +16,20 @@ use crate::cluster::rpc::client::is_network_like_disk_error;
 use crate::config::storageclass;
 use crate::core::pools::PoolMetaBootstrapAuthority;
 use crate::disk::error_reduce::{count_errs, reduce_write_quorum_errs};
+#[cfg(test)]
+use crate::disk::format::FormatBackend;
 use crate::disk::{self, DiskAPI};
 use crate::error::{Error, Result};
+use crate::layout::format::{legacy_format_max_bytes, validate_format_erasure_layout};
 use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources::{
-    quarantine_local_disks, quarantine_local_disks_preserving_bootstrap_routes, reconcile_local_disk_ids,
+    PreparedLocalDisk, prepare_disks, publish_local_disks, quarantine_local_disks,
+    quarantine_local_disks_preserving_bootstrap_routes, reconcile_local_disk_ids,
 };
 use crate::{
     disk::{
-        DiskInfoOptions, DiskOption, DiskStore, FORMAT_CONFIG_FILE, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET,
-        error::DiskError,
-        format::{FormatBackend, FormatErasureVersion, FormatMetaVersion, FormatV3},
+        DiskInfoOptions, DiskOption, DiskStore, FORMAT_CONFIG_FILE, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET, error::DiskError,
+        format::FormatV3,
     },
     layout::endpoints::Endpoints,
 };
@@ -40,54 +43,104 @@ use tracing::{debug, error, info, warn};
 use uuid::Uuid;
 
 const LEGACY_FORMAT_READ_CONCURRENCY: usize = 16;
-const LEGACY_FORMAT_BASE_MAX_BYTES: usize = 16 * 1024;
-const LEGACY_FORMAT_MAX_BYTES_PER_DISK: usize = 64;
 
 pub async fn init_disks(
     instance_ctx: &InstanceContext,
     eps: &Endpoints,
     opt: &DiskOption,
 ) -> (Vec<Option<DiskStore>>, Vec<Option<DiskError>>) {
-    let mut futures = Vec::with_capacity(eps.as_ref().len());
-    let local_disk_map = instance_ctx.local_disk_map();
-    let local_disks = local_disk_map.read().await;
-
-    for ep in eps.as_ref().iter() {
-        // Foundation disks may already serve RPC writes. Opening them again
-        // with startup cleanup can move live staging out from under a writer.
-        let registered = if ep.is_local {
-            local_disks.get(&ep.to_string()).cloned().flatten()
-        } else {
-            None
+    let _registration = instance_ctx.local_disk_registration_lock().lock().await;
+    let results = prepare_disks(instance_ctx, eps.as_ref(), opt).await;
+    let mut disks = vec![None; results.len()];
+    let mut errors = vec![None; results.len()];
+    let mut prepared = Vec::new();
+    let mut publication_indices = Vec::new();
+    for (index, opened) in results.into_iter().enumerate() {
+        let disk = match opened.result {
+            Ok(disk) => Some(disk),
+            Err(error) => {
+                errors[index] = Some(error);
+                None
+            }
         };
-        futures.push(async move {
-            match registered {
-                Some(disk) if disk.endpoint() == *ep => Ok(disk),
-                Some(_) => Err(DiskError::InconsistentDisk),
-                None => crate::disk::new_disk_for_instance(instance_ctx, ep, opt).await,
+        let registration = match opened.registration {
+            Ok(Some(registration)) => registration,
+            Ok(None) => {
+                // A raw opener owns no runtime membership to publish.
+                disks[index] = disk;
+                continue;
             }
-        });
-    }
-    drop(local_disks);
-
-    let mut res = Vec::with_capacity(eps.as_ref().len());
-    let mut errors = Vec::with_capacity(eps.as_ref().len());
-
-    let results = join_all(futures).await;
-    for result in results {
-        match result {
-            Ok(s) => {
-                res.push(Some(s));
-                errors.push(None);
+            Err(error) => {
+                errors[index].get_or_insert(error);
+                continue;
             }
-            Err(e) => {
-                res.push(None);
-                errors.push(Some(e));
+        };
+        match PreparedLocalDisk::new(registration, disk) {
+            Ok(registration) => {
+                prepared.push(registration);
+                publication_indices.push(index);
+            }
+            Err(error) => {
+                errors[index].get_or_insert_with(|| error.narrow_to_disk().unwrap_or_else(DiskError::other));
             }
         }
     }
+    match publish_local_disks(instance_ctx, prepared).await {
+        Ok(published) => {
+            for (index, disk) in publication_indices.into_iter().zip(published) {
+                if disk.is_none() && errors[index].is_none() {
+                    errors[index] = Some(DiskError::DiskNotFound);
+                }
+                disks[index] = disk;
+            }
+        }
+        Err(error) => {
+            let error = error.narrow_to_disk().unwrap_or_else(DiskError::other);
+            for index in publication_indices {
+                errors[index].get_or_insert_with(|| error.clone());
+            }
+        }
+    }
+    (disks, errors)
+}
 
-    (res, errors)
+pub(crate) async fn refresh_missing_local_disks(
+    instance_ctx: &Arc<InstanceContext>,
+    eps: &Endpoints,
+    disks: &mut [Option<DiskStore>],
+) -> Result<()> {
+    if eps.as_ref().len() != disks.len() {
+        return Err(Error::CorruptedFormat);
+    }
+    let _registration = instance_ctx.local_disk_registration_lock().lock().await;
+    let pending = eps
+        .as_ref()
+        .iter()
+        .enumerate()
+        .filter(|(index, endpoint)| disks[*index].is_none() && endpoint.is_local)
+        .map(|(index, endpoint)| (index, endpoint.clone()))
+        .collect::<Vec<_>>();
+    let endpoints = pending.iter().map(|(_, endpoint)| endpoint.clone()).collect::<Vec<_>>();
+    let opened = prepare_disks(
+        instance_ctx,
+        &endpoints,
+        &DiskOption {
+            cleanup: false,
+            health_check: true,
+        },
+    )
+    .await;
+    let mut prepared = Vec::with_capacity(opened.len());
+    for opened in opened {
+        prepared.push(opened.into_local_registration()?);
+    }
+    let published = publish_local_disks(instance_ctx, prepared).await?;
+    for ((index, _), disk) in pending.into_iter().zip(published) {
+        if let Some(disk) = disk {
+            disks[index] = Some(disk);
+        }
+    }
+    Ok(())
 }
 
 #[cfg(test)]
@@ -602,13 +655,6 @@ fn select_persisted_migration_format(
     Err(write_error.map_or(quorum_error, Into::into))
 }
 
-fn legacy_format_max_bytes(disk_count: usize) -> Result<usize> {
-    disk_count
-        .checked_mul(LEGACY_FORMAT_MAX_BYTES_PER_DISK)
-        .and_then(|size| size.checked_add(LEGACY_FORMAT_BASE_MAX_BYTES))
-        .ok_or_else(|| Error::other("legacy format size limit overflow"))
-}
-
 async fn read_legacy_format_bytes(reader: impl tokio::io::AsyncRead + Unpin, max_bytes: usize) -> disk::error::Result<Vec<u8>> {
     let read_limit = max_bytes.checked_add(1).ok_or(DiskError::CorruptedFormat)?;
     let mut data = Vec::with_capacity(read_limit.min(64 * 1024));
@@ -799,58 +845,6 @@ fn check_format_erasure_value_for_topology(format: &FormatV3, format_count: usiz
     Ok(())
 }
 
-fn check_format_erasure_value(format: &FormatV3) -> Result<()> {
-    if format.version != FormatMetaVersion::V1 {
-        return Err(Error::other("invalid FormatMetaVersion"));
-    }
-
-    if !matches!(format.format, FormatBackend::Erasure | FormatBackend::ErasureSingle) {
-        return Err(Error::other("invalid FormatBackend"));
-    }
-
-    if format.id.is_nil() || format.id == Uuid::max() {
-        return Err(Error::other("invalid deployment ID"));
-    }
-
-    if format.erasure.version != FormatErasureVersion::V3 {
-        return Err(Error::other("invalid FormatErasureVersion"));
-    }
-    Ok(())
-}
-
-fn validate_format_erasure_layout(format: &FormatV3) -> Result<usize> {
-    check_format_erasure_value(format)?;
-
-    let set_drive_count = format
-        .erasure
-        .sets
-        .first()
-        .map(Vec::len)
-        .filter(|count| *count > 0)
-        .ok_or_else(|| Error::other("erasure.sets must contain at least one drive"))?;
-    if format.erasure.sets.iter().any(|set| set.len() != set_drive_count) {
-        return Err(Error::other("erasure.sets must be rectangular"));
-    }
-    let drive_count = format
-        .erasure
-        .sets
-        .len()
-        .checked_mul(set_drive_count)
-        .ok_or_else(|| Error::other("erasure set drive count overflow"))?;
-    let mut disk_ids = HashSet::with_capacity(drive_count);
-
-    for disk_id in format.erasure.sets.iter().flatten() {
-        if disk_id.is_nil() || *disk_id == Uuid::max() {
-            return Err(Error::other("erasure.sets contains an invalid disk UUID"));
-        }
-        if !disk_ids.insert(*disk_id) {
-            return Err(Error::other("erasure.sets contains a duplicate disk UUID"));
-        }
-    }
-
-    Ok(set_drive_count)
-}
-
 // load_format_erasure_all reads all format.json files
 pub async fn load_format_erasure_all(disks: &[Option<DiskStore>], heal: bool) -> (Vec<Option<FormatV3>>, Vec<Option<DiskError>>) {
     let (formats, errors, _) = load_format_erasure_all_observing_bootstrap_routes(disks, heal).await;
@@ -1069,6 +1063,318 @@ mod tests {
         (temp_dir, disks)
     }
 
+    #[tokio::test]
+    async fn refresh_missing_local_disks_restores_format_quorum_and_bootstrap_routes() {
+        use crate::layout::endpoints::{EndpointServerPools, PoolEndpoints};
+
+        let (directory, original_disks) = local_disks(4).await;
+        let format = FormatV3::new(1, 4);
+        for (index, disk) in original_disks.iter().enumerate() {
+            let mut disk_format = format.clone();
+            disk_format.erasure.this = format.erasure.sets[0][index];
+            save_format_file(disk, &Some(disk_format))
+                .await
+                .expect("seed existing format");
+        }
+        let endpoints = Endpoints::from(
+            original_disks
+                .iter()
+                .flatten()
+                .map(|disk| disk.endpoint())
+                .collect::<Vec<_>>(),
+        );
+        drop(original_disks);
+        for index in 1..4 {
+            tokio::fs::rename(
+                directory.path().join(format!("disk-{index}")),
+                directory.path().join(format!("disk-{index}.offline")),
+            )
+            .await
+            .expect("take a startup drive offline");
+        }
+        let pools = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: true,
+            set_count: 1,
+            drives_per_set: 4,
+            endpoints: endpoints.clone(),
+            cmd_line: "startup-format-retry".to_string(),
+            platform: String::new(),
+        }]);
+        let context = Arc::new(InstanceContext::new());
+        crate::runtime::sources::initialize_local_disk_maps(&context, pools, &DiskOption::default())
+            .await
+            .expect("register the surviving local drive");
+        let (mut disks, _) = init_disks(&context, &endpoints, &DiskOption::default()).await;
+        let healthy = disks[0].clone().expect("surviving drive handle");
+        let staging = healthy.path().join(RUSTFS_META_BUCKET).join("tmp/live-startup-writer");
+        tokio::fs::write(&staging, b"live staging").await.expect("seed live staging");
+        assert!(matches!(
+            connect_load_init_formats_with_instance_ctx(&context, true, &mut disks, 1, 4, None).await,
+            Err(Error::ErasureReadQuorum)
+        ));
+
+        tokio::fs::rename(directory.path().join("disk-1.offline"), directory.path().join("disk-1"))
+            .await
+            .expect("restore a drive while startup waits for quorum");
+        refresh_missing_local_disks(&context, &endpoints, &mut disks)
+            .await
+            .expect("refresh returned local drives");
+        assert!(Arc::ptr_eq(&healthy, disks[0].as_ref().expect("preserved healthy handle")));
+        let returned = disks[1].as_ref().expect("returned drive handle").clone();
+        let registered = context.local_disk_map();
+        let registered = registered.read().await;
+        assert!(Arc::ptr_eq(
+            &returned,
+            registered
+                .get(&endpoints.as_ref()[1].to_string())
+                .and_then(Option::as_ref)
+                .expect("returned drive bootstrap route")
+        ));
+        drop(registered);
+        let registered_slots = context.local_disk_set_drives();
+        let registered_slots = registered_slots.read().await;
+        assert!(Arc::ptr_eq(
+            &returned,
+            registered_slots[0][0][1].as_ref().expect("returned drive original slot")
+        ));
+        drop(registered_slots);
+
+        let loaded = connect_load_init_formats_with_instance_ctx(&context, true, &mut disks, 1, 4, None)
+            .await
+            .expect("returned drive restores existing-format read quorum");
+        assert_eq!(loaded.format.shared_identity(), format.shared_identity());
+        assert!(matches!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::None));
+        assert!(disks[2..].iter().all(Option::is_none));
+        assert_eq!(tokio::fs::read(staging).await.expect("preserved staging"), b"live staging");
+    }
+
+    #[tokio::test]
+    async fn refresh_missing_local_disks_preserves_remote_slots_and_rejects_bad_local_format() {
+        use crate::layout::endpoints::{EndpointServerPools, PoolEndpoints};
+
+        let directory = tempfile::tempdir().expect("startup retry fixture");
+        let path = directory.path().join("disk-0");
+        let mut local = Endpoint::try_from(path.to_str().expect("UTF-8 local endpoint")).expect("local endpoint");
+        let mut remote = Endpoint::try_from("http://remote.invalid:9000/disk-1").expect("remote endpoint");
+        local.set_pool_index(0);
+        local.set_set_index(0);
+        local.set_disk_index(0);
+        remote.set_pool_index(0);
+        remote.set_set_index(0);
+        remote.set_disk_index(1);
+        let endpoints = Endpoints::from(vec![local, remote]);
+        let pools = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: true,
+            set_count: 1,
+            drives_per_set: 2,
+            endpoints: endpoints.clone(),
+            cmd_line: "startup-format-retry".to_string(),
+            platform: String::new(),
+        }]);
+        let context = Arc::new(InstanceContext::new());
+        crate::runtime::sources::initialize_local_disk_maps(&context, pools, &DiskOption::default())
+            .await
+            .expect("missing local drive remains registered offline");
+        let mut disks = vec![None, None];
+        refresh_missing_local_disks(&context, &endpoints, &mut disks)
+            .await
+            .expect("missing local and remote slots remain offline");
+        assert!(disks.iter().all(Option::is_none));
+
+        tokio::fs::create_dir_all(path.join(RUSTFS_META_BUCKET))
+            .await
+            .expect("restore metadata directory");
+        tokio::fs::write(path.join(RUSTFS_META_BUCKET).join(FORMAT_CONFIG_FILE), b"invalid-format")
+            .await
+            .expect("corrupt returned format");
+        refresh_missing_local_disks(&context, &endpoints, &mut disks)
+            .await
+            .expect_err("returned corrupt format must retain its startup error");
+        assert!(disks.iter().all(Option::is_none));
+        assert!(context.local_disk_map().read().await[&endpoints.as_ref()[0].to_string()].is_none());
+    }
+
+    #[tokio::test]
+    async fn init_disks_registers_drive_returned_after_foundation_before_format_loading() {
+        use crate::layout::endpoints::{EndpointServerPools, PoolEndpoints, SetupType};
+
+        let (directory, originals) = local_disks(4).await;
+        let format = FormatV3::new(1, 4);
+        for (index, disk) in originals.iter().enumerate() {
+            let mut current = format.clone();
+            current.erasure.this = format.erasure.sets[0][index];
+            save_format_file(disk, &Some(current)).await.expect("seed existing format");
+        }
+        let endpoints = Endpoints::from(originals.iter().flatten().map(|disk| disk.endpoint()).collect::<Vec<_>>());
+        let pool = PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 4,
+            endpoints: endpoints.clone(),
+            cmd_line: "returned-during-startup".to_string(),
+            platform: String::new(),
+        };
+        let missing = directory.path().join("disk-1");
+        let parked = directory.path().join("parked-disk-1");
+        tokio::fs::rename(&missing, &parked)
+            .await
+            .expect("remove disk before foundation");
+        let context = Arc::new(InstanceContext::new());
+        context.update_erasure_type(SetupType::DistErasure).await;
+        crate::runtime::sources::initialize_local_disk_maps(
+            &context,
+            EndpointServerPools::from(vec![pool.clone()]),
+            &DiskOption::default(),
+        )
+        .await
+        .expect("register surviving disks");
+        assert!(context.local_disk_map().read().await[&endpoints.as_ref()[1].to_string()].is_none());
+        tokio::fs::rename(&parked, &missing)
+            .await
+            .expect("return disk before initial format load");
+
+        let (mut disks, errors) = init_disks(
+            &context,
+            &endpoints,
+            &DiskOption {
+                cleanup: true,
+                health_check: false,
+            },
+        )
+        .await;
+
+        assert!(errors.iter().all(Option::is_none), "initial disk errors: {errors:?}");
+        let returned = disks[1].clone().expect("returned disk handle");
+        let routes = context.local_disk_map();
+        assert!(Arc::ptr_eq(
+            &returned,
+            routes.read().await[&endpoints.as_ref()[1].to_string()]
+                .as_ref()
+                .expect("returned RPC route")
+        ));
+        let slots = context.local_disk_set_drives();
+        assert!(Arc::ptr_eq(&returned, slots.read().await[0][0][1].as_ref().expect("returned slot")));
+        let loaded = connect_load_init_formats_with_instance_ctx(&context, true, &mut disks, 1, 4, None)
+            .await
+            .expect("first format load succeeds without retry");
+        let sets = crate::core::sets::Sets::new_with_instance_ctx(disks, &pool, &loaded.format, 0, 2, context)
+            .await
+            .expect("construct distributed erasure set");
+        let active = sets.disk_set[0].disks.read().await;
+        assert!(Arc::ptr_eq(&returned, active[1].as_ref().expect("returned drive is immediately active")));
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn init_disks_reopens_a_replaced_registered_root_without_cleaning_staging() {
+        let (directory, originals) = local_disks(1).await;
+        let original = originals[0].clone().expect("original disk");
+        let endpoint = original.endpoint();
+        let context = Arc::new(InstanceContext::new());
+        record_local_disks(&context, vec![original.clone()]).await;
+        let root = original.path();
+        tokio::fs::rename(&root, directory.path().join("retired-root"))
+            .await
+            .expect("replace configured root");
+        let metadata = root.join(RUSTFS_META_BUCKET);
+        tokio::fs::create_dir_all(&metadata)
+            .await
+            .expect("replacement metadata directory");
+        let mut format = FormatV3::new(1, 1);
+        format.erasure.this = format.erasure.sets[0][0];
+        tokio::fs::write(metadata.join(FORMAT_CONFIG_FILE), format.to_json().expect("replacement format"))
+            .await
+            .expect("persist replacement format");
+        let staging = metadata.join("tmp/live-replacement/xl.meta");
+        tokio::fs::create_dir_all(staging.parent().expect("staging parent"))
+            .await
+            .expect("replacement staging directory");
+        tokio::fs::write(&staging, b"replacement staging")
+            .await
+            .expect("seed replacement staging");
+
+        let (disks, errors) = init_disks(
+            &context,
+            &Endpoints::from(vec![endpoint.clone()]),
+            &DiskOption {
+                cleanup: true,
+                health_check: false,
+            },
+        )
+        .await;
+
+        assert!(errors.iter().all(Option::is_none), "replacement initialization errors: {errors:?}");
+        let replacement = disks[0].as_ref().expect("replacement disk");
+        assert!(!Arc::ptr_eq(&original, replacement), "a stale physical root must not be reused");
+        assert_eq!(replacement.get_disk_id().await.expect("replacement identity"), Some(format.erasure.this));
+        assert!(Arc::ptr_eq(
+            replacement,
+            context.local_disk_map().read().await[&endpoint.to_string()]
+                .as_ref()
+                .expect("replacement RPC route")
+        ));
+        assert_eq!(
+            tokio::fs::read(staging).await.expect("preserved replacement staging"),
+            b"replacement staging"
+        );
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn init_disks_clears_stale_registration_when_reopened_root_fails() {
+        use crate::layout::endpoints::{EndpointServerPools, PoolEndpoints};
+
+        for corrupt_replacement in [false, true] {
+            let (directory, original) = local_disks(1).await;
+            let endpoint = original[0].as_ref().expect("original disk").endpoint();
+            let context = Arc::new(InstanceContext::new());
+            let endpoints = Endpoints::from(vec![endpoint.clone()]);
+            let pools = EndpointServerPools::from(vec![PoolEndpoints {
+                legacy: false,
+                set_count: 1,
+                drives_per_set: 1,
+                endpoints: endpoints.clone(),
+                cmd_line: "failed-reopen".to_string(),
+                platform: String::new(),
+            }]);
+            crate::runtime::sources::initialize_local_disk_maps(&context, pools, &DiskOption::default())
+                .await
+                .expect("register original root");
+            let root = std::path::PathBuf::from(endpoint.get_file_path());
+            tokio::fs::rename(&root, directory.path().join("retired-root"))
+                .await
+                .expect("remove registered root");
+            if corrupt_replacement {
+                let metadata = root.join(RUSTFS_META_BUCKET);
+                tokio::fs::create_dir_all(&metadata)
+                    .await
+                    .expect("replacement metadata directory");
+                let nil_identity = FormatV3::new(1, 1).to_json().expect("nil identity format");
+                tokio::fs::write(metadata.join(FORMAT_CONFIG_FILE), nil_identity)
+                    .await
+                    .expect("seed corrupt replacement");
+            }
+
+            let (disks, errors) = init_disks(&context, &endpoints, &DiskOption::default()).await;
+
+            let expected = if corrupt_replacement {
+                DiskError::CorruptedFormat
+            } else {
+                DiskError::DiskNotFound
+            };
+            assert_eq!(errors[0], Some(expected), "the original endpoint error must be preserved");
+            assert!(disks[0].is_none());
+            assert!(
+                context.local_disk_map().read().await[&endpoint.to_string()].is_none(),
+                "stale RPC route must be withdrawn"
+            );
+            assert!(
+                context.local_disk_set_drives().read().await[0][0][0].is_none(),
+                "stale erasure slot must be withdrawn"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn init_disks_preserves_live_staging_on_registered_local_disks() {
         let (temp_dir, disks) = local_disks(1).await;
@@ -1131,6 +1437,43 @@ mod tests {
         assert!(!staging.exists(), "a fresh disk open must still clear stale staging");
     }
 
+    #[tokio::test]
+    async fn init_disks_preserves_registered_raw_endpoint_without_geometry() {
+        let directory = tempfile::tempdir().expect("raw local endpoint fixture");
+        let endpoint = Endpoint::try_from(directory.path().to_str().expect("UTF-8 raw endpoint")).expect("raw endpoint");
+        assert_eq!((endpoint.pool_idx, endpoint.set_idx, endpoint.disk_idx), (-1, -1, -1));
+        let original = crate::disk::new_disk(&endpoint, &DiskOption::default())
+            .await
+            .expect("open raw endpoint");
+        let context = Arc::new(InstanceContext::new());
+        record_local_disks(&context, vec![original.clone()]).await;
+        let staging = directory.path().join(RUSTFS_META_BUCKET).join("tmp/raw-live-staging");
+        tokio::fs::write(&staging, b"live raw staging")
+            .await
+            .expect("seed raw live staging");
+
+        let (opened, errors) = init_disks(
+            &context,
+            &Endpoints::from(vec![endpoint.clone()]),
+            &DiskOption {
+                cleanup: true,
+                health_check: false,
+            },
+        )
+        .await;
+
+        assert!(errors.iter().all(Option::is_none), "raw endpoint errors: {errors:?}");
+        assert!(Arc::ptr_eq(opened[0].as_ref().expect("raw handle remains available"), &original));
+        assert!(Arc::ptr_eq(
+            context.local_disk_map().read().await[&endpoint.to_string()]
+                .as_ref()
+                .expect("raw registration remains available"),
+            &original,
+        ));
+        assert!(context.local_disk_set_drives().read().await.is_empty());
+        assert_eq!(tokio::fs::read(staging).await.expect("preserved raw staging"), b"live raw staging");
+    }
+
     #[tokio::test]
     async fn init_disks_rejects_registered_disk_topology_mismatch() {
         let (_temp_dir, disks) = local_disks(1).await;

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

@@ -69,7 +69,7 @@ async fn auto_replacement_path_requires_a_non_root_mount() {
                 },
             )
             .await,
-            Err(DiskError::VolumeNotFound)
+            Err(DiskError::DiskNotFound)
         ),
         "a missing replacement path must be rejected before admission"
     );

docs/operations/drive-timeout-tuning.md

@@ -14,6 +14,16 @@ If you are debugging a listing that returns *fewer* keys than the bucket holds,
 read [Listing truncation and the walk stall budget](#listing-truncation-and-the-walk-stall-budget)
 first — that is the case this document exists for.
 
+## Startup with unavailable drives
+
+A missing local endpoint stays in its configured erasure slot while healthy local drives initialize. Existing format, object, and lock quorums determine service availability. Startup retries reopen missing local slots, and the running endpoint monitor reconnects drives that return. Quorum waits remain cancellable.
+
+Local preparation uses bounded concurrency and performs filesystem work outside the local registry locks. Full registry initialization publishes a validated batch in a short critical section, preserving the previous registry on failure or cancellation. Reopening an unavailable registered endpoint withdraws its stale handle. Root identity checks gate reuse of registered handles.
+
+On Linux, a URL-style erasure endpoint with no format on the system filesystem is kept offline before write probes, internal directory creation, or startup cleanup. Restore its data mount at the configured path. This also applies to single-host URL layouts that mix a system-filesystem directory with independent data drives. Path-style single-drive deployments retain their directory initialization behavior. The existing explicit `RUSTFS_UNSAFE_BYPASS_DISK_CHECK` override and `MINIO_CI` alias retain directory-test behavior.
+
+Existing formats retain their identity and quorum validation. A legacy format on the system filesystem must pass bounded decoding and layout/slot checks before startup writes. This legacy exception has a 1 MiB admission limit, followed by the existing topology-derived migration limit. Legacy formats on independent data filesystems retain the existing migration limit. Detecting a missing mount that exposes an arbitrary non-system parent filesystem requires additional expected-mount provenance.
+
 ## Idle local drives
 
 `RUSTFS_DRIVE_ACTIVE_MONITORING=false` also suppresses the endpoint monitor's

rustfs/src/server/http.rs

@@ -6802,6 +6802,8 @@ mod tests {
         temp_env::async_with_vars(
             [
                 ("RUSTFS_INTERNODE_PREWARM", Some("false")),
+                // The transport fixture uses the existing directory deployment test mode.
+                ("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", Some("true")),
                 ("RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES", Some("0")),
             ],
             async {

scripts/error-other-format-baseline.txt

@@ -42,7 +42,6 @@
 1|crates/ecstore/src/layout/set_layout.rs
 19|crates/ecstore/src/object_api/readers.rs
 1|crates/ecstore/src/object_api/types.rs
-3|crates/ecstore/src/runtime/sources.rs
 4|crates/ecstore/src/services/batch_processor.rs
 12|crates/ecstore/src/services/notification_sys.rs
 15|crates/ecstore/src/services/rebalance/control.rs