From 14e9fa2c8dabd684be86bbc8fe3be2355403404d Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Mon, 29 Jun 2026 20:19:38 +0800 Subject: [PATCH] fix(ecstore): avoid startup pool meta locks (#4056) --- crates/ecstore/src/core/pools.rs | 28 +++++- crates/ecstore/src/store/init.rs | 151 +++++++++++++++++++++++++++---- 2 files changed, 158 insertions(+), 21 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index fa2d740bc..aa867b1b0 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -45,7 +45,7 @@ use crate::storage_api_contracts::{ bucket::{BucketOperations, BucketOptions, MakeBucketOptions}, heal::HealOperations as _, namespace::NamespaceLocking as _, - object::{ObjectIO as _, ObjectOperations as _}, + object::{EcstoreObjectIO, ObjectIO as _, ObjectOperations as _}, }; use crate::{core::sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; @@ -1464,7 +1464,18 @@ impl PoolMeta { self.load_from_config_data(data) } - async fn load_no_lock(&mut self, pool: Arc) -> Result<()> { + /// Startup loads pool metadata before the full namespace-lock RPC surface is ready. + pub(crate) async fn load_for_startup(&mut self, pool: Arc) -> Result<()> + where + S: EcstoreObjectIO, + { + self.load_no_lock(pool).await + } + + async fn load_no_lock(&mut self, pool: Arc) -> Result<()> + where + S: EcstoreObjectIO, + { let data = match read_config_no_lock(pool, POOL_META_NAME).await { Ok(data) => data, Err(err) => { @@ -1502,7 +1513,18 @@ impl PoolMeta { Ok(()) } - async fn save_no_lock(&self, pools: Vec>) -> Result<()> { + /// Startup has a single elected local writer, so it must not depend on namespace locks here. + pub(crate) async fn save_for_startup(&self, pools: Vec>) -> Result<()> + where + S: EcstoreObjectIO, + { + self.save_no_lock(pools).await + } + + async fn save_no_lock(&self, pools: Vec>) -> Result<()> + where + S: EcstoreObjectIO, + { let data = self.encode_config_data()?; if data.is_empty() { return Ok(()); diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index fab34cc79..2d062d7f9 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -16,6 +16,7 @@ use super::*; use crate::core::pools::local_decommission_queue_prefix; use crate::error::is_err_decommission_running; use crate::runtime::sources as runtime_sources; +use crate::storage_api_contracts::object::EcstoreObjectIO; use tracing::{debug, error, info, warn}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -79,6 +80,22 @@ fn resolve_store_init_stage_result(result: Result<()>, stage: &str) -> Result<() result.map_err(|err| Error::other(format!("store init failed during {stage}: {err}"))) } +async fn load_pool_meta_for_startup(pool: Arc) -> Result +where + S: EcstoreObjectIO, +{ + let mut meta = PoolMeta::default(); + resolve_store_init_stage_result(meta.load_for_startup(pool).await, "load_pool_meta")?; + Ok(meta) +} + +async fn save_validated_pool_meta_for_startup(meta: &PoolMeta, pools: Vec>) -> Result<()> +where + S: EcstoreObjectIO, +{ + resolve_store_init_stage_result(meta.save_for_startup(pools).await, "save_validated_pool_meta") +} + async fn resume_local_decommission_after_init(store: Arc, rx: CancellationToken, pool_indices: Vec) { for attempt in 0..=LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES { if rx.is_cancelled() { @@ -353,18 +370,13 @@ impl ECStore { pub async fn init(self: &Arc, rx: CancellationToken) -> Result<()> { runtime_sources::ensure_boot_time().await; - let mut meta = PoolMeta::default(); - resolve_store_init_stage_result( - meta.load( - self.pools - .first() - .cloned() - .ok_or_else(|| Error::other("store init failed: no storage pools available"))?, - self.pools.clone(), - ) - .await, - "load_pool_meta", - )?; + let meta = load_pool_meta_for_startup( + self.pools + .first() + .cloned() + .ok_or_else(|| Error::other("store init failed: no storage pools available"))?, + ) + .await?; let update = meta.validate(self.pools.clone())?; let endpoints = runtime_sources::endpoint_pools_or_default(); let should_persist_pool_meta = runtime_sources::first_cluster_node_is_local().await; @@ -376,7 +388,7 @@ impl ECStore { // Only one local node should persist validated pool metadata here; otherwise // distributed startup can race on the same lock and replay the prior init bug. if should_persist_pool_meta { - resolve_store_init_stage_result(new_meta.save(self.pools.clone()).await, "save_validated_pool_meta")?; + save_validated_pool_meta_for_startup(&new_meta, self.pools.clone()).await?; } new_meta }; @@ -454,21 +466,102 @@ impl ECStore { #[cfg(test)] mod tests { use super::{ - LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, pool_first_endpoint_is_local, resolve_store_init_stage_result, - should_auto_start_rebalance_after_init, should_auto_start_rebalance_after_recovered_meta, - should_resume_local_decommission, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay, + LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, load_pool_meta_for_startup, pool_first_endpoint_is_local, + resolve_store_init_stage_result, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init, + should_auto_start_rebalance_after_recovered_meta, should_resume_local_decommission, + should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay, }; use crate::{ core::pools::{POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolStatus}, disk::endpoint::Endpoint, - error::StorageError, + error::{Error, Result, StorageError}, layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}, + object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}, services::rebalance::RebalanceMeta, + storage_api_contracts::{object::ObjectIO, range::HTTPRangeSpec}, + }; + use http::HeaderMap; + use std::{ + io::Cursor, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, }; - use std::time::Duration; use time::OffsetDateTime; use tokio_util::sync::CancellationToken; + #[derive(Debug)] + struct StartupPoolMetaStorage { + read_payload: Vec, + read_without_lock: AtomicBool, + wrote_without_lock: AtomicBool, + wrote_with_max_parity: AtomicBool, + } + + impl StartupPoolMetaStorage { + fn new(read_payload: Vec) -> Self { + Self { + read_payload, + read_without_lock: AtomicBool::new(false), + wrote_without_lock: AtomicBool::new(false), + wrote_with_max_parity: AtomicBool::new(false), + } + } + + fn object_info(&self, bucket: &str, object: &str, size: usize) -> ObjectInfo { + ObjectInfo { + bucket: bucket.to_string(), + name: object.to_string(), + size: size as i64, + actual_size: size as i64, + ..Default::default() + } + } + } + + #[async_trait::async_trait] + impl ObjectIO for StartupPoolMetaStorage { + type Error = Error; + type RangeSpec = HTTPRangeSpec; + type HeaderMap = HeaderMap; + type ObjectOptions = ObjectOptions; + type ObjectInfo = ObjectInfo; + type GetObjectReader = GetObjectReader; + type PutObjectReader = PutObjReader; + + async fn get_object_reader( + &self, + bucket: &str, + object: &str, + _range: Option, + _h: HeaderMap, + opts: &ObjectOptions, + ) -> Result { + assert!(opts.no_lock, "store init pool metadata load must not require namespace locks"); + self.read_without_lock.store(true, Ordering::SeqCst); + + Ok(GetObjectReader { + stream: Box::new(Cursor::new(self.read_payload.clone())), + object_info: self.object_info(bucket, object, self.read_payload.len()), + }) + } + + async fn put_object( + &self, + bucket: &str, + object: &str, + _data: &mut PutObjReader, + opts: &ObjectOptions, + ) -> Result { + assert!(opts.no_lock, "store init pool metadata save must not require namespace locks"); + self.wrote_without_lock.store(true, Ordering::SeqCst); + self.wrote_with_max_parity.store(opts.max_parity, Ordering::SeqCst); + Ok(self.object_info(bucket, object, 0)) + } + } + fn init_test_pool_meta(decommission: Option) -> PoolMeta { PoolMeta { version: POOL_META_VERSION, @@ -482,6 +575,28 @@ mod tests { } } + #[tokio::test] + async fn test_store_init_pool_meta_io_bypasses_namespace_lock_surface() { + let storage = Arc::new(StartupPoolMetaStorage::new(Vec::new())); + + let loaded = load_pool_meta_for_startup(storage.clone()) + .await + .expect("startup pool metadata load should tolerate missing metadata without locks"); + assert!(loaded.pools.is_empty()); + assert!(storage.read_without_lock.load(Ordering::SeqCst)); + + let meta = PoolMeta { + version: POOL_META_VERSION, + pools: Vec::new(), + dont_save: false, + }; + save_validated_pool_meta_for_startup(&meta, vec![storage.clone()]) + .await + .expect("startup pool metadata save should bypass locks"); + assert!(storage.wrote_without_lock.load(Ordering::SeqCst)); + assert!(storage.wrote_with_max_parity.load(Ordering::SeqCst)); + } + #[test] fn test_should_resume_local_decommission_respects_local_flag() { let mut local_endpoint = Endpoint::try_from("http://127.0.0.1:9000/data").expect("endpoint should parse");