fix(ecstore): apply read quorum to bucket validation

crates/e2e_test/src/namespace_lock_quorum_test.rs

@@ -25,6 +25,190 @@ const KEY: &str = "thumb/79/concurrent-overwrite.jpg";
 
 type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
 
+async fn assert_quorum_object_body(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
+    let body = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await?
+        .body
+        .collect()
+        .await?
+        .into_bytes();
+    assert_eq!(body.as_ref(), expected, "quorum read returned incorrect contents for {key}");
+    Ok(())
+}
+
+async fn wait_for_quorum_read_admission(clients: &[Client], bucket: &str) -> TestResult {
+    // SIGKILL can orphan a granted lease. Wait for shared metadata-lock
+    // admission before asserting the stable quorum boundary; cold bodies
+    // remain unread throughout this readiness probe.
+    let deadline =
+        tokio::time::Instant::now() + rustfs_lock::fast_lock::DEFAULT_LOCK_TIMEOUT + std::time::Duration::from_secs(15);
+    loop {
+        let mut ready = true;
+        for client in clients {
+            for key in ["warm-small", "warm-large"] {
+                match client.head_object().bucket(bucket).key(key).send().await {
+                    Ok(_) => {}
+                    Err(error) if error.raw_response().is_some_and(|response| response.status().as_u16() == 503) => {
+                        ready = false;
+                        break;
+                    }
+                    Err(error) => return Err(error.into()),
+                }
+            }
+            if !ready {
+                break;
+            }
+        }
+        if ready {
+            return Ok(());
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(format!("read quorum did not become available after lease convergence for {bucket}").into());
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    }
+}
+
+#[tokio::test]
+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)] {
+        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}"));
+        // 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");
+        cluster.set_env("RUST_LOG", "warn,rustfs_lock=debug");
+        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!("read-quorum-{node_count}-{parity}");
+        clients[0].create_bucket().bucket(&bucket).send().await?;
+        let small = b"read quorum is derived from the erasure layout".to_vec();
+        let large = (0..1_048_576)
+            .map(|index| u8::try_from(index % 251).expect("bounded payload byte"))
+            .collect::<Vec<_>>();
+        for (key, body) in [
+            ("warm-small", &small),
+            ("warm-large", &large),
+            ("cold-small", &small),
+            ("cold-large", &large),
+            ("below-quorum", &large),
+        ] {
+            clients[node_count - 1]
+                .put_object()
+                .bucket(&bucket)
+                .key(key)
+                .body(Bytes::copy_from_slice(body).into())
+                .send()
+                .await?;
+        }
+        for node in &cluster.nodes {
+            for key in ["warm-small", "warm-large", "cold-small", "cold-large", "below-quorum"] {
+                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 be complete before fault injection: {census:?}");
+                assert_eq!(census.data_blocks, Some(read_quorum));
+                assert_eq!(census.parity_blocks, Some(parity));
+            }
+        }
+        for client in &clients {
+            assert_quorum_object_body(client, &bucket, "warm-small", &small).await?;
+            assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
+        }
+
+        for offline_node in (read_quorum..node_count).rev() {
+            cluster.stop_node(offline_node)?;
+            wait_for_quorum_read_admission(&clients[..offline_node], &bucket).await?;
+            for client in clients.iter().take(offline_node) {
+                client.head_bucket().bucket(&bucket).send().await?;
+                assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
+            }
+        }
+
+        // Exercise more than the five-second positive bucket-validation TTL.
+        // Every sample must succeed; polling must not hide a transient failure.
+        let validation_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(6);
+        loop {
+            for client in clients.iter().take(read_quorum) {
+                assert_quorum_object_body(client, &bucket, "warm-small", &small).await?;
+                assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
+                let listing = client.list_objects_v2().bucket(&bucket).send().await?;
+                for key in ["warm-small", "warm-large", "cold-small", "cold-large", "below-quorum"] {
+                    assert!(listing.contents().iter().any(|entry| entry.key() == Some(key)), "listing omitted {key}");
+                }
+            }
+            if tokio::time::Instant::now() >= validation_deadline {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(250)).await;
+        }
+        for client in clients.iter().take(read_quorum) {
+            assert_quorum_object_body(client, &bucket, "cold-small", &small).await?;
+            assert_quorum_object_body(client, &bucket, "cold-large", &large).await?;
+        }
+
+        let write = clients[0]
+            .put_object()
+            .bucket(&bucket)
+            .key("quorum-write")
+            .body(Bytes::copy_from_slice(&small).into())
+            .send()
+            .await;
+        if read_quorum >= write_quorum {
+            write?;
+        } else {
+            let error = write.expect_err("a read quorum must not authorize a write that needs more votes");
+            assert_eq!(error.as_service_error().and_then(|error| error.meta().code()), Some("ServiceUnavailable"));
+        }
+
+        cluster.stop_node(read_quorum - 1)?;
+        for client in clients.iter().take(read_quorum - 1) {
+            match client.get_object().bucket(&bucket).key("below-quorum").send().await {
+                Ok(response) => assert!(
+                    response.body.collect().await.is_err(),
+                    "fewer than {read_quorum} valid fragments must not reconstruct an uncached object"
+                ),
+                Err(error) => assert_eq!(
+                    error.as_service_error().and_then(|error| error.meta().code()),
+                    Some("ServiceUnavailable"),
+                    "a quorum loss must not be mistaken for a missing object"
+                ),
+            }
+        }
+
+        for node in 0..read_quorum - 1 {
+            cluster.stop_node(node)?;
+        }
+        cluster.start().await?;
+        for client in &clients {
+            assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
+            assert_quorum_object_body(client, &bucket, "below-quorum", &large).await?;
+        }
+    }
+
+    Ok(())
+}
+
 async fn put_object(client: Client, payload: Vec<u8>, writer_id: usize) -> Result<(), String> {
     client
         .put_object()

crates/ecstore/src/set_disk/metadata.rs

@@ -326,14 +326,15 @@ impl SetDisks {
         let parity_blocks = Self::common_parity(&parities, default_parity_count as i32);
 
         if parity_blocks < 0 {
-            // No parity value reached read quorum. Distinguish two cases:
-            // enough disks answered with valid-looking metadata that simply
-            // cannot be reconciled (corrupt/foreign entries — retrying cannot
-            // help, and heal should see Corrupt, rustfs#5801) versus too few
-            // healthy answers (a genuine quorum condition where retry may
-            // succeed once disks recover).
+            // A consistent layout can require more replies than the initial
+            // half-set probe. Reaching that probe alone is not corruption;
+            // only invalid or conflicting healthy replies establish that.
             let healthy_replies = errs.iter().filter(|err| err.is_none()).count();
-            if healthy_replies >= expected_rquorum {
+            let consistent_parity = parities
+                .iter()
+                .find(|&&parity| parity >= 0)
+                .filter(|&&parity| parities.iter().filter(|&&candidate| candidate == parity).count() == healthy_replies);
+            if healthy_replies >= expected_rquorum && consistent_parity.is_none() {
                 error!(
                     "object_quorum_from_meta: irreconcilable parity across {healthy_replies} healthy replies (corrupt metadata), errs={errs:?}"
                 );
@@ -1652,6 +1653,40 @@ mod tests {
         assert_eq!(err, DiskError::FileCorrupt);
     }
 
+    #[test]
+    fn consistent_parity_below_its_data_shard_quorum_is_not_corruption() {
+        for (drive_count, parity) in [(6, 2), (8, 2), (12, 4)] {
+            let data = drive_count - parity;
+            let mut metas = (1..=drive_count)
+                .map(|index| {
+                    let mut info = FileInfo::new("bucket/object", data, parity);
+                    info.size = 1024;
+                    info.erasure.index = index;
+                    info
+                })
+                .collect::<Vec<_>>();
+            let mut errs = vec![Some(DiskError::DiskNotFound); drive_count];
+            errs[..data].fill(None);
+            assert_eq!(
+                SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect("exact data quorum should resolve"),
+                (data as i32, data as i32)
+            );
+
+            errs[data - 1] = Some(DiskError::DiskNotFound);
+            assert_eq!(
+                SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect_err("one fewer shard cannot resolve"),
+                DiskError::ErasureReadQuorum,
+                "layout {drive_count}/{parity} has consistent metadata but insufficient shards"
+            );
+
+            metas[0].erasure.parity_blocks = usize::MAX;
+            assert_eq!(
+                SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect_err("corrupt healthy replies must be rejected"),
+                DiskError::FileCorrupt
+            );
+        }
+    }
+
     /// Too few healthy replies remains a genuine quorum condition where a
     /// retry may succeed once disks recover.
     #[test]

crates/ecstore/src/set_disk/mod.rs

@@ -865,6 +865,7 @@ pub(crate) use core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, ren
 mod ctx;
 mod metadata;
 mod ops;
+pub(crate) use ops::bucket::BucketInfoQuorum;
 
 #[cfg(test)]
 pub(crate) use ops::hermetic_set_disks_isolated;

crates/ecstore/src/set_disk/ops/bucket.rs

@@ -21,12 +21,72 @@
 
 use super::super::{
     BUCKET_OP_IGNORED_ERRS, BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, DiskError, Error, HashMap,
-    MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_write_quorum_errs,
+    MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_read_quorum_errs,
+    reduce_write_quorum_errs,
 };
 use crate::api::bucket::metadata_sys;
 use crate::disk::DiskAPI;
 
+#[derive(Clone, Copy)]
+pub(crate) enum BucketInfoQuorum {
+    Read,
+    Write,
+}
+
 impl SetDisks {
+    pub(crate) async fn stat_bucket_with_quorum(&self, bucket: &str, quorum: BucketInfoQuorum) -> Result<BucketInfo> {
+        let disks = self.disk_inventory().await;
+        let disk_count = disks.len();
+        let mut futures = Vec::with_capacity(disk_count);
+        for disk in disks {
+            let bucket = bucket.to_string();
+            futures.push(async move {
+                match disk {
+                    Some(disk) => disk.stat_volume(&bucket).await,
+                    None => Err(DiskError::DiskNotFound),
+                }
+            });
+        }
+
+        let results = join_all(futures).await;
+        let mut infos = Vec::with_capacity(results.len());
+        let mut errs = Vec::with_capacity(results.len());
+        for result in results {
+            match result {
+                Ok(info) => {
+                    infos.push(Some(info));
+                    errs.push(None);
+                }
+                Err(err) => {
+                    infos.push(None);
+                    errs.push(Some(err));
+                }
+            }
+        }
+
+        let error = match quorum {
+            // Bucket mutations use a majority regardless of object storage
+            // class. A namespace read must intersect that majority; object
+            // readers still enforce the persisted layout's data-shard quorum.
+            BucketInfoQuorum::Read => reduce_read_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count.div_ceil(2).max(1)),
+            BucketInfoQuorum::Write => reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count / 2 + 1),
+        };
+        if let Some(err) = error {
+            return Err(err.into());
+        }
+
+        infos
+            .into_iter()
+            .flatten()
+            .next()
+            .map(|info| BucketInfo {
+                name: info.name,
+                created: info.created,
+                ..Default::default()
+            })
+            .ok_or(Error::VolumeNotFound)
+    }
+
     pub(crate) async fn list_bucket_for_scanner(&self, _opts: &BucketOptions) -> Result<(Vec<BucketInfo>, bool)> {
         let disks = self.disk_inventory().await;
         let write_quorum = (disks.len() / 2) + 1;
@@ -131,59 +191,12 @@ impl BucketOperations for SetDisks {
 
     #[tracing::instrument(skip(self))]
     async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
-        let disks = self.disk_inventory().await;
-        let write_quorum = (disks.len() / 2) + 1;
-
-        let mut futures = Vec::with_capacity(disks.len());
-        for disk in disks {
-            let bucket = bucket.to_string();
-            futures.push(async move {
-                match disk {
-                    Some(disk) => disk.stat_volume(&bucket).await,
-                    None => Err(DiskError::DiskNotFound),
-                }
-            });
-        }
-
-        let results = join_all(futures).await;
-        let mut infos = Vec::with_capacity(results.len());
-        let mut errs = Vec::with_capacity(results.len());
-        for result in results {
-            match result {
-                Ok(info) => {
-                    infos.push(Some(info));
-                    errs.push(None);
-                }
-                Err(err) => {
-                    infos.push(None);
-                    errs.push(Some(err));
-                }
-            }
-        }
-
-        if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
-            return Err(err.into());
-        }
-
-        let mut versioning = false;
-        let mut object_locking = false;
+        let mut info = self.stat_bucket_with_quorum(bucket, BucketInfoQuorum::Write).await?;
         if let Ok(sys) = metadata_sys::get(bucket).await {
-            versioning = sys.versioning();
-            object_locking = sys.object_locking();
+            info.versioning = sys.versioning();
+            info.object_locking = sys.object_locking();
         }
-
-        infos
-            .into_iter()
-            .flatten()
-            .next()
-            .map(|info| BucketInfo {
-                name: info.name,
-                created: info.created,
-                versioning,
-                object_locking,
-                ..Default::default()
-            })
-            .ok_or(Error::VolumeNotFound)
+        Ok(info)
     }
 
     #[tracing::instrument(skip(self))]

crates/ecstore/src/store/bucket.rs

@@ -19,7 +19,7 @@ use crate::bucket::{
 };
 use crate::error::is_err_bucket_not_found;
 use crate::runtime::sources as runtime_sources;
-use crate::set_disk::get_lock_acquire_timeout;
+use crate::set_disk::{BucketInfoQuorum, get_lock_acquire_timeout};
 use crate::storage_api_contracts::bucket::{BUCKET_LIFECYCLE_LOCK_OBJECT, SRBucketDeleteOp};
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
 use futures::stream::{self, StreamExt};
@@ -772,17 +772,30 @@ impl ECStore {
 
     #[instrument(skip(self))]
     pub(crate) async fn get_bucket_info_from_sets(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
+        self.get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Write)
+            .await
+    }
+
+    async fn get_bucket_info_from_sets_with_quorum(
+        &self,
+        bucket: &str,
+        opts: &BucketOptions,
+        quorum: BucketInfoQuorum,
+    ) -> Result<BucketInfo> {
         // One host may participate in several pools after expansion. Resolve the
         // namespace against each erasure set so disks from different pools can
         // never be combined into one bucket quorum.
         // Bucket validation is request-path IO. Keep the previous peer fanout's
         // latency shape by probing every set concurrently; scanner listings use
         // a separate bounded path below because they run continuously.
-        let mut scoped_results =
-            futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move {
-                (pool_index, set_index, set.get_bucket_info(bucket, opts).await)
-            }))
-            .await;
+        let mut scoped_results = futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move {
+            let result = match quorum {
+                BucketInfoQuorum::Read => set.stat_bucket_with_quorum(bucket, quorum).await,
+                BucketInfoQuorum::Write => set.get_bucket_info(bucket, opts).await,
+            };
+            (pool_index, set_index, result)
+        }))
+        .await;
         scoped_results.sort_unstable_by_key(|(pool_index, set_index, _)| (*pool_index, *set_index));
 
         let mut first_info = None;
@@ -806,7 +819,11 @@ impl ECStore {
 
     #[instrument(skip(self))]
     pub(super) async fn handle_get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
-        let mut info = self.get_bucket_info_from_sets(bucket, opts).await?;
+        let mut info = match self.get_bucket_info_from_sets(bucket, opts).await {
+            Ok(info) => info,
+            Err(Error::ErasureWriteQuorum) => return self.get_bucket_info_at_read_quorum(bucket, opts).await,
+            Err(err) => return Err(err),
+        };
 
         if let Ok(sys) = metadata_sys::get_in(&self.ctx, bucket).await {
             if should_override_created_from_metadata(sys.created) {
@@ -819,6 +836,35 @@ impl ECStore {
         Ok(info)
     }
 
+    async fn get_bucket_info_at_read_quorum(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
+        // Lock order: bucket lifecycle -> internal metadata object read locks.
+        // Keep create/delete from changing the namespace while a read quorum
+        // confirms both physical presence and persisted bucket metadata.
+        let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
+        await_bucket_namespace_operation(Some(&guard), bucket, "bucket read quorum validation", async {
+            let mut info = self
+                .get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Read)
+                .await?;
+            let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
+            if !persisted {
+                // A minority of directories left by failed creation is not an
+                // authoritative bucket. Never turn fabricated defaults into
+                // permission to serve degraded reads.
+                return Err(Error::ErasureReadQuorum);
+            }
+            if metadata.name != bucket {
+                return Err(Error::FileCorrupt);
+            }
+            if should_override_created_from_metadata(metadata.created) {
+                info.created = Some(metadata.created);
+            }
+            info.versioning = metadata.versioning();
+            info.object_locking = metadata.object_locking();
+            Ok(info)
+        })
+        .await
+    }
+
     #[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
@@ -1049,7 +1095,7 @@ mod tests {
         run_physical_bucket_deletion, scan_metadata_less_residue, scan_metadata_less_residue_with_budget,
         should_override_created_from_metadata, validate_table_bucket_delete_allowed,
     };
-    use crate::bucket::metadata::table_bucket_catalog_metadata_prefix;
+    use crate::bucket::metadata::{BucketMetadata, table_bucket_catalog_metadata_prefix};
     use crate::bucket::metadata_sys;
     use crate::cluster::rpc::peer_s3_client::install_delete_bucket_empty_scan_barrier;
     use crate::disk::{BUCKET_META_PREFIX, DiskAPI, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE};
@@ -1076,6 +1122,7 @@ mod tests {
     use std::sync::atomic::{AtomicBool, Ordering};
     use std::time::{Duration, SystemTime};
     use time::OffsetDateTime;
+    use tokio::io::AsyncReadExt;
     use tokio::sync::{Notify, OnceCell};
     use tokio_util::sync::CancellationToken;
     use uuid::Uuid;
@@ -1359,11 +1406,18 @@ mod tests {
     }
 
     async fn setup_multi_pool_bucket_test_env() -> (tempfile::TempDir, Arc<ECStore>) {
+        setup_bucket_quorum_test_env(&[4, 4], None).await
+    }
+
+    async fn setup_bucket_quorum_test_env(
+        drives_per_pool: &[usize],
+        standard_parity: Option<usize>,
+    ) -> (tempfile::TempDir, Arc<ECStore>) {
         let temp_dir = tempfile::tempdir().expect("multi-pool bucket test directory should be created");
         let mut pools = Vec::new();
-        for pool_index in 0..2 {
+        for (pool_index, &drive_count) in drives_per_pool.iter().enumerate() {
             let mut endpoints = Vec::new();
-            for disk_index in 0..4 {
+            for disk_index in 0..drive_count {
                 let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
                 tokio::fs::create_dir_all(&disk_path)
                     .await
@@ -1378,7 +1432,7 @@ mod tests {
             pools.push(PoolEndpoints {
                 legacy: false,
                 set_count: 1,
-                drives_per_set: 4,
+                drives_per_set: drive_count,
                 endpoints: Endpoints::from(endpoints),
                 cmd_line: format!("bucket-test-pool-{pool_index}"),
                 platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
@@ -1399,9 +1453,12 @@ mod tests {
         )
         .await
         .expect("multi-pool ECStore should initialize");
-        let storage_class =
-            crate::config::storageclass::lookup_config_for_pools_without_env(&rustfs_config::server_config::KVS::new(), &[4, 4])
-                .expect("multi-pool storage class should match both four-disk pools");
+        let mut storage_class_kvs = rustfs_config::server_config::KVS::new();
+        if let Some(parity) = standard_parity {
+            storage_class_kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), format!("EC:{parity}"));
+        }
+        let storage_class = crate::config::storageclass::lookup_config_for_pools_without_env(&storage_class_kvs, drives_per_pool)
+            .expect("storage class should match every test erasure set");
         for pool in &ecstore.pools {
             for set in &pool.disk_set {
                 set.set_test_storage_class_config(storage_class.clone());
@@ -2067,6 +2124,218 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn bucket_info_read_quorum_tracks_erasure_layout() {
+        for (drive_count, parity) in [(2, 1), (3, 1), (4, 2), (5, 2), (6, 3), (8, 4), (6, 2), (12, 6)] {
+            let (_temp_dir, store) = setup_bucket_quorum_test_env(&[drive_count], Some(parity)).await;
+            metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+            let bucket = format!("read-quorum-{drive_count}-{parity}");
+            let object = "uncached-object";
+            let body = b"erasure read quorum must follow the persisted layout".repeat(32_768);
+            store
+                .make_bucket(&bucket, &MakeBucketOptions::default())
+                .await
+                .expect("healthy namespace should accept bucket creation");
+            store
+                .put_object(&bucket, object, &mut PutObjReader::from_vec(body.clone()), &ObjectOptions::default())
+                .await
+                .expect("healthy erasure set should accept the seed object");
+            let set = &store.pools[0].disk_set[0];
+            let lock = set
+                .new_ns_lock(&bucket, object)
+                .await
+                .expect("seed namespace lock should resolve");
+            drop(
+                lock.get_write_lock(Duration::from_secs(30))
+                    .await
+                    .expect("seed physical fanout must finish before taking disks offline"),
+            );
+            if (drive_count, parity) == (6, 3) {
+                let mut kvs = rustfs_config::server_config::KVS::new();
+                kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), "EC:2".to_string());
+                set.set_test_storage_class_config(
+                    crate::config::storageclass::lookup_config_for_pools_without_env(&kvs, &[drive_count])
+                        .expect("a later storage-class change must not raise old objects' read quorum"),
+                );
+            }
+
+            let offline_indexes = (0..parity).collect::<Vec<_>>();
+            let offline = take_set_disks_offline(&store, set, &offline_indexes).await;
+            let info = store
+                .get_bucket_info(&bucket, &BucketOptions::default())
+                .await
+                .expect("bucket validation must admit the object's exact read quorum");
+            assert_eq!(info.name, bucket);
+
+            let mut reader = store
+                .get_object_reader(&bucket, object, None, Default::default(), &ObjectOptions::default())
+                .await
+                .expect("the persisted layout should remain readable at its exact data-shard quorum");
+            let mut restored = Vec::new();
+            reader
+                .stream
+                .read_to_end(&mut restored)
+                .await
+                .expect("quorum read should reconstruct the body");
+            assert_eq!(restored, body, "layout {drive_count}/{parity} must retain exact object contents");
+            drop(reader);
+
+            if drive_count - parity == drive_count / 2 {
+                let error = store
+                    .get_bucket_info_from_sets(&bucket, &BucketOptions::default())
+                    .await
+                    .expect_err("bucket mutations must retain their majority namespace check");
+                assert_eq!(error, StorageError::ErasureWriteQuorum);
+            }
+
+            let below_quorum = take_set_disks_offline(&store, set, &[parity]).await;
+            let read = store
+                .get_object_reader(&bucket, object, None, Default::default(), &ObjectOptions::default())
+                .await;
+            match read {
+                Ok(mut reader) => assert!(
+                    reader.stream.read_to_end(&mut Vec::new()).await.is_err(),
+                    "layout {drive_count}/{parity} must reject fewer than its data-shard quorum"
+                ),
+                Err(error) => assert!(
+                    matches!(error, StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _)),
+                    "a missing shard must report read quorum loss, got {error}"
+                ),
+            }
+            restore_set_disks(&store, set, below_quorum).await;
+            restore_set_disks(&store, set, offline).await;
+        }
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn bucket_info_read_quorum_is_scoped_to_each_erasure_set() {
+        let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4, 6], None).await;
+        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = "read-quorum-mixed-pools";
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("healthy pools should accept bucket creation");
+
+        let first_set = &store.pools[0].disk_set[0];
+        let second_set = &store.pools[1].disk_set[0];
+        let first_offline = take_set_disks_offline(&store, first_set, &[0, 1]).await;
+        let second_offline = take_set_disks_offline(&store, second_set, &[0, 1, 2]).await;
+        store
+            .get_bucket_info(bucket, &BucketOptions::default())
+            .await
+            .expect("each set independently satisfies its namespace read quorum");
+
+        for (set, extra_disk) in [(first_set, 2), (second_set, 3)] {
+            let extra_offline = take_set_disks_offline(&store, set, &[extra_disk]).await;
+            assert_eq!(
+                store
+                    .get_bucket_info(bucket, &BucketOptions::default())
+                    .await
+                    .expect_err("another pool must not subsidize a set below its read quorum"),
+                StorageError::ErasureReadQuorum
+            );
+            restore_set_disks(&store, set, extra_offline).await;
+        }
+        restore_set_disks(&store, first_set, first_offline).await;
+        restore_set_disks(&store, second_set, second_offline).await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn bucket_info_read_quorum_requires_authoritative_metadata() {
+        for state in ["missing", "corrupt", "foreign", "incarnation"] {
+            let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await;
+            metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+            let bucket = format!("read-quorum-{state}-metadata");
+            let mut metadata = if state == "missing" {
+                store
+                    .make_bucket_on_sets(&bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("simulate directories left before bucket metadata is published");
+                BucketMetadata::new(&bucket)
+            } else {
+                store
+                    .make_bucket(&bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("healthy bucket should publish metadata");
+                metadata_sys::get_in(&store.ctx, &bucket)
+                    .await
+                    .expect("seed metadata should be cached")
+                    .as_ref()
+                    .clone()
+            };
+            let path = metadata.save_file_path();
+            match state {
+                "corrupt" => crate::config::com::save_config(store.clone(), &path, b"corrupt".to_vec())
+                    .await
+                    .expect("persist corrupt metadata while the cached copy remains valid"),
+                "foreign" => {
+                    metadata.name = "different-bucket".to_string();
+                    let mut encoded = vec![1, 0, 1, 0];
+                    encoded.extend(metadata.marshal_msg().expect("foreign metadata should encode"));
+                    crate::config::com::save_config(store.clone(), &path, encoded)
+                        .await
+                        .expect("persist metadata for a different bucket at the requested path");
+                }
+                "incarnation" => crate::bucket::metadata::save_bucket_incarnation(store.clone(), &bucket, Uuid::new_v4())
+                    .await
+                    .expect("persist a different bucket generation"),
+                _ => {}
+            }
+
+            let set = &store.pools[0].disk_set[0];
+            let offline = take_set_disks_offline(&store, set, &[0, 1]).await;
+            let error = store
+                .get_bucket_info(&bucket, &BucketOptions::default())
+                .await
+                .expect_err("read admission must not trust residual directories or cached metadata");
+            match state {
+                "missing" => assert_eq!(error, StorageError::ErasureReadQuorum),
+                "foreign" => assert_eq!(error, StorageError::FileCorrupt),
+                "incarnation" => assert!(error.to_string().contains("sidecar does not match bucket metadata")),
+                "corrupt" => assert!(error.to_string().contains("format invalid"), "unexpected corruption error: {error}"),
+                _ => unreachable!(),
+            }
+            restore_set_disks(&store, set, offline).await;
+        }
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn bucket_info_read_quorum_accepts_persisted_legacy_metadata() {
+        let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await;
+        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = "interop";
+        store
+            .make_bucket_on_sets(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("legacy bucket directories should exist");
+        let hex = include_str!("../../tests/fixtures/minio/bucket_metadata.blob.hex")
+            .split_whitespace()
+            .collect::<String>();
+        let body = (0..hex.len())
+            .step_by(2)
+            .map(|index| u8::from_str_radix(&hex[index..index + 2], 16).expect("pinned MinIO metadata fixture"))
+            .collect();
+        crate::config::com::save_config(store.clone(), &BucketMetadata::new(bucket).save_file_path(), body)
+            .await
+            .expect("legacy metadata should be persisted without an incarnation sidecar");
+
+        let set = &store.pools[0].disk_set[0];
+        let offline = take_set_disks_offline(&store, set, &[0, 1]).await;
+        let info = store
+            .get_bucket_info(bucket, &BucketOptions::default())
+            .await
+            .expect("persisted MinIO metadata should authorize reads at the namespace read quorum");
+        assert_eq!(info.name, bucket);
+        assert!(info.versioning);
+        assert!(info.object_locking);
+        restore_set_disks(&store, set, offline).await;
+    }
+
     #[tokio::test]
     #[serial]
     async fn bucket_namespace_reads_report_missing_when_every_set_is_absent() {
@@ -2090,6 +2359,7 @@ mod tests {
     #[serial]
     async fn bucket_namespace_reads_fail_closed_when_any_set_loses_quorum() {
         let (_temp_dir, ecstore) = setup_multi_pool_bucket_test_env().await;
+        metadata_sys::init_bucket_metadata_sys(ecstore.clone(), Vec::new()).await;
         let bucket = format!("degraded-expansion-{}", Uuid::new_v4().simple());
         ecstore.pools[0].disk_set[0]
             .make_bucket(&bucket, &MakeBucketOptions::default())
@@ -2097,6 +2367,7 @@ mod tests {
             .expect("bucket should be created in the original pool only");
         ecstore.pools[1].disk_set[0].disks.write().await[0] = None;
         ecstore.pools[1].disk_set[0].disks.write().await[1] = None;
+        ecstore.pools[1].disk_set[0].disks.write().await[2] = None;
 
         let list_err = ecstore
             .list_bucket(&BucketOptions::default())
@@ -2108,7 +2379,7 @@ mod tests {
             .get_bucket_info(&bucket, &BucketOptions::default())
             .await
             .expect_err("bucket validation must fail when an expansion pool is unavailable");
-        assert_eq!(info_err, StorageError::ErasureWriteQuorum);
+        assert_eq!(info_err, StorageError::ErasureReadQuorum);
     }
 
     #[tokio::test]