feat(heal): admit physical resources before object repair

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<Self>) -> 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<Self>, owners: &[Arc<AdaptiveAdmission>]) -> Result<AdaptiveResourceSelection, CaptureError> {

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;

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;

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<AdaptivePermitSetOwner>,
     origin: Instant,
     valid_until_millis: AtomicU64,
 }
 
 impl PhysicalTopology {
-    fn new() -> Self {
+    fn new(catalog: Arc<AdaptivePermitSetOwner>) -> 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<NodeAdmission>) -> Option<NodeResourcePermit> {
+        self.resource_plan().try_acquire(node)
+    }
+
+    pub async fn acquire(&self, node: &Arc<NodeAdmission>) -> Option<NodeResourcePermit> {
+        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<PhysicalTopology>,
+}
+
+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<NodeAdmission>) -> Option<NodeResourcePermit> {
@@ -167,6 +205,37 @@ impl PhysicalDiskPlan {
     }
 }
 
+#[derive(Default)]
+pub(crate) struct PhysicalHealBinding(ArcSwapOption<PhysicalResourcePlan>);
+
+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<PhysicalResourcePlan> {
+        self.0
+            .load()
+            .as_ref()
+            .filter(|plan| plan.is_current())
+            .map(|plan| (**plan).clone())
+    }
+}
+
+struct HealSetSample {
+    binding: Arc<PhysicalHealBinding>,
+    disks: Option<Vec<Arc<PhysicalDiskBinding>>>,
+}
+
 struct Target {
     disk: Arc<LocalDisk>,
     binding: Arc<PhysicalDiskBinding>,
@@ -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<HealSetSample>,
     targets: Vec<Target>,
     resolved: Vec<Resolved>,
     devices: BTreeMap<PhysicalDeviceIdentity, Option<Observation>>,
@@ -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::<Vec<_>>();
@@ -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<PhysicalDiskBinding>], 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());

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";

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<RwLock<Vec<Option<DiskStore>>>>,
+    pub(crate) physical_heal_admission: Arc<crate::disk::PhysicalHealBinding>,
     pub set_endpoints: Vec<Endpoint>,
     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,

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<Option<crate::disk::PhysicalResourcePlan>> {
+        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<crate::disk::PhysicalResourcePlan> = 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<Vec<Arc<Sets>>> {
         match opts.pool {
             Some(pool_idx) => Ok(vec![
@@ -1554,21 +1581,25 @@ mod tests {
         }
     }
 
-    async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
-        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<Sets> {
+        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::<Vec<_>>();
         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]

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<T>(
     admission: Option<&Arc<NodeAdmission>>,
+    physical_plan: impl FnOnce() -> Result<Option<PhysicalResourcePlan>>,
     opts: &HealOpts,
     completed: impl FnOnce(&T) -> bool,
     operation: impl std::future::Future<Output = Result<T>>,
@@ -760,15 +764,17 @@ async fn run_background_object<T>(
     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<HealStorageObjectResult> {
         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<HealStorageObjectResult> {
         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<Error>)> {
         run_background_object(
             self.admission.as_ref(),
+            || self.ecstore.physical_heal_plan(object, opts).map_err(Error::Storage),
             opts,
             |result: &(HealResultItem, Option<Error>)| result.1.is_none(),
             async {
@@ -1590,6 +1599,7 @@ impl HealStorageAPI for ECStoreHealStorage {
     ) -> Result<HealStorageObjectResult> {
         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<Option<super::PhysicalResourcePlan>> {
+        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]

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::{