From 3391528025984690da7acb0e7913cbe2b140d3a5 Mon Sep 17 00:00:00 2001
From: houseme <housemecn@gmail.com>
Date: Thu, 10 Sep 2026 15:34:13 +0800
Subject: [PATCH] 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 |  5 +-
 crates/ecstore/src/store/bucket.rs        |  9 +++
 crates/ecstore/src/store/init.rs          | 74 +++++++++++++++++++++++
 3 files changed, 87 insertions(+), 1 deletion(-)

diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs
index fe376f343..8f39640c5 100644
--- a/crates/ecstore/src/bucket/metadata_sys.rs
+++ b/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
             },
         )
diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs
index 73ba00fd5..73aaa59af 100644
--- a/crates/ecstore/src/store/bucket.rs
+++ b/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,
diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs
index 11b2ed59e..c2e087acc 100644
--- a/crates/ecstore/src/store/init.rs
+++ b/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() {