fix(ecstore): isolate pool metadata read probes (#7365)

* fix(ecstore): isolate pool metadata read probes

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

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

* fix(readiness): surface blocked pool metadata writes

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

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

* fix(scanner): remove unused digest import

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

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

---------

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

crates/ecstore/src/core/pools.rs

@@ -4490,11 +4490,20 @@ impl PoolMetaWriteState {
 
     fn observe_selection(&mut self, selection: &PoolMetaSelection) -> Result<()> {
         self.pool_meta_absent = selection.absent;
+        self.validate_selection(selection)?;
+        if self.cluster_epoch.is_none()
+            && let Some((_, metadata_epoch)) = selection.generation_identity
+        {
+            self.cluster_epoch = Some(metadata_epoch);
+        }
+        Ok(())
+    }
+
+    fn validate_selection(&self, selection: &PoolMetaSelection) -> Result<()> {
         if let Some(expected_cluster_id) = self.expected_cluster_id
             && let Some((cluster_id, _)) = selection.generation_identity
             && cluster_id != expected_cluster_id
         {
-            self.block_writes();
             return Err(Error::other(format!(
                 "pool metadata incompatible: cluster identity {cluster_id} does not match deployment {expected_cluster_id}"
             )));
@@ -4503,17 +4512,11 @@ impl PoolMetaWriteState {
             && let Some((_, metadata_epoch)) = selection.generation_identity
             && metadata_epoch != identity_epoch
         {
-            self.block_writes();
             return Err(Error::other(format!(
                 "pool metadata recovery required: committed epoch {} does not match cluster identity epoch {identity_epoch}",
                 metadata_epoch
             )));
         }
-        if self.cluster_epoch.is_none()
-            && let Some((_, metadata_epoch)) = selection.generation_identity
-        {
-            self.cluster_epoch = Some(metadata_epoch);
-        }
         Ok(())
     }
 
@@ -4546,26 +4549,25 @@ impl PoolMetaWriteState {
         if !self.pool_meta_absent {
             return Ok(());
         }
+        let result = self.validate_missing_metadata_can_initialize();
+        if result.is_err() {
+            self.block_writes();
+        }
+        result
+    }
+
+    fn validate_missing_metadata_can_initialize(&self) -> Result<()> {
         match self.identity_initialized {
             Some(false) if self.bootstrap_identity_proven() && self.identity_fresh_bootstrap_nonce.is_some() => Ok(()),
-            Some(false) => {
-                self.block_writes();
-                Err(Error::other(
-                    "pool metadata recovery required: pending cluster identity exists but this startup has no verified fresh-bootstrap proof or legacy-adoption proof",
-                ))
-            }
-            Some(true) => {
-                self.block_writes();
-                Err(Error::other(
-                    "pool metadata recovery required: initialized cluster identity exists but every pool.bin replica is missing",
-                ))
-            }
-            None => {
-                self.block_writes();
-                Err(Error::other(
-                    "pool metadata recovery required: no durable bootstrap identity or pool.bin replica is available",
-                ))
-            }
+            Some(false) => Err(Error::other(
+                "pool metadata recovery required: pending cluster identity exists but this startup has no verified fresh-bootstrap proof or legacy-adoption proof",
+            )),
+            Some(true) => Err(Error::other(
+                "pool metadata recovery required: initialized cluster identity exists but every pool.bin replica is missing",
+            )),
+            None => Err(Error::other(
+                "pool metadata recovery required: no durable bootstrap identity or pool.bin replica is available",
+            )),
         }
     }
 
@@ -5145,12 +5147,12 @@ fn select_pool_meta_replicas_for_read_probe<R>(
 where
     R: Into<PoolMetaReplicaRead>,
 {
-    // Read-only planning probes must fail the current request on unsafe pool
-    // metadata, but they must not permanently poison the shared writer gate.
-    let mut probe_state = write_state.clone();
-    let selection = select_pool_meta_replicas_observing(&mut probe_state, replicas)?;
-    probe_state.observe_replicas(selection.replica_state);
-    probe_state.ensure_write_safe(operation)?;
+    let selection = select_pool_meta_replica_reads(replicas.into_iter().map(Into::into).collect())?;
+    write_state.validate_selection(&selection)?;
+    selection.replica_state.ensure_write_safe(operation)?;
+    if selection.absent && (write_state.expected_cluster_id.is_some() || write_state.identity_initialized.is_some()) {
+        write_state.validate_missing_metadata_can_initialize()?;
+    }
     Ok(selection)
 }
 
@@ -9014,7 +9016,7 @@ impl ECStore {
 
     async fn acquire_pool_meta_read_guard(
         &self,
-        write_state: &mut PoolMetaWriteState,
+        write_state: &PoolMetaWriteState,
         operation: &str,
     ) -> Result<(rustfs_lock::NamespaceLockGuard, PoolMeta)> {
         write_state.ensure_write_safe(operation)?;
@@ -9172,9 +9174,9 @@ impl ECStore {
         target_pool_indices: &[usize],
         phase: &'static str,
     ) -> Result<(rustfs_lock::NamespaceLockGuard, bool)> {
-        let mut save_guard = self.pool_meta_save_gate.lock().await;
+        let save_guard = self.pool_meta_save_gate.lock().await;
         let (pool_meta_guard, snapshot) = self
-            .acquire_pool_meta_read_guard(&mut save_guard, "target capacity admission failed")
+            .acquire_pool_meta_read_guard(&save_guard, "target capacity admission failed")
             .await?;
         for target_pool_index in target_pool_indices.iter().copied() {
             ensure_external_decommission_target_admission(&snapshot, target_pool_index, phase)?;
@@ -9208,9 +9210,9 @@ impl ECStore {
     pub(crate) async fn acquire_decommission_capacity_release_fence_with_active_source(
         &self,
     ) -> Result<(rustfs_lock::NamespaceLockGuard, bool)> {
-        let mut save_guard = self.pool_meta_save_gate.lock().await;
+        let save_guard = self.pool_meta_save_gate.lock().await;
         let (pool_meta_guard, snapshot) = self
-            .acquire_pool_meta_read_guard(&mut save_guard, "capacity release fence failed")
+            .acquire_pool_meta_read_guard(&save_guard, "capacity release fence failed")
             .await?;
         let has_active_source = pool_meta_has_active_decommission(&snapshot);
         drop(save_guard);
@@ -9275,9 +9277,9 @@ impl ECStore {
         }
 
         let (reconciliations, model_version) = {
-            let mut save_guard = self.pool_meta_save_gate.lock().await;
+            let save_guard = self.pool_meta_save_gate.lock().await;
             let (_read_guard, snapshot) = self
-                .acquire_pool_meta_read_guard(&mut save_guard, "exact delete capacity reconciliation failed")
+                .acquire_pool_meta_read_guard(&save_guard, "exact delete capacity reconciliation failed")
                 .await?;
             let reconciliations = plan_exact_delete_capacity_reconciliations(&snapshot, object, exact)?;
             let model_version = active_decommission_capacity_model(&snapshot)?;
@@ -9882,9 +9884,9 @@ impl ECStore {
         let non_growing_replacement = matches!(mode, DecommissionCapacityMutationMode::NonGrowingReplacement);
         let temporary_release = matches!(mode, DecommissionCapacityMutationMode::TemporaryRelease);
         let mut operation = Some(operation);
-        let mut save_guard = self.pool_meta_save_gate.lock().await;
+        let save_guard = self.pool_meta_save_gate.lock().await;
         let (read_guard, snapshot) = self
-            .acquire_pool_meta_read_guard(&mut save_guard, "target capacity admission failed")
+            .acquire_pool_meta_read_guard(&save_guard, "target capacity admission failed")
             .await?;
         let admission_now = OffsetDateTime::now_utc();
         let admitted_owner = capacity_owner.and_then(|owner| {
@@ -10269,6 +10271,14 @@ impl ECStore {
         self.pool_meta_save_gate.lock().await.ensure_write_safe(operation)
     }
 
+    /// Reports whether pool metadata side effects are currently writable.
+    /// Read-only admission probes do not change this state; startup and real
+    /// metadata transactions still latch it on unrecoverable conditions.
+    pub async fn pool_meta_writes_ready(&self) -> bool {
+        let write_state = self.pool_meta_save_gate.lock().await;
+        !write_state.write_blocked && !write_state.aborted_transaction.load(Ordering::SeqCst)
+    }
+
     async fn load_runtime_pool_meta_observing(&self, write_state: &mut PoolMetaWriteState, operation: &str) -> Result<PoolMeta> {
         write_state.ensure_write_safe(operation)?;
         load_pool_meta_identity_observing(self.pools.clone(), write_state).await?;
@@ -10891,9 +10901,9 @@ impl ECStore {
         // global lock, then fence the exact target cohort before taking the
         // write lock used to publish the terminal transition.
         let terminal_fence_plan = if acquire_runtime_fence {
-            let mut read_save_guard = self.pool_meta_save_gate.lock().await;
+            let read_save_guard = self.pool_meta_save_gate.lock().await;
             let (read_guard, snapshot) = self
-                .acquire_pool_meta_read_guard(&mut read_save_guard, "decommission cancel fence planning failed")
+                .acquire_pool_meta_read_guard(&read_save_guard, "decommission cancel fence planning failed")
                 .await?;
             let plan = decommission_capacity_terminal_fence_plan(&snapshot, idx)?;
             drop(read_guard);
@@ -17961,6 +17971,67 @@ mod tests {
         );
     }
 
+    #[test]
+    fn pool_meta_read_probe_does_not_latch_writer_state() {
+        let write_state = PoolMetaWriteState::default();
+        select_pool_meta_replicas_for_read_probe(
+            &write_state,
+            vec![PoolMetaReplica::Unreadable("transient read failure".to_string())],
+            "capacity probe",
+        )
+        .expect_err("an unreadable probe replica must fail the current admission");
+
+        assert!(
+            write_state.ensure_write_safe("ordinary object write").is_ok(),
+            "a read-only capacity probe must not permanently latch the pool metadata writer"
+        );
+    }
+
+    #[test]
+    fn pool_meta_read_probe_rejects_missing_runtime_metadata_without_latching() {
+        let write_state = PoolMetaWriteState {
+            expected_cluster_id: Some(uuid::Uuid::new_v4()),
+            identity_initialized: Some(true),
+            ..Default::default()
+        };
+        select_pool_meta_replicas_for_read_probe(&write_state, vec![PoolMetaReplica::Missing], "capacity probe")
+            .expect_err("runtime metadata disappearance must reject the current probe");
+        write_state
+            .ensure_write_safe("ordinary object write")
+            .expect("a missing-metadata probe must not permanently latch the writer");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn pool_meta_read_guard_does_not_latch_after_unreadable_replica() {
+        let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
+        let mut saved_disks = Vec::new();
+        for set in &store.pools[1].disk_set {
+            let mut disks = set.disks.write().await;
+            let original = std::mem::take(&mut *disks);
+            let disk_count = original.len();
+            saved_disks.push((set.clone(), original));
+            *disks = vec![None; disk_count];
+        }
+
+        let write_state = store.pool_meta_save_gate.lock().await;
+        store
+            .acquire_pool_meta_read_guard(&write_state, "capacity probe")
+            .await
+            .expect_err("an unreadable metadata replica must reject this probe");
+        write_state
+            .ensure_write_safe("ordinary object write")
+            .expect("a failed read-only probe must remain retryable");
+
+        for (set, disks) in saved_disks {
+            *set.disks.write().await = disks;
+        }
+        store
+            .acquire_pool_meta_read_guard(&write_state, "capacity probe retry")
+            .await
+            .expect("a read-only probe must succeed after the replica recovers");
+    }
+
     #[test]
     fn pool_meta_write_state_blocks_when_selection_has_no_valid_replica() {
         let replicas = vec![
@@ -17980,41 +18051,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn pool_meta_read_probe_does_not_latch_writer_state() {
-        let write_state = PoolMetaWriteState::default();
-        select_pool_meta_replicas_for_read_probe(
-            &write_state,
-            vec![PoolMetaReplica::Unreadable("transient read failure".to_string())],
-            "capacity probe",
-        )
-        .expect_err("an unreadable probe replica must fail the current admission");
-
-        write_state
-            .ensure_write_safe("ordinary object write")
-            .expect("a read-only capacity probe must not permanently latch the pool metadata writer");
-    }
-
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn pool_meta_read_guard_does_not_latch_after_unreadable_replica() {
-        let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
-        for set in &store.pools[1].disk_set {
-            let mut disks = set.disks.write().await;
-            let disk_count = disks.len();
-            *disks = vec![None; disk_count];
-        }
-
-        let mut write_state = store.pool_meta_save_gate.lock().await;
-        store
-            .acquire_pool_meta_read_guard(&mut write_state, "capacity probe")
-            .await
-            .expect_err("an unreadable metadata replica must reject this probe");
-        write_state
-            .ensure_write_safe("ordinary object write")
-            .expect("a failed read-only probe must remain retryable");
-    }
-
     #[test]
     fn pool_meta_write_state_blocks_on_any_recovery_required_selection() {
         fn assert_selection_blocks(replicas: Vec<PoolMetaReplica>) {

crates/ecstore/src/store/multipart.rs

@@ -526,12 +526,8 @@ impl ECStore {
         if self.single_pool() {
             self.apply_decommission_target_mutation_fence(0, object, &mut opts, mutation_fence)
                 .await;
-            return self
-                .run_decommission_capacity_admitted_mutation(0, None, None, || async {
-                    self.pools[0].new_multipart_upload(bucket, object, &opts).await
-                })
-                .await
-                .map(|res| (res, 0, opts.expected_bucket_incarnation_id));
+            let result = self.pools[0].new_multipart_upload(bucket, object, &opts).await?;
+            return Ok((result, 0, opts.expected_bucket_incarnation_id));
         }
 
         if opts.data_movement && opts.version_id.is_some() {
@@ -658,7 +654,9 @@ impl ECStore {
     ) -> Result<PartInfo> {
         check_put_object_part_args(bucket, object, upload_id)?;
         let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
-        opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+        if !self.single_pool() {
+            opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+        }
         let opts = &opts;
 
         if self.single_pool() {
@@ -982,7 +980,9 @@ impl ECStore {
     ) -> Result<ObjectInfo> {
         check_complete_multipart_args(bucket, object, upload_id)?;
         let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
-        opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+        if !self.single_pool() {
+            opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+        }
         let opts = &opts;
 
         if self.single_pool() {

crates/ecstore/src/store/object.rs

@@ -3123,6 +3123,9 @@ impl ECStore {
         Fut: std::future::Future<Output = Result<T>>,
     {
         let (lock_object, target_object) = objects;
+        if self.single_pool() {
+            return operation(opts).await;
+        }
         let (capacity_guard, has_active_decommission) = if capacity_releasing {
             self.acquire_decommission_capacity_release_fence_with_active_source().await?
         } else {
@@ -3220,6 +3223,9 @@ impl ECStore {
         F: FnOnce(HealOpts) -> Fut,
         Fut: std::future::Future<Output = Result<T>>,
     {
+        if self.single_pool() {
+            return operation(opts).await;
+        }
         let (capacity_guard, has_active_decommission) = self
             .acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "heal")
             .await?;
@@ -4162,7 +4168,9 @@ impl ECStore {
             .select_put_object_pool_idx(bucket, object.as_str(), data.size(), &opts)
             .await?;
         let mut opts = opts;
-        opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+        if !self.single_pool() {
+            opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+        }
         self.pools[idx]
             .put_object_with_old_current_size(bucket, object.as_str(), data, &opts)
             .await
@@ -4340,8 +4348,10 @@ impl ECStore {
                     object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
                     ..Default::default()
                 };
-                put_opts.decommission_capacity_admission =
-                    crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+                if !self.single_pool() {
+                    put_opts.decommission_capacity_admission =
+                        crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+                }
                 return if let Some(reader) = src_info.put_object_reader.as_mut() {
                     self.pools[pool_idx]
                         .put_object(dst_bucket, &dst_object, reader, &put_opts)
@@ -4376,8 +4386,10 @@ impl ECStore {
                         object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
                         ..Default::default()
                     };
-                    put_opts.decommission_capacity_admission =
-                        crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+                    if !self.single_pool() {
+                        put_opts.decommission_capacity_admission =
+                            crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+                    }
                     return self.pools[pool_idx]
                         .put_object(dst_bucket, &dst_object, reader, &put_opts)
                         .await;
@@ -4422,7 +4434,10 @@ impl ECStore {
             object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
             ..Default::default()
         };
-        put_opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+        if !self.single_pool() {
+            put_opts.decommission_capacity_admission =
+                crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
+        }
 
         if let Some(put_object_reader) = src_info.put_object_reader.as_mut() {
             return self.pools[pool_idx]
@@ -5057,7 +5072,7 @@ impl ECStore {
             }
         }
 
-        let _capacity_fence = if latest_marker_objects.iter().any(|creates_marker| *creates_marker) {
+        let _capacity_fence = if !self.single_pool() && latest_marker_objects.iter().any(|creates_marker| *creates_marker) {
             let target_pool_indices = (0..self.pools.len()).collect::<Vec<_>>();
             match self
                 .acquire_external_decommission_capacity_fence(&target_pool_indices, "batch_delete")
@@ -5375,7 +5390,6 @@ impl ECStore {
         // self-deadlocked on the inner commits.
         let object_name = object.as_str();
         if self.single_pool() {
-            opts.decommission_capacity_admission = Some(Arc::clone(&self));
             return self.pools[0]
                 .clone()
                 .restore_transitioned_object(bucket, object_name, &opts)

crates/scanner/src/scanner_folder/item_actions.rs

@@ -15,7 +15,7 @@
 use super::*;
 #[cfg(test)]
 use rustfs_filemeta::MetadataResolutionParams;
-use sha2::{Digest as _, Sha256};
+use sha2::Sha256;
 
 /// Cached folder information for scanning
 #[derive(Clone, Debug)]

rustfs/src/server/readiness.rs

@@ -608,7 +608,11 @@ where
 }
 
 fn storage_ready_from_runtime_state(info: &StorageInfo) -> bool {
-    storage_ready_from_runtime_state_with_quorum(info, pool_write_quorum)
+    storage_ready_from_runtime_state_with_pool_meta(info, true)
+}
+
+fn storage_ready_from_runtime_state_with_pool_meta(info: &StorageInfo, pool_meta_ready: bool) -> bool {
+    pool_meta_ready && storage_ready_from_runtime_state_with_quorum(info, pool_write_quorum)
 }
 
 fn storage_read_ready_from_runtime_state(info: &StorageInfo) -> bool {
@@ -720,7 +724,10 @@ pub async fn collect_cluster_read_health_report() -> DependencyReadinessReport {
 
 pub async fn collect_node_readiness_report() -> DependencyReadinessReport {
     let readiness = DependencyReadiness {
-        storage_ready: runtime_sources::current_object_store_handle().is_some(),
+        storage_ready: match runtime_sources::current_object_store_handle() {
+            Some(store) => store.pool_meta_writes_ready().await,
+            None => false,
+        },
         iam_ready: runtime_sources::current_iam_ready(),
         lock_quorum_ready: collect_lock_quorum_status().await.ready,
         peer_health_ready: collect_peer_health_readiness(),
@@ -810,6 +817,9 @@ async fn collect_lock_quorum_status() -> LockQuorumStatus {
 
 async fn collect_storage_readiness_uncached() -> bool {
     if let Some(store) = runtime_sources::current_object_store_handle() {
+        if !store.pool_meta_writes_ready().await {
+            return false;
+        }
         let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
         storage_ready_from_runtime_state(&storage_info)
     } else {
@@ -1555,6 +1565,7 @@ mod tests {
         };
 
         assert!(storage_ready_from_runtime_state(&info));
+        assert!(!storage_ready_from_runtime_state_with_pool_meta(&info, false));
     }
 
     #[test]