fix(ecstore): allow metadata snapshots at read quorum (#7627)

fix(ecstore): admit metadata snapshots at read quorum

Allow guarded bucket metadata snapshot existence checks to use read quorum so degraded erasure sets can continue Object Lock snapshot reads without weakening bucket mutation or object write quorum.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/bucket/metadata_sys.rs

@@ -2526,7 +2526,10 @@ impl BucketMetadataSys {
             "bucket metadata snapshot existence check",
             async {
                 self.object_store()
-                    .get_bucket_info_from_sets(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default())
+                    .get_bucket_info_from_sets_at_read_quorum(
+                        bucket,
+                        &crate::storage_api_contracts::bucket::BucketOptions::default(),
+                    )
                     .await
             },
         )

crates/ecstore/src/store/bucket.rs

@@ -801,6 +801,15 @@ impl ECStore {
             .await
     }
 
+    pub(crate) async fn get_bucket_info_from_sets_at_read_quorum(
+        &self,
+        bucket: &str,
+        opts: &BucketOptions,
+    ) -> Result<BucketInfo> {
+        self.get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Read)
+            .await
+    }
+
     async fn get_bucket_info_from_sets_with_quorum(
         &self,
         bucket: &str,

crates/ecstore/src/store/init.rs

@@ -3707,6 +3707,31 @@ mod tests {
         OfflineTestDisks { disks }
     }
 
+    #[cfg(feature = "test-util")]
+    async fn force_set_disk_range_offline_for_test(
+        set: &Arc<crate::set_disk::SetDisks>,
+        range: std::ops::Range<usize>,
+    ) -> OfflineTestDisks {
+        let disks = set
+            .disks
+            .read()
+            .await
+            .get(range)
+            .expect("offline test range must fit the set")
+            .iter()
+            .map(|disk| disk.clone().expect("fault-injection disk should start online"))
+            .collect::<Vec<_>>();
+        for disk in &disks {
+            disk.close().await.expect("fault injection should stop per-disk monitoring");
+            disk.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Offline);
+        }
+        set.connect_disks().await;
+        for disk in &disks {
+            assert_eq!(disk.runtime_state(), crate::disk::health_state::RuntimeDriveHealthState::Offline);
+        }
+        OfflineTestDisks { disks }
+    }
+
     fn active_rebalance_meta_for_pool(pool_count: usize, active_pool_idx: usize) -> RebalanceMeta {
         let now = OffsetDateTime::now_utc();
         let mut pool_stats = vec![RebalanceStats::default(); pool_count];
@@ -17322,6 +17347,55 @@ mod tests {
         assert_eq!(body, original_body);
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn object_lock_snapshot_uses_read_quorum_bucket_existence_probe() {
+        let temp = tempfile::tempdir().expect("create degraded snapshot store dir");
+        let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store_with_layout(
+            temp.path(),
+            "degraded-object-lock-snapshot",
+            &[(2, 12)],
+            CancellationToken::new(),
+            None,
+        ))
+        .await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+
+        let bucket = format!("degraded-ol-{}", uuid::Uuid::new_v4());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create snapshot bucket");
+        let expected_incarnation = store
+            .bucket_incarnation_id(&bucket)
+            .await
+            .expect("read bucket incarnation before degrading sets");
+
+        let mut offline_disks = Vec::new();
+        for set in store.all_set_disks() {
+            offline_disks.push(force_set_disk_range_offline_for_test(&set, 6..12).await);
+        }
+
+        let snapshot = store
+            .object_lock_config_snapshot(&bucket)
+            .await
+            .expect("read-quorum bucket existence should admit guarded Object Lock snapshot");
+        assert!(matches!(
+            snapshot.state(),
+            crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent
+        ));
+        assert!(snapshot.is_valid_for_destructive_put(store.id, &bucket, expected_incarnation));
+
+        let current_incarnation = crate::bucket::metadata_sys::get_object_lock_config_and_incarnation_from_disk_in(&ctx, &bucket)
+            .await
+            .expect("authoritative metadata read should also survive at read quorum")
+            .1;
+        assert_eq!(current_incarnation, expected_incarnation);
+
+        drop(offline_disks);
+    }
+
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]
     async fn force_create_existing_bucket_preserves_incarnation_and_inflight_request() {