diff --git a/crates/concurrency/src/adaptive/permit_set.rs b/crates/concurrency/src/adaptive/permit_set.rs index 23dce1e68..f5547a97e 100644 --- a/crates/concurrency/src/adaptive/permit_set.rs +++ b/crates/concurrency/src/adaptive/permit_set.rs @@ -215,6 +215,15 @@ impl AdaptivePermitSetOwner { })) } + /// Start a node-scoped operation with no planned local leaves. Auxiliary + /// local I/O may still extend it through this immutable resource catalog. + pub fn empty_selection(self: &Arc) -> AdaptiveResourceSelection { + AdaptiveResourceSelection { + owner: self.clone(), + selected: ResourceMask::empty(), + } + } + /// Resolve an attachment's leaves against this immutable owner outside the /// object path. Repeated aliases select the same unit; foreign owners fail. pub fn select(self: &Arc, owners: &[Arc]) -> Result { diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 1751e8155..3a6284f75 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -442,7 +442,7 @@ pub mod disk { ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, is_quota_mutation_fence_path, new_disk, validate_batch_read_version_item_count, }; - pub use crate::disk::{PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler}; + pub use crate::disk::{PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler, PhysicalResourcePlan}; pub use bytes::Bytes; pub use endpoint::Endpoint; pub use error::DiskError; diff --git a/crates/ecstore/src/disk/io_admission.rs b/crates/ecstore/src/disk/io_admission.rs index c588a8db8..f3594b837 100644 --- a/crates/ecstore/src/disk/io_admission.rs +++ b/crates/ecstore/src/disk/io_admission.rs @@ -18,7 +18,10 @@ mod physical_identity; mod physical_owners; mod physical_sampler; -pub use physical_sampler::{PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler}; +pub(crate) use physical_sampler::PhysicalHealBinding; +pub use physical_sampler::{ + PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler, PhysicalResourcePlan, +}; use std::collections::VecDeque; use std::future::Future; diff --git a/crates/ecstore/src/disk/io_admission/physical_sampler.rs b/crates/ecstore/src/disk/io_admission/physical_sampler.rs index 3dea7532c..fd0351d36 100644 --- a/crates/ecstore/src/disk/io_admission/physical_sampler.rs +++ b/crates/ecstore/src/disk/io_admission/physical_sampler.rs @@ -61,7 +61,7 @@ impl PhysicalDiskBinding { identities: Vec::new(), owner, selected, - topology: Arc::new(PhysicalTopology::new()), + topology: Arc::new(PhysicalTopology::new(shared)), }); plan.topology.renew(); self.0.store(Some(plan)); @@ -91,13 +91,15 @@ pub struct PhysicalDiskPlan { } struct PhysicalTopology { + catalog: Arc, origin: Instant, valid_until_millis: AtomicU64, } impl PhysicalTopology { - fn new() -> Self { + fn new(catalog: Arc) -> Self { Self { + catalog, origin: Instant::now(), valid_until_millis: AtomicU64::new(0), } @@ -136,6 +138,42 @@ impl PhysicalDiskPlan { self.owner.try_capture_or_acquire_from(admitted, &self.selected) } + pub fn resource_plan(&self) -> PhysicalResourcePlan { + PhysicalResourcePlan { + selected: self.selected.clone(), + topology: self.topology.clone(), + } + } + + pub fn try_acquire(&self, node: &Arc) -> Option { + self.resource_plan().try_acquire(node) + } + + pub async fn acquire(&self, node: &Arc) -> Option { + self.resource_plan().acquire(node).await + } +} + +/// Cached local participation in a Heal operation. Combining plans is bounded +/// and allocation-free; sysfs discovery and disk locks stay in the sampler. +#[derive(Clone)] +pub struct PhysicalResourcePlan { + selected: AdaptiveResourceSelection, + topology: Arc, +} + +impl PhysicalResourcePlan { + pub fn is_current(&self) -> bool { + self.topology.is_current() + } + + pub(crate) fn union_with(&mut self, other: &Self) -> Result<(), CaptureError> { + if !Arc::ptr_eq(&self.topology, &other.topology) || !other.is_current() { + return Err(CaptureError::ResourceOwner); + } + self.selected.union_with(&other.selected) + } + /// Never wait under a storage lock. A refresh during acquisition rejects /// the old plan before its future can issue I/O. pub fn try_acquire(&self, node: &Arc) -> Option { @@ -167,6 +205,37 @@ impl PhysicalDiskPlan { } } +#[derive(Default)] +pub(crate) struct PhysicalHealBinding(ArcSwapOption); + +impl std::fmt::Debug for PhysicalHealBinding { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PhysicalHealBinding") + .field("ready", &self.plan().is_some()) + .finish() + } +} + +impl PhysicalHealBinding { + #[cfg(test)] + pub(crate) fn set_test_plan(&self, plan: PhysicalResourcePlan) { + self.0.store(Some(Arc::new(plan))); + } + + pub(crate) fn plan(&self) -> Option { + self.0 + .load() + .as_ref() + .filter(|plan| plan.is_current()) + .map(|plan| (**plan).clone()) + } +} + +struct HealSetSample { + binding: Arc, + disks: Option>>, +} + struct Target { disk: Arc, binding: Arc, @@ -175,6 +244,7 @@ struct Target { /// Collected handles contain no storage locks. Move this snapshot into the /// existing blocking sampler; only `publish` may renew admission leases. pub struct PhysicalDiskSample { + heal_sets: Vec, targets: Vec, resolved: Vec, devices: BTreeMap>, @@ -242,13 +312,23 @@ impl PhysicalDiskSample { /// cannot hold the sampler indefinitely or renew stale capacity. pub fn collect(store: &ECStore) -> Self { let mut targets = Vec::new(); + let mut heal_sets = Vec::new(); let mut seen = BTreeSet::new(); for pool in &store.pools { for set in &pool.disk_set { - let Ok(disks) = set.disks.try_read() else { continue }; + let mut heal_set = HealSetSample { + binding: set.physical_heal_admission.clone(), + disks: None, + }; + let Ok(disks) = set.disks.try_read() else { + heal_sets.push(heal_set); + continue; + }; + let mut bindings = Vec::new(); for disk in disks.iter().flatten() { let Disk::Local(local) = disk.as_ref() else { continue }; let binding = local.physical_admission(); + bindings.push(binding.clone()); if targets.len() < MAX_LOGICAL_DISKS && seen.insert(Arc::as_ptr(&binding)) { targets.push(Target { disk: local.get_disk(), @@ -256,9 +336,12 @@ impl PhysicalDiskSample { }); } } + heal_set.disks = Some(bindings); + heal_sets.push(heal_set); } } Self { + heal_sets, targets, resolved: Vec::new(), devices: BTreeMap::new(), @@ -398,7 +481,7 @@ impl PhysicalDiskSampler { let next = (|| { let all = AdaptivePermitSetOwner::new(MAXIMUM, self.devices.values().map(|device| device.gate.clone()).collect()) .ok()?; - let topology = Arc::new(PhysicalTopology::new()); + let topology = Arc::new(PhysicalTopology::new(all.clone())); let mut plans = Vec::with_capacity(prepared.len()); for (binding, identities, owners) in &prepared { let gates = owners.iter().map(|owner| owner.gate().clone()).collect::>(); @@ -431,6 +514,20 @@ impl PhysicalDiskSampler { if let Some(topology) = &self.topology { topology.renew(); } + for set in sample.heal_sets { + let plan = (|| { + let topology = self.topology.as_ref().filter(|topology| topology.is_current())?; + let mut plan = PhysicalResourcePlan { + selected: topology.catalog.empty_selection(), + topology: topology.clone(), + }; + for binding in set.disks? { + plan.union_with(&binding.plan()?.resource_plan()).ok()?; + } + Some(Arc::new(plan)) + })(); + set.binding.0.store(plan); + } for (index, (identity, device)) in self.devices.iter_mut().enumerate() { let observation = sample.devices.get(identity).copied().flatten().filter(|sample| { fresh @@ -556,6 +653,7 @@ mod tests { fn sample(bindings: &[Arc], id: PhysicalDeviceIdentity) -> PhysicalDiskSample { let now = Instant::now(); PhysicalDiskSample { + heal_sets: Vec::new(), targets: Vec::new(), resolved: bindings .iter() @@ -593,6 +691,85 @@ mod tests { sample } + #[test] + fn heal_plans_select_only_participating_sets_and_reject_unknown_members() { + let a = Arc::new(PhysicalDiskBinding::default()); + let b = Arc::new(PhysicalDiskBinding::default()); + let first = Arc::new(PhysicalHealBinding::default()); + let second = Arc::new(PhysicalHealBinding::default()); + let remote_only = Arc::new(PhysicalHealBinding::default()); + let mut sampler = PhysicalDiskSampler::default(); + let mut sample = two_disks(&a, &b); + sample.heal_sets = vec![ + HealSetSample { + binding: first.clone(), + disks: Some(vec![a.clone(), a.clone()]), + }, + HealSetSample { + binding: second.clone(), + disks: Some(vec![b.clone()]), + }, + HealSetSample { + binding: remote_only.clone(), + disks: Some(Vec::new()), + }, + ]; + sampler.publish(sample, ResourcePressure::Stable); + let node = node(); + let a_gate = sampler.devices.get(&identity(1)).expect("A owner").gate.clone(); + let b_gate = sampler + .devices + .get(&PhysicalDeviceIdentity { minor: 1, ..identity(1) }) + .expect("B owner") + .gate + .clone(); + let first_plan = first.plan().expect("set A"); + let a_permit = first_plan.try_acquire(&node).expect("A admitted"); + assert_eq!((a_gate.snapshot().active, b_gate.snapshot().active), (1, 0)); + let a_native = a + .plan() + .expect("A binding") + .capture_native(a_permit.resources()) + .expect("A native tail"); + drop(a_permit); + let b_permit = second + .plan() + .expect("set B") + .try_acquire(&node) + .expect("B independent of full A"); + assert_eq!((a_gate.snapshot().active, b_gate.snapshot().active), (1, 1)); + drop((a_native, b_permit)); + let empty = remote_only + .plan() + .expect("known remote-only set") + .try_acquire(&node) + .expect("node scope"); + assert_eq!((a_gate.snapshot().active, b_gate.snapshot().active), (0, 0)); + empty.complete(); + let mut both = first_plan.clone(); + both.union_with(&second.plan().expect("B plan")).expect("common catalog"); + let permit = both.try_acquire(&node).expect("union"); + assert_eq!((a_gate.snapshot().active, b_gate.snapshot().active), (1, 1)); + permit.complete(); + assert_eq!((node.completed(), a_gate.completed(), b_gate.completed()), (Some(2), Some(1), Some(1))); + let mut sample = two_disks(&a, &b); + sample.heal_sets = vec![HealSetSample { + binding: first.clone(), + disks: None, + }]; + sampler.publish(sample, ResourcePressure::Stable); + assert!(first.plan().is_none(), "a busy set cannot reuse its prior membership"); + let mut sample = two_disks(&a, &b); + sample.resolved[0].identities = None; + sample.heal_sets = vec![HealSetSample { + binding: first.clone(), + disks: Some(vec![a.clone()]), + }]; + sampler.publish(sample, ResourcePressure::Stable); + assert!(first.plan().is_none(), "unknown local disks never become empty selections"); + assert!(first_plan.try_acquire(&node).is_none(), "old plans reject a changed generation"); + } + #[test] fn native_tails_cannot_exhaust_shared_carriers_needed_by_another_disk() { let a = Arc::new(PhysicalDiskBinding::default()); diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index edb415739..417cdb9bf 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -32,12 +32,12 @@ pub mod os; #[cfg(test)] pub(crate) use io_admission::DiskIoClass; pub use io_admission::{ - PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler, spawn_blocking_with_resource_ownership, - with_background_disk_io, with_background_resource_set, with_background_resource_unit, + PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler, PhysicalResourcePlan, + spawn_blocking_with_resource_ownership, with_background_disk_io, with_background_resource_set, with_background_resource_unit, }; pub(crate) use io_admission::{ - acquire_context_disk_io_permit, current_disk_io_context, current_resource_unit, with_optional_disk_io_context, - with_optional_resource_unit, + PhysicalHealBinding, acquire_context_disk_io_permit, current_disk_io_context, current_resource_unit, + with_optional_disk_io_context, with_optional_resource_unit, }; pub const RUSTFS_META_BUCKET: &str = ".rustfs.sys"; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 73fba73a3..dc30d535e 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -4466,6 +4466,7 @@ fn resolve_tiered_decommission_write_quorum_result( pub struct SetDisks { pub locker_owner: String, pub disks: Arc>>>, + pub(crate) physical_heal_admission: Arc, pub set_endpoints: Vec, pub set_drive_count: usize, pub default_parity_count: usize, @@ -5427,6 +5428,7 @@ impl SetDisks { Arc::new(SetDisks { locker_owner, disks, + physical_heal_admission: Arc::default(), set_drive_count, default_parity_count, set_index, diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index 03bb192d5..e637d6d60 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -599,6 +599,33 @@ impl ECStore { Ok((write_state, pool_guard, rebalance_guard, pool_meta, rebalance_meta)) } + /// Snapshot only the object sets participating in this request. The sampler + /// resolves their physical leaves before Heal enters namespace/cache locks. + pub fn physical_heal_plan(&self, object: &str, opts: &HealOpts) -> Result> { + if let Some(index) = opts.pool { + if index >= self.pools.len() { + return Err(invalid_heal_pool_index(index, self.pools.len())); + } + } + let object = rustfs_utils::path::encode_dir_object_ref(object); + let mut selected: Option = None; + for (index, pool) in self.pools.iter().enumerate() { + if opts.pool.is_some_and(|requested| requested != index) { + continue; + } + let set = pool.get_disks_for_heal_object(&object, opts)?; + let Some(plan) = set.physical_heal_admission.plan() else { return Ok(None) }; + if let Some(selected) = &mut selected { + if selected.union_with(&plan).is_err() { + return Ok(None); + } + } else { + selected = Some(plan); + } + } + Ok(selected) + } + fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result>> { match opts.pool { Some(pool_idx) => Ok(vec![ @@ -1554,21 +1581,25 @@ mod tests { } } - async fn minimal_heal_pool(pool_idx: usize) -> Arc { - let format = FormatV3::new(1, 1); - let endpoint_url = format!("http://127.0.0.1:{}/data", 19000 + pool_idx); - let mut endpoint = Endpoint::try_from(endpoint_url.as_str()).expect("endpoint should parse"); - endpoint.set_pool_index(pool_idx); - endpoint.set_set_index(0); - endpoint.set_disk_index(0); - + async fn minimal_heal_pool(pool_idx: usize, set_count: usize) -> Arc { + let format = FormatV3::new(set_count, 1); + let endpoints = (0..set_count) + .map(|set_idx| { + let endpoint_url = format!("http://127.0.0.1:{}/data", 19000 + pool_idx + set_idx * 2); + let mut endpoint = Endpoint::try_from(endpoint_url.as_str()).expect("endpoint should parse"); + endpoint.set_pool_index(pool_idx); + endpoint.set_set_index(set_idx); + endpoint.set_disk_index(0); + endpoint + }) + .collect::>(); Sets::new( - vec![None], + vec![None; set_count], &PoolEndpoints { legacy: false, - set_count: 1, + set_count, drives_per_set: 1, - endpoints: Endpoints::from(vec![endpoint]), + endpoints: Endpoints::from(endpoints), cmd_line: String::new(), platform: String::new(), }, @@ -2033,7 +2064,7 @@ mod tests { ECStore { id: Uuid::new_v4(), disk_map: HashMap::new(), - pools: vec![minimal_heal_pool(0).await, minimal_heal_pool(1).await], + pools: vec![minimal_heal_pool(0, 1).await, minimal_heal_pool(1, 1).await], peer_sys: S3PeerSys { clients: Vec::new(), pools_count: 2, @@ -2048,6 +2079,88 @@ mod tests { } } + #[tokio::test] + async fn physical_heal_plan_uses_the_real_pool_set_and_directory_placement() { + use crate::disk::PhysicalDiskBinding; + use rustfs_concurrency::adaptive::{ + AdaptiveAdmission, FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure, + }; + use std::num::NonZeroU16; + + let mut store = minimal_heal_store().await; + store.pools[0] = minimal_heal_pool(0, 2).await; + let pool = &store.pools[0]; + let a = AdaptiveAdmission::new(NonZeroU16::MIN, 1); + let b = AdaptiveAdmission::new(NonZeroU16::MIN, 1); + for (set, gate) in pool.disk_set.iter().zip([&a, &b]) { + let binding = PhysicalDiskBinding::default(); + binding.set_test_capacity(vec![gate.clone()]).expect("physical fixture"); + set.physical_heal_admission + .set_test_plan(binding.plan().expect("live").resource_plan()); + } + 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 directory = (0..100) + .map(|n| format!("dir{n}/")) + .find(|name| { + store.pools[0].get_disks_by_key(name).set_index + != store.pools[0].get_disks_by_key(&encode_dir_object(name)).set_index + }) + .expect("different directory placement"); + for object in ["plain-object", directory.as_str()] { + for set in [None, Some(0), Some(1)] { + let opts = HealOpts { + pool: Some(0), + set, + ..Default::default() + }; + let expected = store.pools[0] + .get_disks_for_heal_object(&encode_dir_object(object), &opts) + .expect("placement") + .set_index; + let permit = store + .physical_heal_plan(object, &opts) + .expect("valid selection") + .expect("known set") + .try_acquire(&node) + .expect("physical admission"); + assert_eq!((a.snapshot().active, b.snapshot().active), if expected == 0 { (1, 0) } else { (0, 1) }); + drop(permit); + } + } + assert!( + store + .physical_heal_plan("object", &HealOpts::default()) + .expect("valid all pools") + .is_none(), + "an undiscovered participating pool cannot be skipped" + ); + for opts in [ + HealOpts { + pool: Some(2), + ..Default::default() + }, + HealOpts { + pool: Some(0), + set: Some(2), + ..Default::default() + }, + ] { + assert!(matches!( + store.physical_heal_plan("object", &opts), + Err(StorageError::InvalidArgument(..)) + )); + } + } + async fn remove_pool_meta_shard(store: &ECStore, pool_idx: usize) -> DiskStore { let target_set = store.pools[pool_idx].get_disks_by_key(POOL_META_NAME); let missing_disk = target_set.disks.read().await[0] diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index b11c185a5..14bfc78a3 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -27,7 +27,10 @@ use uuid::Uuid; use super::outcome::{HealObjectDisposition, HealObjectIdentity, HealObjectKind, HealObjectReceipt}; use super::progress::stable_generation; pub use super::replacement_execution::ReplacementExecution; -use super::storage_api::owner::{EcstoreHealLifecycleExpiryContext, ecstore_load_admin_data_usage_from_backend_cached}; +use super::storage_api::owner::{ + EcstoreHealLifecycleExpiryContext, PhysicalResourcePlan, ecstore_load_admin_data_usage_from_backend_cached, + ecstore_with_background_resource_set, +}; use super::storage_api::storage::{ BucketInfo, BucketOperations, DiskSetSelector, EcstoreHealObjectStorageResult, HealOperations as _, ListOperations as _, ObjectIO as _, ObjectOperations as _, StorageAdminApi, @@ -753,6 +756,7 @@ pub trait HealStorageAPI: Send + Sync { // locks. No-lock callers may already own a namespace lock: never park them. async fn run_background_object( admission: Option<&Arc>, + physical_plan: impl FnOnce() -> Result>, opts: &HealOpts, completed: impl FnOnce(&T) -> bool, operation: impl std::future::Future>, @@ -760,15 +764,17 @@ async fn run_background_object( let Some(admission) = admission else { return operation.await; }; + let plan = physical_plan()?.ok_or_else(|| Error::transient_skip("physical resource discovery unavailable"))?; let permit = if opts.no_lock { - admission.try_acquire() + plan.try_acquire(admission) } else { - tokio::time::timeout(std::time::Duration::from_millis(250), admission.acquire()) + tokio::time::timeout(std::time::Duration::from_millis(250), plan.acquire(admission)) .await .ok() + .flatten() } - .ok_or_else(|| Error::transient_skip("node resource admission deferred"))?; - let result = operation.await; + .ok_or_else(|| Error::transient_skip("node or physical resource admission deferred"))?; + let result = ecstore_with_background_resource_set(permit.resources().clone(), operation).await; if !opts.dry_run && result.as_ref().is_ok_and(completed) { permit.complete(); } @@ -1079,6 +1085,7 @@ impl HealStorageAPI for ECStoreHealStorage { ) -> Result { run_background_object( self.admission.as_ref(), + || self.ecstore.physical_heal_plan(object, opts).map_err(Error::Storage), opts, |result: &HealStorageObjectResult| result.error.is_none(), async { @@ -1105,6 +1112,7 @@ impl HealStorageAPI for ECStoreHealStorage { ) -> Result { run_background_object( self.admission.as_ref(), + || self.ecstore.physical_heal_plan(object, opts).map_err(Error::Storage), opts, |result: &HealStorageObjectResult| result.error.is_none(), async { @@ -1549,6 +1557,7 @@ impl HealStorageAPI for ECStoreHealStorage { ) -> Result<(HealResultItem, Option)> { run_background_object( self.admission.as_ref(), + || self.ecstore.physical_heal_plan(object, opts).map_err(Error::Storage), opts, |result: &(HealResultItem, Option)| result.1.is_none(), async { @@ -1590,6 +1599,7 @@ impl HealStorageAPI for ECStoreHealStorage { ) -> Result { run_background_object( self.admission.as_ref(), + || self.ecstore.physical_heal_plan(object, opts).map_err(Error::Storage), opts, |result: &HealStorageObjectResult| result.error.is_none(), async { @@ -2228,6 +2238,79 @@ mod tests { gate } + fn physical_plan() -> crate::Result> { + let binding = super::super::storage_api::owner::PhysicalDiskBinding::default(); + binding + .set_test_capacity(vec![rustfs_concurrency::adaptive::AdaptiveAdmission::new(NonZeroU16::MIN, 1)]) + .expect("synthetic physical capacity"); + Ok(Some(binding.plan().expect("live physical binding").resource_plan())) + } + + #[tokio::test] + async fn object_admission_releases_node_and_retains_only_a_running_native_disk() { + use super::super::storage_api::owner::{PhysicalDiskBinding, ecstore_spawn_blocking_with_resource_ownership}; + let node = admission(true); + let disk = rustfs_concurrency::adaptive::AdaptiveAdmission::new(NonZeroU16::MIN, 1); + let binding = PhysicalDiskBinding::default(); + binding.set_test_capacity(vec![disk.clone()]).expect("physical binding"); + let plan = binding.plan().expect("physical plan").resource_plan(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let handle = tokio::spawn({ + let node = node.clone(); + async move { + super::run_background_object(Some(&node), || Ok(Some(plan)), &super::HealOpts::default(), |_: &()| true, async { + ecstore_spawn_blocking_with_resource_ownership(move || { + started_tx.send(()).expect("native started"); + release_rx.recv().expect("release native"); + }) + .await + .expect("native job"); + Ok(()) + }) + .await + } + }); + started_rx.await.expect("native dispatched"); + assert_eq!((node.snapshot().active, disk.snapshot().active), (1, 1)); + handle.abort(); + assert!(handle.await.expect_err("waiter aborted").is_cancelled()); + assert_eq!((node.snapshot().active, disk.snapshot().active), (0, 1)); + assert_eq!((node.completed(), disk.completed()), (Some(0), Some(0))); + release_tx.send(()).expect("release native"); + tokio::time::timeout(Duration::from_secs(2), async { + while disk.snapshot().active != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("native releases physical resource"); + } + + #[tokio::test(start_paused = true)] + async fn object_admission_closed_physical_capacity_defers_before_storage_locks() { + use super::super::storage_api::owner::PhysicalDiskBinding; + let node = admission(true); + let disk = rustfs_concurrency::adaptive::AdaptiveAdmission::new(NonZeroU16::MIN, 0); + let binding = PhysicalDiskBinding::default(); + binding.set_test_capacity(vec![disk.clone()]).expect("physical binding"); + let plan = binding.plan().expect("physical plan").resource_plan(); + for no_lock in [false, true] { + let began = tokio::time::Instant::now(); + let opts = super::HealOpts { + no_lock, + ..Default::default() + }; + let result = super::run_background_object(Some(&node), || Ok(Some(plan.clone())), &opts, |_: &()| true, async { + panic!("closed disk must prevent storage polling"); + }) + .await; + assert!(matches!(result, Err(super::Error::TransientSkip { .. }))); + assert_eq!(began.elapsed(), if no_lock { Duration::ZERO } else { Duration::from_millis(250) }); + assert_eq!((node.snapshot().active, disk.snapshot().active), (0, 0)); + } + } + #[tokio::test] async fn object_admission_is_wired_to_all_four_real_storage_entry_points() { use super::{ECStoreHealStorage, Error, HealOpts, HealStorageAPI}; @@ -2236,7 +2319,7 @@ mod tests { .base_dir(root.path()) .build() .await; - let gate = admission(false); + let gate = admission(true); let storage = ECStoreHealStorage::new(env.ecstore.clone()).with_admission(Some(gate.clone())); let opts = HealOpts { no_lock: true, @@ -2261,7 +2344,7 @@ mod tests { for result in results { assert!( matches!(result, Err(Error::TransientSkip { .. })), - "closed admission precedes even bucket validation: {result:?}" + "unknown physical ownership precedes even bucket validation: {result:?}" ); } assert_eq!(gate.completed(), Some(0)); @@ -2285,11 +2368,12 @@ mod tests { let gate = admission(false); let polled = std::cell::Cell::new(false); let began = tokio::time::Instant::now(); - let result = super::run_background_object(Some(&gate), &super::HealOpts::default(), |_: &()| true, async { - polled.set(true); - Ok(()) - }) - .await; + let result = + super::run_background_object(Some(&gate), physical_plan, &super::HealOpts::default(), |_: &()| true, async { + polled.set(true); + Ok(()) + }) + .await; assert!(matches!(result, Err(super::Error::TransientSkip { .. }))); assert!(!polled.get(), "no locks or storage mutation before admission"); assert_eq!(began.elapsed(), Duration::from_millis(250)); @@ -2304,9 +2388,10 @@ mod tests { ..Default::default() }; let began = tokio::time::Instant::now(); - let result = - super::run_background_object(Some(&gate), &opts, |_: &()| true, async { panic!("storage must remain unpolled") }) - .await; + let result = super::run_background_object(Some(&gate), physical_plan, &opts, |_: &()| true, async { + panic!("storage must remain unpolled") + }) + .await; assert!(matches!(result, Err(super::Error::TransientSkip { .. }))); assert_eq!(began.elapsed(), Duration::ZERO); } @@ -2316,15 +2401,17 @@ mod tests { use std::task::{Context, Poll, Waker}; let gate = admission(true); let opts = super::HealOpts::default(); - let future = super::run_background_object(Some(&gate), &opts, |_: &()| true, std::future::pending()); + let future = super::run_background_object(Some(&gate), physical_plan, &opts, |_: &()| true, std::future::pending()); let mut future = Box::pin(future); assert!(matches!(future.as_mut().poll(&mut Context::from_waker(Waker::noop())), Poll::Pending)); assert_eq!(gate.snapshot().active, 1); drop(future); assert_eq!(gate.snapshot().active, 0); assert_eq!(gate.completed(), Some(0)); - let result = - super::run_background_object(Some(&gate), &opts, |_: &()| true, async { Err(super::Error::TaskCancelled) }).await; + let result = super::run_background_object(Some(&gate), physical_plan, &opts, |_: &()| true, async { + Err(super::Error::TaskCancelled) + }) + .await; assert!(matches!(result, Err(super::Error::TaskCancelled))); assert_eq!(gate.completed(), Some(0)); assert_eq!(gate.snapshot().active, 0); @@ -2337,7 +2424,7 @@ mod tests { read_repair: true, ..Default::default() }; - let result = super::run_background_object(Some(&gate), &opts, |_: &()| true, async { + let result = super::run_background_object(Some(&gate), physical_plan, &opts, |_: &()| true, async { panic!("queued read repair must not bypass the closed node gate"); }) .await; @@ -2355,7 +2442,7 @@ mod tests { ..Default::default() }; assert_eq!( - super::run_background_object(Some(&gate), &opts, |ok: &bool| *ok, async { Ok(successful) }) + super::run_background_object(Some(&gate), physical_plan, &opts, |ok: &bool| *ok, async { Ok(successful) }) .await .expect("operation"), successful @@ -2365,9 +2452,15 @@ mod tests { } gate.update(&mut FeedbackController::new(), None); assert_eq!(gate.completed(), Some(1)); - super::run_background_object(None, &super::HealOpts::default(), |_: &()| true, async { Ok(()) }) - .await - .expect("default path stays available"); + super::run_background_object( + None, + || panic!("disabled admission must not read discovery"), + &super::HealOpts::default(), + |_: &()| true, + async { Ok(()) }, + ) + .await + .expect("default path stays available"); } #[test] diff --git a/crates/heal/src/heal/storage_api.rs b/crates/heal/src/heal/storage_api.rs index c1640b757..4350cf917 100644 --- a/crates/heal/src/heal/storage_api.rs +++ b/crates/heal/src/heal/storage_api.rs @@ -41,6 +41,13 @@ pub(crate) use rustfs_ecstore::api::storage::{ use rustfs_storage_api as storage_contracts; pub(crate) mod owner { + #[cfg(test)] + pub(crate) use rustfs_ecstore::api::disk::PhysicalDiskBinding; + pub(crate) use rustfs_ecstore::api::disk::PhysicalResourcePlan; + #[cfg(test)] + pub(crate) use rustfs_ecstore::spawn_blocking_with_resource_ownership as ecstore_spawn_blocking_with_resource_ownership; + pub(crate) use rustfs_ecstore::with_background_resource_set as ecstore_with_background_resource_set; + pub(crate) use super::storage_contracts::{ObjectIO, ObjectOperations}; pub(crate) use super::{