From 5211b5627726effe99074573593f74406ead7519 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 16:42:39 +0800 Subject: [PATCH] fix(ecstore): isolate pool metadata read probes (#7365) * fix(ecstore): isolate pool metadata read probes Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(readiness): surface blocked pool metadata writes Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(scanner): remove unused digest import Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/ecstore/src/core/pools.rs | 190 +++++++++++------- crates/ecstore/src/store/multipart.rs | 16 +- crates/ecstore/src/store/object.rs | 30 ++- .../src/scanner_folder/item_actions.rs | 2 +- rustfs/src/server/readiness.rs | 15 +- 5 files changed, 157 insertions(+), 96 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index d98c29c2d..e555573ae 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/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( where R: Into, { - // 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 { 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) { diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 469a5e365..2b14dc6a0 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/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 { 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 { 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() { diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index f231e738a..a012df8de 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -3123,6 +3123,9 @@ impl ECStore { Fut: std::future::Future>, { 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>, { + 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::>(); 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) diff --git a/crates/scanner/src/scanner_folder/item_actions.rs b/crates/scanner/src/scanner_folder/item_actions.rs index 5120f8181..1186e356e 100644 --- a/crates/scanner/src/scanner_folder/item_actions.rs +++ b/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)] diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 8a934e06e..db35d7966 100644 --- a/rustfs/src/server/readiness.rs +++ b/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]