diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 5883a51c0..b251cfb7f 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/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> { - self.disk.get_disk_id().await + self.with_native_resources(self.disk.get_disk_id()).await? } async fn set_disk_id(&self, id: Option) -> 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); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index bb47d56e0..760b7ae48 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/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; +#[cfg(test)] +static FORMAT_PROBE_HOOKS: std::sync::LazyLock>> = + 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, @@ -6050,6 +6064,13 @@ fn mount_id_from_mountinfo_contents(mountinfo: &str, path: &Path) -> Option } 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 { Ok(root.to_path_buf()) @@ -6689,9 +6710,39 @@ impl LocalDisk { #[tracing::instrument(level = "trace", skip_all)] async fn check_format_json(&self) -> Result { - 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> { + 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;