From b8889fccc9ef16d734fab58229d47867ab60b1ce Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 11 Oct 2026 11:50:24 +0800 Subject: [PATCH] fix(ecstore): allow cold-start recovery at EC read quorum (#8458) * fix(ecstore): load existing formats at EC read quorum * fix(ecstore): complete startup at EC read quorum --- .config/e2e-nightly-selection.txt | 4 +- crates/e2e_test/src/common.rs | 9 + .../src/namespace_lock_quorum_test.rs | 284 +++++++++++++++++- crates/ecstore/src/api/mod.rs | 3 +- crates/ecstore/src/bucket/migration.rs | 24 ++ crates/ecstore/src/set_disk/ops/bucket.rs | 18 +- crates/ecstore/src/store/bucket.rs | 117 +++++++- crates/ecstore/src/store/init.rs | 33 +- crates/ecstore/src/store/init_format.rs | 233 ++++++++++++++ rustfs/src/server/readiness.rs | 27 +- rustfs/src/startup_bucket_metadata.rs | 21 +- rustfs/src/storage/storage_api.rs | 4 +- 12 files changed, 708 insertions(+), 69 deletions(-) diff --git a/.config/e2e-nightly-selection.txt b/.config/e2e-nightly-selection.txt index 07dfe2811..bf672a947 100644 --- a/.config/e2e-nightly-selection.txt +++ b/.config/e2e-nightly-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=efe68675f4676a0d2901242c5815b1f7af5cdec8231f7cccd408f6dacf5c75db -sha256-linux=003ed6480935a0a95680cc061be143722fbd37bd477b8ff98607b1a064f3aef1 +sha256-darwin=5dbb9154b8feb94cedc8fb4aa6895d4b3259cc31b7df1f5c3b9928ef07f04535 +sha256-linux=d3a210be9eccf593d3723ae669b66680e9751dd01a6949f03df20614f913b707 diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index 622fe9015..e573fb00f 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -1542,6 +1542,15 @@ impl RustFSTestClusterEnvironment { self.start_node_from_binary(node_idx, &binary_path).await } + /// Start a process and wait only for TCP, allowing tests to bring up a + /// quorum one node at a time before the S3 service can become ready. + pub async fn start_node_process(&mut self, node_idx: usize) -> Result<(), Box> { + let binary_path = rustfs_binary_path(); + let volumes_arg = self.build_volumes_arg(); + self.spawn_node(node_idx, &binary_path, &volumes_arg)?; + self.wait_for_node_ready(&self.nodes[node_idx].address, node_idx).await + } + /// Start one stopped cluster node with a specific RustFS binary while /// preserving the cluster's volume layout and that node's data directory. pub async fn start_node_from_binary( diff --git a/crates/e2e_test/src/namespace_lock_quorum_test.rs b/crates/e2e_test/src/namespace_lock_quorum_test.rs index 46a59a943..1edf47c2b 100644 --- a/crates/e2e_test/src/namespace_lock_quorum_test.rs +++ b/crates/e2e_test/src/namespace_lock_quorum_test.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::common::RustFSTestClusterEnvironment; +use crate::common::{ClusterTopology, RustFSTestClusterEnvironment}; use aws_sdk_s3::Client; use aws_sdk_s3::error::SdkError; use bytes::Bytes; @@ -185,7 +185,11 @@ async fn wait_for_quorum_read_admission(clients: &[Client], bucket: &str) -> Tes return Ok(()); } if tokio::time::Instant::now() >= deadline { - return Err(format!("read quorum did not become available after lease convergence for {bucket}").into()); + let diagnostics = clients[0].get_object().bucket(bucket).key("warm-small").send().await; + return Err(format!( + "read quorum did not become available after lease convergence for {bucket}: GET={diagnostics:?}" + ) + .into()); } tokio::time::sleep(std::time::Duration::from_millis(100)).await; } @@ -195,11 +199,14 @@ async fn wait_for_quorum_read_admission(clients: &[Client], bucket: &str) -> Tes async fn test_degraded_cluster_read_quorum_follows_erasure_layout() -> TestResult { crate::common::init_logging(); - for (node_count, parity) in [(4, 2), (6, 3), (6, 2)] { + for (node_count, parity) in [(3, 1), (4, 2), (6, 3), (6, 2), (4, 0)] { let read_quorum = node_count - parity; let write_quorum = read_quorum + usize::from(read_quorum == parity); let mut cluster = RustFSTestClusterEnvironment::new(node_count).await?; cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}")); + if parity == 0 { + cluster.set_env("RUSTFS_STORAGE_CLASS_RRS", "EC:0"); + } // Wait for every seed fanout before removing any physical shard. cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false"); cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false"); @@ -338,6 +345,277 @@ async fn test_degraded_cluster_read_quorum_follows_erasure_layout() -> TestResul Ok(()) } +#[tokio::test] +async fn test_cold_start_read_quorum_follows_erasure_layout() -> TestResult { + crate::common::init_logging(); + for (node_count, parity) in [(3, 1), (4, 2), (6, 3), (6, 2), (4, 0)] { + let read_quorum = node_count - parity; + let write_quorum = read_quorum + usize::from(read_quorum == parity); + let mut cluster = RustFSTestClusterEnvironment::new(node_count).await?; + cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}")); + if parity == 0 { + cluster.set_env("RUSTFS_STORAGE_CLASS_RRS", "EC:0"); + } + cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false"); + cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false"); + cluster.set_env("RUST_LOG", "warn"); + cluster.start().await?; + let clients = cluster + .create_all_clients()? + .into_iter() + .map(|client| { + Client::from_conf( + client + .config() + .to_builder() + .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1)) + .build(), + ) + }) + .collect::>(); + let bucket = format!("cold-read-quorum-{node_count}-{parity}"); + clients[0].create_bucket().bucket(&bucket).send().await?; + let small = b"cold startup follows the stored EC layout".to_vec(); + let large = vec![73; 256 * 1024]; + for (key, body) in [ + ("warm-small", &small), + ("warm-large", &large), + ("cold-large", &large), + ("below-quorum", &large), + ] { + clients[0] + .put_object() + .bucket(&bucket) + .key(key) + .body(Bytes::copy_from_slice(body).into()) + .send() + .await?; + for node in &cluster.nodes { + let census = + crate::chaos::census_object_version_on_disk(std::path::Path::new(&node.data_dir), &bucket, key, None)?; + assert!(census.is_complete(), "seed shard must exist before cold startup: {census:?}"); + assert_eq!((census.data_blocks, census.parity_blocks), (Some(read_quorum), Some(parity))); + } + } + for node in 0..node_count { + cluster.stop_node(node)?; + } + // Rebuild every process's in-memory state below read quorum, then + // cross the EC boundary. A format bootstrap must not admit S3 reads + // merely because half the peers are online (e.g. six disks, EC:2). + for node in 0..read_quorum - 1 { + cluster.start_node_process(node).await?; + } + let error = clients[0] + .get_object() + .bucket(&bucket) + .key("below-quorum") + .send() + .await + .expect_err("cold startup below the configured EC read quorum must stay unavailable"); + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503)); + cluster.start_node_process(read_quorum - 1).await?; + wait_for_quorum_read_admission(&clients[..read_quorum], &bucket).await?; + for client in clients.iter().take(read_quorum) { + assert_quorum_object_body(client, &bucket, "warm-small", &small).await?; + assert_quorum_object_body(client, &bucket, "cold-large", &large).await?; + let write = client + .put_object() + .bucket(&bucket) + .key("cold-quorum-write") + .body(Bytes::copy_from_slice(&small).into()) + .send() + .await; + if read_quorum >= write_quorum { + write?; + assert_quorum_object_body(client, &bucket, "cold-quorum-write", &small).await?; + } else { + let error = write.expect_err("cold startup at read quorum must preserve the EC write quorum"); + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503)); + } + } + for node in read_quorum..node_count { + cluster.start_node(node).await?; + } + for client in &clients { + assert_quorum_object_body(client, &bucket, "cold-large", &large).await?; + } + } + Ok(()) +} + +#[tokio::test] +async fn test_four_by_six_cold_start_at_read_quorum() -> TestResult { + crate::common::init_logging(); + let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(4, 6)).await?; + // Explicit URLs are consumed in order. Place one drive from each node in + // every set, matching a four-host, six-drive-per-host deployment. + let volumes = (0..6) + .flat_map(|drive| { + cluster + .nodes + .iter() + .map(move |node| format!("http://{}{}", node.address, node.data_dirs[drive])) + }) + .collect::>() + .join(" "); + cluster.set_env("RUSTFS_VOLUMES", volumes); + cluster.set_env("RUSTFS_ERASURE_SET_DRIVE_COUNT", "4"); + cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:2"); + cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false"); + cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false"); + cluster.set_env("RUST_LOG", "error"); + cluster.start().await?; + let clients = cluster + .create_all_clients()? + .into_iter() + .map(|client| { + Client::from_conf( + client + .config() + .to_builder() + .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1)) + .build(), + ) + }) + .collect::>(); + let bucket = "cold-start-read-quorum"; + clients[0].create_bucket().bucket(bucket).send().await?; + let mut fixtures = Vec::new(); + let mut covered = [[false; 2]; 6]; + for index in 0..128 { + let key = match index { + 0 => "warm-small".to_string(), + 1 => "warm-large".to_string(), + _ => format!("cold-{index}"), + }; + let size_class = index % 2; + let size = if size_class == 0 { 71 } else { 256 * 1024 }; + let body = (0..size) + .map(|offset| u8::try_from(offset % 251).expect("bounded fixture byte")) + .collect::>(); + clients[0] + .put_object() + .bucket(bucket) + .key(&key) + .body(Bytes::copy_from_slice(&body).into()) + .send() + .await?; + for node in &cluster.nodes { + let mut complete_shards = 0; + for (drive, path) in node.data_dirs.iter().enumerate() { + let census = crate::chaos::census_object_version_on_disk(std::path::Path::new(path), bucket, &key, None)?; + if census.has_xl_meta { + assert!(census.is_complete(), "seed shard must be complete before shutdown: {census:?}"); + assert_eq!((census.data_blocks, census.parity_blocks), (Some(2), Some(2))); + covered[drive][size_class] = true; + complete_shards += 1; + } + } + assert_eq!(complete_shards, 1, "each node must own one shard of every seed object"); + } + fixtures.push((key, body)); + if covered.iter().flatten().all(|covered| *covered) { + break; + } + } + assert!( + covered.iter().flatten().all(|covered| *covered), + "fixtures must cover all six sets with inline and large objects" + ); + + for first in 0..4 { + for second in first + 1..4 { + for node in 0..4 { + cluster.stop_node(node)?; + } + cluster.start_node_process(first).await?; + let error = clients[first] + .get_object() + .bucket(bucket) + .key("warm-large") + .send() + .await + .expect_err("one node cannot satisfy read quorum after a cold start"); + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503)); + cluster.start_node_process(second).await?; + wait_for_quorum_read_admission(&[clients[first].clone(), clients[second].clone()], bucket).await?; + for node in [first, second] { + for (key, body) in &fixtures { + assert_quorum_object_body(&clients[node], bucket, key, body).await?; + } + let range = clients[node] + .get_object() + .bucket(bucket) + .key("warm-large") + .range("bytes=17-1023") + .send() + .await?; + assert_eq!(range.body.collect().await?.into_bytes().as_ref(), &fixtures[1].1[17..1024]); + let listing = clients[node].list_objects_v2().bucket(bucket).send().await?; + for (key, _) in &fixtures { + assert!( + listing.contents().iter().any(|object| object.key() == Some(key.as_str())), + "listing omitted {key}" + ); + } + let error = clients[node] + .put_object() + .bucket(bucket) + .key("recovery-write") + .body(Bytes::from_static(b"after recovery").into()) + .send() + .await + .expect_err("two nodes must not authorize a write"); + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503)); + } + let error = clients[first] + .delete_object() + .bucket(bucket) + .key("warm-large") + .send() + .await + .expect_err("two nodes must not authorize a delete"); + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503)); + assert_quorum_object_body(&clients[first], bucket, "warm-large", &fixtures[1].1).await?; + + let remaining = (0..4).filter(|node| *node != first && *node != second).collect::>(); + cluster.start_node(remaining[0]).await?; + let deadline = tokio::time::Instant::now() + Duration::from_secs(90); + loop { + match clients[first] + .put_object() + .bucket(bucket) + .key("recovery-write") + .body(Bytes::from_static(b"after recovery").into()) + .send() + .await + { + Ok(_) => break, + Err(error) + if error.raw_response().is_some_and(|response| response.status().as_u16() == 503) + && tokio::time::Instant::now() < deadline => + { + tokio::time::sleep(Duration::from_millis(200)).await + } + Err(error) => return Err(error.into()), + } + } + for node in [first, second, remaining[0]] { + assert_quorum_object_body(&clients[node], bucket, "recovery-write", b"after recovery").await?; + } + cluster.start_node(remaining[1]).await?; + for client in &clients { + for (key, body) in &fixtures { + assert_quorum_object_body(client, bucket, key, body).await?; + } + assert_quorum_object_body(client, bucket, "recovery-write", b"after recovery").await?; + } + } + } + Ok(()) +} + async fn put_object(client: Client, payload: Vec, writer_id: usize) -> Result<(), String> { client .put_object() diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 3a28b840f..e5e22995f 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -210,7 +210,8 @@ pub mod bucket { pub mod migration { pub use crate::bucket::migration::{ - LegacyBlobDecryptFn, migration_startup_error, try_migrate_bucket_metadata, try_migrate_iam_config, + LegacyBlobDecryptFn, migration_startup_error, try_migrate_bucket_metadata, try_migrate_bucket_metadata_for_buckets, + try_migrate_iam_config, }; } diff --git a/crates/ecstore/src/bucket/migration.rs b/crates/ecstore/src/bucket/migration.rs index 723b75c69..6785795c2 100644 --- a/crates/ecstore/src/bucket/migration.rs +++ b/crates/ecstore/src/bucket/migration.rs @@ -273,6 +273,30 @@ where let buckets: Vec = buckets_list.into_iter().map(|b| b.name).collect(); + try_migrate_bucket_metadata_for_buckets(store, &buckets).await +} + +/// Migrate the already discovered startup candidates without repeating a +/// majority-only namespace listing. Every migration write retains its quorum. +pub async fn try_migrate_bucket_metadata_for_buckets(store: Arc, buckets: &[String]) -> Result<()> +where + S: ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + > + ObjectOperations< + Error = Error, + ObjectInfo = ObjectInfo, + ObjectOptions = ObjectOptions, + FileInfo = FileInfo, + ObjectToDelete = ObjectToDelete, + DeletedObject = DeletedObject, + >, +{ if buckets.is_empty() { debug!("No migrating bucket metadata found"); return Ok(()); diff --git a/crates/ecstore/src/set_disk/ops/bucket.rs b/crates/ecstore/src/set_disk/ops/bucket.rs index d1dc1770a..738648df7 100644 --- a/crates/ecstore/src/set_disk/ops/bucket.rs +++ b/crates/ecstore/src/set_disk/ops/bucket.rs @@ -123,8 +123,18 @@ impl SetDisks { } pub(crate) async fn list_bucket_for_scanner(&self, _opts: &BucketOptions) -> Result<(Vec, bool)> { + self.list_bucket_with_quorum(BucketInfoQuorum::Write).await + } + + pub(crate) async fn list_bucket_with_quorum(&self, quorum: BucketInfoQuorum) -> Result<(Vec, bool)> { let disks = self.disk_inventory().await; let write_quorum = (disks.len() / 2) + 1; + // Discovery uses the same namespace read boundary as stat_bucket. + // It grants no authority to mutate buckets or accept object reads. + let required = match quorum { + BucketInfoQuorum::Read => disks.len().div_ceil(2).max(1), + BucketInfoQuorum::Write => write_quorum, + }; let mut futures = Vec::with_capacity(disks.len()); for disk in disks { @@ -153,7 +163,11 @@ impl SetDisks { } } - if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { + let error = match quorum { + BucketInfoQuorum::Read => reduce_read_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, required), + BucketInfoQuorum::Write => reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, required), + }; + if let Some(err) = error { return Err(err.into()); } @@ -179,7 +193,7 @@ impl SetDisks { topology_complete &= counts.values().all(|(count, _)| *count >= write_quorum); let mut buckets = counts .into_values() - .filter_map(|(count, bucket)| (count >= write_quorum).then_some(bucket)) + .filter_map(|(count, bucket)| (count >= required).then_some(bucket)) .collect::>(); buckets.sort_by(|left, right| left.name.cmp(&right.name)); Ok((buckets, topology_complete)) diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 327769638..57ccd3581 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -1231,34 +1231,46 @@ impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result> { // TODO(backlog): support cached bucket listing via opts.cached - Ok(self.list_bucket_from_sets(opts, usize::MAX).await?.buckets) + Ok(self + .list_bucket_from_sets(opts, usize::MAX, BucketInfoQuorum::Write) + .await? + .buckets) } pub async fn list_bucket_for_scanner(&self, opts: &BucketOptions) -> Result { - self.list_bucket_from_sets(opts, SCANNER_BUCKET_LIST_SET_CONCURRENCY).await + self.list_bucket_from_sets(opts, SCANNER_BUCKET_LIST_SET_CONCURRENCY, BucketInfoQuorum::Write) + .await + } + + /// Discover candidates for metadata initialization at each set's namespace + /// read quorum. Metadata loading still validates persisted configuration; + /// this listing must not be used as a complete scanner or mutation view. + pub async fn list_buckets_for_metadata_startup(&self) -> Result> { + Ok(self + .list_bucket_from_sets( + &BucketOptions { + no_metadata: true, + ..Default::default() + }, + SCANNER_BUCKET_LIST_SET_CONCURRENCY, + BucketInfoQuorum::Read, + ) + .await? + .buckets) } async fn list_bucket_from_sets( &self, opts: &BucketOptions, max_concurrency: usize, + quorum: BucketInfoQuorum, ) -> Result { let set_count = self.pools.iter().map(|pool| pool.disk_set.len()).sum(); let concurrency = bucket_list_set_concurrency(set_count, max_concurrency); - let deleted = opts.deleted; - let cached = opts.cached; - let no_metadata = opts.no_metadata; - let mut set_listings = stream::iter(self.bucket_sets().map(move |(pool_index, set_index, set)| { - let opts = BucketOptions { - deleted, - cached, - no_metadata, - }; - async move { - set.list_bucket_for_scanner(&opts) - .await - .map(|(buckets, complete)| (pool_index, set_index, buckets, complete)) - } + let mut set_listings = stream::iter(self.bucket_sets().map(move |(pool_index, set_index, set)| async move { + set.list_bucket_with_quorum(quorum) + .await + .map(|(buckets, complete)| (pool_index, set_index, buckets, complete)) })) .buffer_unordered(concurrency); let mut topology_complete = set_count != 0; @@ -3543,6 +3555,79 @@ mod tests { assert_eq!(info_err, StorageError::ErasureReadQuorum); } + #[tokio::test] + #[serial] + async fn metadata_startup_listing_uses_each_sets_read_quorum() { + for drives_per_pool in [vec![2], vec![3], vec![4], vec![6], vec![8], vec![4, 6]] { + let (_temp_dir, store) = setup_bucket_quorum_test_env(&drives_per_pool, None).await; + let bucket = "metadata-startup-read-quorum"; + for pool in &store.pools { + pool.disk_set[0] + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("healthy set should create the startup candidate"); + } + let mut offline = Vec::new(); + for (pool_index, &width) in drives_per_pool.iter().enumerate() { + let set = &store.pools[pool_index].disk_set[0]; + let indexes = (width.div_ceil(2)..width).collect::>(); + offline.push(take_set_disks_offline(&store, set, &indexes).await); + } + let listing = store + .list_buckets_for_metadata_startup() + .await + .expect("startup must discover buckets at every set's read quorum"); + assert_eq!(listing.len(), 1); + assert_eq!(listing[0].name, bucket); + if drives_per_pool.iter().any(|width| width % 2 == 0) { + let error = store + .list_bucket_for_scanner(&BucketOptions::default()) + .await + .expect_err("startup discovery must not weaken scanner quorum"); + assert_eq!(error, StorageError::ErasureWriteQuorum); + let error = store + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + .expect_err("ordinary listing must retain its existing quorum"); + assert_eq!(error, StorageError::ErasureWriteQuorum); + } + // A healthy earlier pool must not hide read-quorum loss in a later pool. + let last_pool = store.pools.len() - 1; + let set = &store.pools[last_pool].disk_set[0]; + let extra = take_set_disks_offline(&store, set, &[0]).await; + let error = store + .list_buckets_for_metadata_startup() + .await + .expect_err("every set must retain read quorum for startup discovery"); + assert_eq!(error, StorageError::ErasureReadQuorum); + restore_set_disks(&store, set, extra).await; + for (pool_index, disks) in offline.into_iter().enumerate() { + restore_set_disks(&store, &store.pools[pool_index].disk_set[0], disks).await; + } + } + } + + #[tokio::test] + #[serial] + async fn metadata_startup_listing_does_not_accept_single_disk_bucket_remnants() { + let (temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await; + let bucket = "incomplete-startup-bucket"; + tokio::fs::create_dir(temp_dir.path().join("pool0-disk0").join(bucket)) + .await + .expect("create one uncommitted directory remnant"); + let set = &store.pools[0].disk_set[0]; + let offline = take_set_disks_offline(&store, set, &[2, 3]).await; + let listing = store + .list_buckets_for_metadata_startup() + .await + .expect("two responsive disks satisfy discovery quorum"); + assert!(listing.iter().all(|info| info.name != bucket)); + restore_set_disks(&store, set, offline).await; + } + #[tokio::test] #[serial] async fn scanner_bucket_listing_unions_every_erasure_set() { diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 0a2a855bb..4b5580e71 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -108,6 +108,10 @@ fn should_retry_format_load(err: &Error) -> bool { ) } +fn format_load_retry_exhausted(err: &Error, retries: u32) -> bool { + retries >= 10 && !matches!(err, Error::ErasureReadQuorum | Error::FirstDiskWait | Error::NotFirstDisk) +} + fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_resume_required: bool) -> bool { rebalance_resume_required && !decommission_running } @@ -484,7 +488,7 @@ impl ECStore { check_disk_fatal_errs(&errs)?; let loaded_format = { - let mut times = 0; + let mut times = 0_u32; let mut interval = 1; loop { match init_format::connect_load_init_formats_with_instance_ctx( @@ -500,13 +504,13 @@ impl ECStore { Ok(fm) => break Ok(fm), Err(e) if !should_retry_format_load(&e) => break Err(e), // Wrap the final error if we are giving up - Err(e) if times >= 10 => { + Err(e) if format_load_retry_exhausted(&e, times) => { break Err(Error::other(format!("store init failed to load formats after {times} retries: {e}"))); } // Retrying so just drop the error Err(_) => {} } - times += 1; + times = times.saturating_add(1); if interval < 16 { interval *= 2; } @@ -519,6 +523,12 @@ impl ECStore { "Retrying storage format load" ); select! { + _ = ctx.cancelled() => { + return Err(Error::other(std::io::Error::new( + std::io::ErrorKind::Interrupted, + "storage format initialization cancelled", + ))); + } _ = tokio::signal::ctrl_c() => { info!( event = EVENT_STORE_FORMAT_RETRY, @@ -2201,6 +2211,23 @@ mod tests { assert_eq!(attempts.load(Ordering::SeqCst), 2); } + #[test] + fn format_peer_wait_does_not_exhaust_the_retry_budget() { + for retries in [0, 9, 10, u32::MAX] { + for error in [ + StorageError::ErasureReadQuorum, + StorageError::FirstDiskWait, + StorageError::NotFirstDisk, + ] { + assert!( + !super::format_load_retry_exhausted(&error, retries), + "peer availability must remain retryable" + ); + } + assert_eq!(super::format_load_retry_exhausted(&StorageError::Timeout, retries), retries >= 10); + } + } + #[test] fn test_should_retry_format_load_rejects_permanent_corruption() { assert!(!should_retry_format_load(&StorageError::CorruptedFormat)); diff --git a/crates/ecstore/src/store/init_format.rs b/crates/ecstore/src/store/init_format.rs index 9bfb62c60..c5f786848 100644 --- a/crates/ecstore/src/store/init_format.rs +++ b/crates/ecstore/src/store/init_format.rs @@ -134,6 +134,18 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx( { return Err(Error::CorruptedFormat); } + if matches!(format_quorum, Some(Err(Error::ErasureReadQuorum))) + && let Some((format, members)) = select_existing_format_read_quorum(&formats, &errs, set_drive_count) + { + retain_format_quorum_members(instance_ctx, disks, &format, &members, set_drive_count).await?; + // Loading an existing format grants no authority to initialize pool + // metadata or migrate formats. The persisted storage classes and each + // object's EC metadata still determine service and operation quorums. + return Ok(LoadedFormat { + format, + pool_meta_bootstrap_authority: PoolMetaBootstrapAuthority::None, + }); + } let resumable_partial_migration = formats_present > 0 && errs.iter().any(|error| matches!(error, Some(DiskError::UnformattedDisk))) && format_quorum.as_ref().is_some_and(Result::is_err) @@ -578,6 +590,43 @@ pub fn get_format_erasure_in_quorum(formats: &[Option], slot_offset: u select_format_erasure_in_quorum(formats, slot_offset).map(|(format, _)| format) } +/// Load a consistent existing layout with enough members per set to read the +/// max-parity system metadata containing the persisted storage classes. This +/// is not object-read admission: STANDARD, RRS and historical objects can have +/// different EC layouts, which are validated after this bootstrap read. +/// Format migration and healing still use strict-majority election. +fn select_existing_format_read_quorum( + formats: &[Option], + errors: &[Option], + set_drive_count: usize, +) -> Option<(FormatV3, Vec)> { + if set_drive_count == 0 || formats.is_empty() || formats.len() != errors.len() { + return None; + } + let metadata_parity = set_drive_count / 2; + let metadata_read_quorum = set_drive_count - metadata_parity; + let reference = formats.iter().flatten().next()?; + if !formats_match_reference_slots(formats, reference, 0) + || check_format_erasure_value_for_topology(reference, formats.len(), set_drive_count).is_err() + || formats.iter().zip(errors).any(|(format, error)| match (format, error) { + (Some(_), None) => false, + (None, Some(DiskError::DiskNotFound)) => false, + (None, Some(error)) => !is_network_like_disk_error(error), + _ => true, + }) + || formats + .chunks(set_drive_count) + .any(|set| set.iter().flatten().count() < metadata_read_quorum) + { + return None; + } + + let mut format = reference.clone(); + format.erasure.this = Uuid::nil(); + format.disk_info = None; + Some((format, formats.iter().map(Option::is_some).collect())) +} + pub(crate) fn select_format_erasure_in_quorum(formats: &[Option], slot_offset: usize) -> Result<(FormatV3, Vec)> { let mut candidates = HashMap::new(); let formats_present = formats.iter().flatten().count(); @@ -1120,6 +1169,190 @@ mod tests { assert!(!formats_match_reference_slots(&formats, &reference, 0)); } + fn half_pool_formats() -> (Vec>, Vec>) { + let reference = FormatV3::new(6, 4); + let formats = (0..24) + .map(|slot| { + (slot % 4 < 2).then(|| { + let mut format = reference.clone(); + format.erasure.this = reference.erasure.sets[slot / 4][slot % 4]; + format + }) + }) + .collect::>(); + let errors = formats + .iter() + .map(|format| format.is_none().then_some(DiskError::DiskNotFound)) + .collect(); + (formats, errors) + } + + #[test] + fn existing_format_read_quorum_follows_each_sets_metadata_layout() { + // These are system-metadata EC boundaries, independent of user-object + // storage classes and of how many drives are hosted by each node. + for (set_width, read_quorum) in [(1, 1), (2, 1), (3, 2), (4, 2), (6, 3), (8, 4), (12, 6), (16, 8)] { + for set_count in [1, 2, 5] { + let reference = FormatV3::new(set_count, set_width); + let mut formats = (0..set_count * set_width) + .map(|slot| { + let set = slot / set_width; + let drive = slot % set_width; + // Rotate surviving slots between sets so admission + // cannot depend on the first drive or node being up. + ((drive + set) % set_width < read_quorum).then(|| { + let mut format = reference.clone(); + format.erasure.this = reference.erasure.sets[set][drive]; + format + }) + }) + .collect::>(); + let mut errors = formats + .iter() + .map(|format| format.is_none().then_some(DiskError::DiskNotFound)) + .collect::>(); + let (loaded, members) = select_existing_format_read_quorum(&formats, &errors, set_width) + .unwrap_or_else(|| panic!("metadata read quorum must load {set_count} sets of {set_width} drives")); + assert_eq!(loaded.shared_identity(), reference.shared_identity()); + assert!(loaded.erasure.this.is_nil()); + assert_eq!(members.iter().filter(|member| **member).count(), set_count * read_quorum); + + // More members in other sets cannot compensate for one set + // falling below its own metadata reconstruction boundary. + for slot in 0..(set_count - 1) * set_width { + let mut format = reference.clone(); + format.erasure.this = reference.erasure.sets[slot / set_width][slot % set_width]; + formats[slot] = Some(format); + errors[slot] = None; + } + let lost_slot = formats + .iter() + .rposition(Option::is_some) + .expect("at least one surviving drive"); + formats[lost_slot] = None; + errors[lost_slot] = Some(DiskError::DiskNotFound); + assert!( + select_existing_format_read_quorum(&formats, &errors, set_width).is_none(), + "one set below read quorum must block {set_count} sets of {set_width} drives" + ); + } + } + } + + #[test] + fn existing_format_read_quorum_preserves_election_and_set_boundaries() { + let (mut formats, mut errors) = half_pool_formats(); + assert!(select_existing_format_read_quorum(&[], &[], 0).is_none()); + assert!(select_existing_format_read_quorum(&formats, &errors[..23], 4).is_none()); + assert!(select_existing_format_read_quorum(&formats, &errors, 0).is_none()); + assert!(select_existing_format_read_quorum(&formats, &errors, 6).is_none()); + assert!(select_existing_format_read_quorum(&formats, &errors, 4).is_some()); + assert!(matches!(get_format_erasure_in_quorum(&formats, 0), Err(Error::ErasureReadQuorum))); + + let mut extra = formats[0].as_ref().expect("seed format").clone(); + extra.erasure.this = extra.erasure.sets[0][2]; + formats[2] = Some(extra); + errors[2] = None; + assert!(get_format_erasure_in_quorum(&formats, 0).is_ok(), "13 votes must establish a majority"); + formats[5] = None; + errors[5] = Some(DiskError::DiskNotFound); + assert!( + select_existing_format_read_quorum(&formats, &errors, 4).is_none(), + "12 votes cannot compensate for a set with only one readable drive" + ); + formats[2] = None; + errors[2] = Some(DiskError::DiskNotFound); + assert!(select_existing_format_read_quorum(&formats, &errors, 4).is_none(), "11 votes must wait"); + } + + #[test] + fn existing_format_read_quorum_rejects_conflicts_invalid_slots_and_blank_disks() { + for failure in [ + "conflict", + "wrong_slot", + "nil", + "duplicate", + "blank", + "corrupt", + "access_denied", + "tie", + ] { + let (mut formats, mut errors) = half_pool_formats(); + match failure { + "conflict" => formats[1].as_mut().expect("second format").id = Uuid::new_v4(), + "wrong_slot" => { + let first_id = formats[0].as_ref().expect("first format").erasure.this; + formats[1].as_mut().expect("second format").erasure.this = first_id; + } + "nil" => { + for format in formats.iter_mut().flatten() { + format.id = Uuid::nil(); + } + } + "duplicate" => { + for format in formats.iter_mut().flatten() { + format.erasure.sets[5][3] = format.erasure.sets[0][0]; + } + } + "blank" => errors[2] = Some(DiskError::UnformattedDisk), + "corrupt" => errors[2] = Some(DiskError::CorruptedFormat), + "access_denied" => errors[2] = Some(DiskError::FileAccessDenied), + "tie" => { + let other = FormatV3::new(6, 4); + for slot in 0..24 { + if formats[slot].is_none() { + let mut format = other.clone(); + format.erasure.this = other.erasure.sets[slot / 4][slot % 4]; + formats[slot] = Some(format); + errors[slot] = None; + } + } + } + _ => unreachable!(), + } + assert!( + select_existing_format_read_quorum(&formats, &errors, 4).is_none(), + "{failure} must not authorize degraded startup" + ); + } + } + + #[tokio::test] + async fn existing_format_load_accepts_each_half_of_four_by_six_without_format_writes() { + let (_root, all_disks) = local_disks_with_set_width(24, 4).await; + let reference = FormatV3::new(6, 4); + for (slot, disk) in all_disks.iter().enumerate() { + let mut format = reference.clone(); + format.erasure.this = reference.erasure.sets[slot / 4][slot % 4]; + save_format_file(disk, &Some(format)).await.expect("persist existing format"); + } + let before = format_bytes(&all_disks).await; + for first in 0..4 { + for second in first + 1..4 { + let mut disks = all_disks.clone(); + for (slot, disk) in disks.iter_mut().enumerate() { + if slot % 4 != first && slot % 4 != second { + *disk = None; + } + } + let loaded = connect_load_init_formats_with_instance_ctx( + &Arc::new(InstanceContext::new()), + first == 0, + &mut disks, + 6, + 4, + None, + ) + .await + .expect("two nodes must load the existing format"); + assert_eq!(loaded.format.shared_identity(), reference.shared_identity()); + assert!(matches!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::None)); + assert_eq!(disks.iter().flatten().count(), 12); + assert_eq!(format_bytes(&all_disks).await, before, "read startup must never rewrite formats"); + } + } + } + #[tokio::test] async fn existing_format_load_succeeds_with_a_strict_majority() { let (_temp_dir, mut disks) = two_local_disks_with_missing_third().await; diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 712522e1a..5de55bb7e 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -237,7 +237,10 @@ pub async fn publish_ready_when_runtime_ready( wait_for_runtime_readiness_with( startup_runtime_readiness_max_wait(), STARTUP_RUNTIME_READINESS_POLL_INTERVAL, - collect_node_readiness, + // Publish service readiness at the same per-set read and shared-lock + // boundary reported by the node probe. Mutation paths independently + // require write quorum and a writable pool-metadata state. + || async { collect_node_readiness_report().await.readiness }, |dependency_readiness| { readiness.mark_stage(rustfs_common::SystemStage::FullReady); mark_get_metadata_read_version_coalescing_service_ready(); @@ -1060,20 +1063,6 @@ fn cluster_health_timeout_report() -> DependencyReadinessReport { } } -async fn collect_node_readiness() -> DependencyReadiness { - // Startup publication retains its local dependency gate. Runtime probes - // additionally report storage quorum without changing S3 admission. - let storage = node_pool_meta_write_readiness().await; - let readiness = DependencyReadiness { - storage_ready: storage.ready, - iam_ready: runtime_sources::current_iam_ready(), - lock_quorum_ready: collect_lock_quorum_status().await.ready, - peer_health_ready: collect_peer_health_readiness(), - }; - record_readiness_report(&dependency_readiness_report_from_write_status(readiness, storage)); - readiness -} - pub async fn collect_cluster_read_dependency_readiness_report() -> DependencyReadinessReport { let iam_ready_raw = runtime_sources::current_iam_ready(); let storage_ready = collect_storage_read_readiness_uncached().await; @@ -1116,14 +1105,6 @@ async fn collect_lock_quorum_observation() -> LockQuorumObservation { } } -async fn node_pool_meta_write_readiness() -> StorageWriteReadinessStatus { - if let Some(store) = runtime_sources::current_object_store_handle() { - return pool_metadata_write_readiness(store.pool_meta_write_status().await); - } - - StorageWriteReadinessStatus::default() -} - async fn collect_storage_write_readiness_uncached() -> StorageWriteReadinessStatus { if let Some(store) = runtime_sources::current_object_store_handle() { let metadata = pool_metadata_write_readiness(store.pool_meta_write_status().await); diff --git a/rustfs/src/startup_bucket_metadata.rs b/rustfs/src/startup_bucket_metadata.rs index ad77a0a79..eed955cff 100644 --- a/rustfs/src/startup_bucket_metadata.rs +++ b/rustfs/src/startup_bucket_metadata.rs @@ -15,7 +15,6 @@ use crate::app::object::OnDemandMigrationWriteBack; use crate::module_switches::{on_demand_migration_enabled_from_env, set_on_demand_migration_module_enabled}; use crate::on_demand_migration::OnDemandMigrationSys; -use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions}; use crate::storage_api::startup::bucket_metadata::{ ECStore, Error as StorageError, Result as StorageResult, get_global_replication_pool, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config, @@ -58,19 +57,13 @@ const STARTUP_BACKGROUND_STATUS_CANCELED: f64 = 3.0; pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc, ctx: &CancellationToken) -> IoResult> { let buckets_list = retry_metadata_startup(ctx, "list_buckets", || async { - store - .list_bucket(&BucketOptions { - no_metadata: true, - ..Default::default() - }) - .await - .map_err(IoError::other) + store.list_buckets_for_metadata_startup().await.map_err(IoError::other) }) .await?; let buckets: Vec = buckets_list.into_iter().map(|v| v.name).collect(); - retry_metadata_startup(ctx, "bucket_migration", || try_migrate_bucket_metadata(store.clone())).await?; + retry_metadata_startup(ctx, "bucket_migration", || try_migrate_bucket_metadata(store.clone(), &buckets)).await?; init_on_demand_migration_runtime(); init_bucket_metadata_sys(store.clone(), buckets.clone()).await; retry_metadata_startup(ctx, "iam_migration", || try_migrate_iam_config(store.clone())).await?; @@ -81,19 +74,13 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc, c pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: CancellationToken) -> IoResult> { let buckets_list = retry_metadata_startup(&ctx, "list_buckets", || async { - store - .list_bucket(&BucketOptions { - no_metadata: true, - ..Default::default() - }) - .await - .map_err(IoError::other) + store.list_buckets_for_metadata_startup().await.map_err(IoError::other) }) .await?; let buckets: Vec = buckets_list.into_iter().map(|v| v.name).collect(); - retry_metadata_startup(&ctx, "bucket_migration", || try_migrate_bucket_metadata(store.clone())).await?; + retry_metadata_startup(&ctx, "bucket_migration", || try_migrate_bucket_metadata(store.clone(), &buckets)).await?; retry_metadata_startup(&ctx, "iam_migration", || try_migrate_iam_config(store.clone())).await?; init_on_demand_migration_runtime(); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 142948502..adc83492a 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -1196,8 +1196,8 @@ pub(crate) fn get_global_transition_state() -> Arc { ecstore_bucket::lifecycle::bucket_lifecycle_ops::get_global_transition_state() } -pub(crate) async fn try_migrate_bucket_metadata(store: Arc) -> std::io::Result<()> { - ecstore_bucket::migration::try_migrate_bucket_metadata(store) +pub(crate) async fn try_migrate_bucket_metadata(store: Arc, buckets: &[String]) -> std::io::Result<()> { + ecstore_bucket::migration::try_migrate_bucket_metadata_for_buckets(store, buckets) .await .map_err(ecstore_bucket::migration::migration_startup_error) }