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
2026-10-12 23:52:45 +00:00 · 2026-10-11 11:50:24 +08:00
parent a424d10fa0
commit b8889fccc9
12 changed files with 708 additions and 69 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=efe68675f4676a0d2901242c5815b1f7af5cdec8231f7cccd408f6dacf5c75db
-sha256-linux=003ed6480935a0a95680cc061be143722fbd37bd477b8ff98607b1a064f3aef1
+sha256-darwin=5dbb9154b8feb94cedc8fb4aa6895d4b3259cc31b7df1f5c3b9928ef07f04535
+sha256-linux=d3a210be9eccf593d3723ae669b66680e9751dd01a6949f03df20614f913b707
@@ -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<dyn std::error::Error + Send + Sync>> {
+        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(
@@ -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::<Vec<_>>();
+        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::<Vec<_>>()
+        .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::<Vec<_>>();
+    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::<Vec<_>>();
+        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::<Vec<_>>();
+            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<u8>, writer_id: usize) -> Result<(), String> {
    client
        .put_object()
@@ -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,
        };
    }

@@ -273,6 +273,30 @@ where

    let buckets: Vec<String> = 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<S>(store: Arc<S>, 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(());
@@ -123,8 +123,18 @@ impl SetDisks {
    }

    pub(crate) async fn list_bucket_for_scanner(&self, _opts: &BucketOptions) -> Result<(Vec<BucketInfo>, bool)> {
+        self.list_bucket_with_quorum(BucketInfoQuorum::Write).await
+    }
+
+    pub(crate) async fn list_bucket_with_quorum(&self, quorum: BucketInfoQuorum) -> Result<(Vec<BucketInfo>, 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::<Vec<_>>();
        buckets.sort_by(|left, right| left.name.cmp(&right.name));
        Ok((buckets, topology_complete))
@@ -1231,34 +1231,46 @@ impl ECStore {
    #[instrument(skip(self))]
    pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
        // 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<crate::cluster::rpc::ScannerBucketListing> {
-        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<Vec<BucketInfo>> {
+        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<crate::cluster::rpc::ScannerBucketListing> {
        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::<Vec<_>>();
+                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() {
@@ -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));
@@ -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<FormatV3>], 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<FormatV3>],
+    errors: &[Option<DiskError>],
+    set_drive_count: usize,
+) -> Option<(FormatV3, Vec<bool>)> {
+    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<FormatV3>], slot_offset: usize) -> Result<(FormatV3, Vec<bool>)> {
    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<Option<FormatV3>>, Vec<Option<DiskError>>) {
+        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::<Vec<_>>();
+        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::<Vec<_>>();
+                let mut errors = formats
+                    .iter()
+                    .map(|format| format.is_none().then_some(DiskError::DiskNotFound))
+                    .collect::<Vec<_>>();
+                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;
@@ -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);
@@ -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<ECStore>, ctx: &CancellationToken) -> IoResult<Vec<String>> {
    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<String> = 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<ECStore>, c

 pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: CancellationToken) -> IoResult<Vec<String>> {
    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<String> = 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();
@@ -1196,8 +1196,8 @@ pub(crate) fn get_global_transition_state() -> Arc<TransitionState> {
    ecstore_bucket::lifecycle::bucket_lifecycle_ops::get_global_transition_state()
 }

-pub(crate) async fn try_migrate_bucket_metadata(store: Arc<ECStore>) -> std::io::Result<()> {
-    ecstore_bucket::migration::try_migrate_bucket_metadata(store)
+pub(crate) async fn try_migrate_bucket_metadata(store: Arc<ECStore>, 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)
 }