fix(storage): retain resources during native format probes

crates/ecstore/src/disk/disk_store.rs

@@ -2613,7 +2613,7 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn is_online(&self) -> bool {
-        let Ok(Some(disk_id)) = self.disk.get_disk_id().await else {
+        let Ok(Some(disk_id)) = self.get_disk_id().await else {
             return false;
         };
 
@@ -2646,7 +2646,7 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn get_disk_id(&self) -> Result<Option<Uuid>> {
-        self.disk.get_disk_id().await
+        self.with_native_resources(self.disk.get_disk_id()).await?
     }
 
     async fn set_disk_id(&self, id: Option<Uuid>) -> Result<()> {
@@ -3911,6 +3911,100 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn local_disk_resource_identity_probes_keep_only_their_native_device_after_cancel() {
+        use rustfs_concurrency::adaptive::{
+            AdaptiveAdmission, FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure,
+            permit_set::AdaptivePermitSetOwner,
+        };
+        use std::num::NonZeroU16;
+
+        for read in [false, true] {
+            for online in [false, true] {
+                let dir = tempfile::tempdir().expect("format probe fixture");
+                let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("fixture path")).expect("local endpoint");
+                endpoint.set_pool_index(0);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(0);
+                let meta = dir.path().join(RUSTFS_META_BUCKET);
+                tokio::fs::create_dir_all(&meta).await.expect("metadata directory");
+                let mut format = FormatV3::new(1, 1);
+                format.erasure.this = format.erasure.sets[0][0];
+                tokio::fs::write(meta.join(FORMAT_CONFIG_FILE), format.to_json().expect("format bytes"))
+                    .await
+                    .expect("write format");
+                let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("formatted disk"));
+                {
+                    let mut cache = disk.format_info.write().await;
+                    cache.last_check = None;
+                    if read {
+                        cache.file_info = None;
+                    }
+                }
+                let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+                let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+                disk.install_format_probe_hook(read, move || {
+                    let _ = started_tx.send(());
+                    let _ = release_rx.blocking_recv();
+                });
+                let wrapper = Arc::new(LocalDiskWrapper::new(disk, false));
+                let a = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+                let b = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+                let shared = AdaptivePermitSetOwner::new(NonZeroU16::MIN, vec![a.clone(), b.clone()]).expect("catalog");
+                wrapper
+                    .physical_admission
+                    .set_test_shared_capacity(shared.clone(), vec![a.clone()])
+                    .expect("native device A");
+                let node = NodeAdmission::new(NonZeroU16::MIN);
+                node.update(
+                    &mut FeedbackController::new(),
+                    Some(FeedbackObservation {
+                        generation: 1,
+                        max_concurrency: 1,
+                        pressure: ResourcePressure::Unknown,
+                        completed_units: None,
+                    }),
+                );
+                let parent = shared.try_acquire_with_node(&node).expect("planned devices");
+                let task = tokio::spawn({
+                    let wrapper = wrapper.clone();
+                    async move {
+                        crate::disk::with_background_resource_set(parent.resources().clone(), async {
+                            if online {
+                                assert!(wrapper.is_online().await);
+                            } else {
+                                wrapper.get_disk_id().await.expect("disk identity");
+                            }
+                        })
+                        .await;
+                        parent.complete();
+                    }
+                });
+                tokio::time::timeout(Duration::from_secs(5), started_rx)
+                    .await
+                    .expect("native probe starts")
+                    .expect("native signal");
+                task.abort();
+                assert!(task.await.expect_err("cancel waiter").is_cancelled());
+                let held = [node.snapshot().active, a.snapshot().active, b.snapshot().active];
+                let healthy_available = b.try_acquire().is_some();
+                release_tx.send(()).expect("release native probe before assertions");
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    while a.snapshot().active != 0 || b.snapshot().active != 0 {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("native probe releases its device");
+                assert_eq!(held, [0, 1, 0], "read={read}, online={online}");
+                assert!(healthy_available);
+                assert_eq!([node.completed(), a.completed(), b.completed()], [Some(0), Some(0), Some(0)]);
+                assert_eq!(wrapper.get_disk_id().await.expect("format remains readable"), Some(format.erasure.this));
+                assert!(wrapper.is_online().await);
+            }
+        }
+    }
+
     #[tokio::test]
     async fn local_disk_resource_projection_releases_unrelated_devices_after_native_cancellation() {
         use crate::disk::io_admission::spawn_blocking_with_disk_io_permit;
@@ -4106,6 +4200,10 @@ mod tests {
                 "auxiliary admission must return immediately under storage locks"
             );
             drop(read);
+            assert!(
+                matches!(wrapper.get_disk_id().await, Err(DiskError::Io(error)) if error.kind() == io::ErrorKind::WouldBlock)
+            );
+            assert!(!wrapper.is_online().await, "identity checks cannot bypass closed physical capacity");
             assert_eq!([a.snapshot().active, b.snapshot().active, node.snapshot().active], [1, 0, 1]);
             assert_eq!([a.completed(), b.completed(), node.completed()], [Some(0), Some(0), Some(0)]);
             b.set_limit(1);

crates/ecstore/src/disk/local.rs

@@ -3224,6 +3224,20 @@ fn log_startup_disk_error(stage: &str, path: &Path, err: &DiskError) {
     );
 }
 
+#[cfg(test)]
+type FormatProbeHook = Box<dyn FnOnce() + Send>;
+#[cfg(test)]
+static FORMAT_PROBE_HOOKS: std::sync::LazyLock<ParkingLotMutex<HashMap<(PathBuf, bool), FormatProbeHook>>> =
+    std::sync::LazyLock::new(|| ParkingLotMutex::new(HashMap::new()));
+
+#[cfg(test)]
+fn run_format_probe_hook(path: &Path, read: bool) {
+    let hook = FORMAT_PROBE_HOOKS.lock().remove(&(path.to_path_buf(), read));
+    if let Some(hook) = hook {
+        hook();
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct FormatInfo {
     pub id: Option<Uuid>,
@@ -6050,6 +6064,13 @@ fn mount_id_from_mountinfo_contents(mountinfo: &str, path: &Path) -> Option<u64>
 }
 
 impl LocalDisk {
+    #[cfg(test)]
+    pub(crate) fn install_format_probe_hook(&self, read: bool, hook: impl FnOnce() + Send + 'static) {
+        FORMAT_PROBE_HOOKS
+            .lock()
+            .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())
@@ -6689,9 +6710,39 @@ impl LocalDisk {
 
     #[tracing::instrument(level = "trace", skip_all)]
     async fn check_format_json(&self) -> Result<Metadata> {
-        let md = fs::metadata(&self.io_format_path).await.map_err(to_unformatted_disk_error)?;
-        Ok(md)
+        let path = self.io_format_path.clone();
+        #[cfg(target_os = "linux")]
+        let mount_lease = self.mount_lease.clone();
+        Ok(spawn_blocking_with_disk_io_permit(None, move || {
+            // The proc-fd path must survive cancellation and disk retirement.
+            #[cfg(target_os = "linux")]
+            let _mount_lease = mount_lease;
+            #[cfg(test)]
+            run_format_probe_hook(&path, false);
+            std::fs::metadata(path)
+        })
+        .await
+        .map_err(DiskError::from)?
+        .map_err(to_unformatted_disk_error)?)
     }
+
+    async fn read_format_json(&self) -> Result<Vec<u8>> {
+        let path = self.io_format_path.clone();
+        #[cfg(target_os = "linux")]
+        let mount_lease = self.mount_lease.clone();
+        Ok(spawn_blocking_with_disk_io_permit(None, move || {
+            // The proc-fd path must survive cancellation and disk retirement.
+            #[cfg(target_os = "linux")]
+            let _mount_lease = mount_lease;
+            #[cfg(test)]
+            run_format_probe_hook(&path, true);
+            std::fs::read(path)
+        })
+        .await
+        .map_err(DiskError::from)?
+        .map_err(to_unformatted_disk_error)?)
+    }
+
     async fn make_meta_volumes(&self) -> Result<()> {
         let buckets = format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}");
         let multipart = format!("{}/{}", RUSTFS_META_BUCKET, "multipart");
@@ -11282,7 +11333,7 @@ impl DiskAPI for LocalDisk {
 
         debug!("get_disk_id: read format.json");
 
-        let b = fs::read(&self.io_format_path).await.map_err(to_unformatted_disk_error)?;
+        let b = self.read_format_json().await?;
 
         let fm = FormatV3::try_from(b.as_slice()).map_err(|e| {
             warn!(
@@ -22809,6 +22860,58 @@ mod test {
         assert!(result.is_err(), "zero timeout must leave stalled reads pending instead of failing");
     }
 
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn resource_format_probe_retains_mount_lease_after_disk_retirement() {
+        for read in [false, true] {
+            let dir = tempfile::tempdir().expect("retired format probe fixture");
+            let endpoint = Endpoint::try_from(dir.path().to_str().expect("fixture path")).expect("local endpoint");
+            let mut disk = LocalDisk::new(&endpoint, false).await.expect("local disk");
+            // The cleanup monitor owns the same lease. Wait for its receiver to
+            // close so it cannot mask an unowned native probe after disk drop.
+            let exit = disk.exit_signal.take().expect("cleanup monitor exit sender");
+            exit.send(()).expect("stop this fixture's cleanup monitor");
+            tokio::time::timeout(Duration::from_secs(5), async {
+                while exit.receiver_count() != 0 {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("cleanup monitor stopped");
+            drop(exit);
+            let lease = Arc::downgrade(&disk.mount_lease);
+            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+            let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+            disk.install_format_probe_hook(read, move || {
+                let _ = started_tx.send(());
+                let _ = release_rx.blocking_recv();
+            });
+            let task = tokio::spawn(async move {
+                if read {
+                    disk.read_format_json().await.map(|_| ())
+                } else {
+                    disk.check_format_json().await.map(|_| ())
+                }
+            });
+            tokio::time::timeout(Duration::from_secs(5), started_rx)
+                .await
+                .expect("native probe starts")
+                .expect("native signal");
+            task.abort();
+            assert!(task.await.expect_err("retire disk with cancelled probe").is_cancelled());
+            let retained = lease.strong_count() > 0;
+            release_tx.send(()).expect("release probe before asserting");
+            tokio::time::timeout(Duration::from_secs(5), async {
+                while lease.strong_count() != 0 {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("native probe releases the mount lease");
+            assert!(retained, "read={read}: the proc-fd path must stay pinned until native I/O returns");
+        }
+    }
+
     #[tokio::test]
     async fn test_get_disk_id_invalidates_cache_after_format_removal() {
         use crate::disk::FORMAT_CONFIG_FILE;