fix(ecstore): preserve recovery quorum identity (#3937)

* fix(ecstore): preserve recovery quorum identity

* fix(get): hold read permits through response streaming

* fix(heal): queue read repair at low priority

* fix(ecstore): harden recovery read validation

crates/ecstore/src/cache_value/metacache_set.rs

@@ -27,6 +27,7 @@ use std::{
 };
 use tokio::io::AsyncRead;
 use tokio::spawn;
+use tokio::sync::Mutex as TokioMutex;
 use tokio::time::timeout;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, warn};
@@ -59,6 +60,10 @@ fn is_missing_path_error(err: &DiskError) -> bool {
     matches!(err, DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound)
 }
 
+async fn take_fallback_candidate<T>(fallback_items: &Arc<TokioMutex<VecDeque<T>>>) -> Option<T> {
+    fallback_items.lock().await.pop_front()
+}
+
 #[cfg(test)]
 #[derive(Clone)]
 pub(crate) enum TestReaderBehavior {
@@ -67,6 +72,7 @@ pub(crate) enum TestReaderBehavior {
     Stall,
     IgnoreCancel,
     ProducerError(DiskError),
+    PrimaryErrorThenFallback(DiskError),
     PartialThenTimeout(Vec<MetaCacheEntry>),
 }
 
@@ -89,6 +95,8 @@ pub struct ListPathRawOptions {
     #[cfg(test)]
     pub(crate) test_reader_behaviors: Vec<TestReaderBehavior>,
     #[cfg(test)]
+    pub(crate) test_fallback_reader_behaviors: Vec<TestReaderBehavior>,
+    #[cfg(test)]
     pub(crate) peek_timeout: Option<Duration>,
     // pub agreed: Option<Arc<dyn Fn(MetaCacheEntry) + Send + Sync>>,
     // pub partial: Option<Arc<dyn Fn(MetaCacheEntries, &[Option<Error>]) + Send + Sync>>,
@@ -112,6 +120,8 @@ impl Clone for ListPathRawOptions {
             #[cfg(test)]
             test_reader_behaviors: self.test_reader_behaviors.clone(),
             #[cfg(test)]
+            test_fallback_reader_behaviors: self.test_fallback_reader_behaviors.clone(),
+            #[cfg(test)]
             peek_timeout: self.peek_timeout,
             ..Default::default()
         }
@@ -128,7 +138,11 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
 
     let mut jobs: Vec<tokio::task::JoinHandle<std::result::Result<(), DiskError>>> = Vec::new();
     let mut readers = Vec::with_capacity(opts.disks.len());
-    let fds = opts.fallback_disks.iter().flatten().cloned().collect::<VecDeque<_>>();
+    let fds = Arc::new(TokioMutex::new(opts.fallback_disks.iter().flatten().cloned().collect::<VecDeque<_>>()));
+    #[cfg(test)]
+    let test_fallbacks = Arc::new(TokioMutex::new(
+        opts.test_fallback_reader_behaviors.iter().cloned().collect::<VecDeque<_>>(),
+    ));
     let max_disk_failures = opts.disks.len().saturating_sub(opts.min_disks);
     let producer_errs: Arc<[OnceLock<DiskError>]> = (0..opts.disks.len()).map(|_| OnceLock::new()).collect::<Vec<_>>().into();
 
@@ -137,14 +151,16 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
     for (disk_idx, disk) in opts.disks.iter().enumerate() {
         let opdisk = disk.clone();
         let opts_clone = opts.clone();
-        let mut fds_clone = fds.clone();
+        let fds_clone = fds.clone();
+        #[cfg(test)]
+        let test_fallbacks_clone = test_fallbacks.clone();
         let cancel_rx_clone = cancel_rx.clone();
         let producer_errs_clone = producer_errs.clone();
         let (rd, wr) = tokio::io::duplex(64);
         readers.push(MetacacheReader::new(rd));
         jobs.push(spawn(async move {
             #[cfg(test)]
-            if let Some(behavior) = opts_clone.test_reader_behaviors.get(disk_idx).cloned() {
+            let test_primary_error = if let Some(behavior) = opts_clone.test_reader_behaviors.get(disk_idx).cloned() {
                 match behavior {
                     TestReaderBehavior::Eof => return Ok(()),
                     TestReaderBehavior::Entries(entries) => {
@@ -168,6 +184,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
                         record_producer_error(&producer_errs_clone, disk_idx, &err);
                         return Err(err);
                     }
+                    TestReaderBehavior::PrimaryErrorThenFallback(err) => Some(err),
                     TestReaderBehavior::PartialThenTimeout(entries) => {
                         let mut wr = wr;
                         let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr);
@@ -178,7 +195,9 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
                         return Err(err);
                     }
                 }
-            }
+            } else {
+                None
+            };
 
             let mut wr = wr;
             let wakl_opts = WalkDirOptions {
@@ -195,7 +214,13 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
 
             let mut need_fallback = false;
             let mut last_err = None;
-            if let Some(disk) = opdisk {
+            #[cfg(test)]
+            if let Some(err) = test_primary_error {
+                last_err = Some(err);
+                need_fallback = true;
+            }
+
+            if !need_fallback && let Some(disk) = opdisk {
                 let primary_walk_started = std::time::Instant::now();
                 match disk.walk_dir(wakl_opts, &mut wr).await {
                     Ok(_res) => {
@@ -238,7 +263,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
                         need_fallback = true;
                     }
                 }
-            } else {
+            } else if !need_fallback {
                 last_err = Some(DiskError::DiskNotFound);
                 need_fallback = true;
             }
@@ -249,8 +274,37 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
             }
 
             while need_fallback {
+                #[cfg(test)]
+                if let Some(behavior) = take_fallback_candidate(&test_fallbacks_clone).await {
+                    match behavior {
+                        TestReaderBehavior::Eof => {
+                            need_fallback = false;
+                            last_err = None;
+                            continue;
+                        }
+                        TestReaderBehavior::Entries(entries) => {
+                            let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr);
+                            out.write(&entries).await.expect("test fallback entries should be written");
+                            out.close().await.expect("test fallback entries should close");
+                            need_fallback = false;
+                            last_err = None;
+                            continue;
+                        }
+                        TestReaderBehavior::ProducerError(err) | TestReaderBehavior::PrimaryErrorThenFallback(err) => {
+                            last_err = Some(err);
+                            continue;
+                        }
+                        TestReaderBehavior::Stall
+                        | TestReaderBehavior::IgnoreCancel
+                        | TestReaderBehavior::PartialThenTimeout(_) => {
+                            last_err = Some(DiskError::Timeout);
+                            continue;
+                        }
+                    }
+                }
+
                 let mut disk_op = None;
-                while let Some(disk) = fds_clone.pop_front() {
+                while let Some(disk) = take_fallback_candidate(&fds_clone).await {
                     if disk.is_online().await {
                         disk_op = Some(disk);
                         break;
@@ -753,6 +807,88 @@ mod tests {
         assert!(!is_missing_path_error(&DiskError::FileAccessDenied));
     }
 
+    #[tokio::test]
+    async fn fallback_candidates_are_claimed_once_across_producers() {
+        let mut queue = VecDeque::new();
+        queue.push_back(1usize);
+        let candidates = Arc::new(TokioMutex::new(queue));
+
+        let first = take_fallback_candidate(&candidates).await;
+        let second = take_fallback_candidate(&candidates).await;
+
+        assert_eq!(first, Some(1));
+        assert_eq!(second, None);
+    }
+
+    fn fallback_test_entry() -> MetaCacheEntry {
+        MetaCacheEntry {
+            name: "bucket/object".to_string(),
+            metadata: vec![1, 2, 3],
+            cached: None,
+            reusable: false,
+        }
+    }
+
+    #[tokio::test]
+    async fn list_path_raw_does_not_reuse_one_fallback_for_multiple_failed_producers() {
+        let err = list_path_raw(
+            CancellationToken::new(),
+            ListPathRawOptions {
+                disks: vec![None, None],
+                min_disks: 2,
+                test_reader_behaviors: vec![
+                    TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound),
+                    TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound),
+                ],
+                test_fallback_reader_behaviors: vec![TestReaderBehavior::Entries(vec![fallback_test_entry()])],
+                ..Default::default()
+            },
+        )
+        .await
+        .expect_err("one fallback must not be counted as two failed primary producers");
+
+        assert_eq!(err, DiskError::DiskNotFound);
+    }
+
+    #[tokio::test]
+    async fn list_path_raw_uses_distinct_fallbacks_to_restore_quorum() {
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let seen_clone = seen.clone();
+
+        list_path_raw(
+            CancellationToken::new(),
+            ListPathRawOptions {
+                disks: vec![None, None],
+                min_disks: 2,
+                test_reader_behaviors: vec![
+                    TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound),
+                    TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound),
+                ],
+                test_fallback_reader_behaviors: vec![
+                    TestReaderBehavior::Entries(vec![fallback_test_entry()]),
+                    TestReaderBehavior::Entries(vec![fallback_test_entry()]),
+                ],
+                partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
+                    let seen = seen_clone.clone();
+                    Box::pin(async move {
+                        let matches = entries
+                            .0
+                            .iter()
+                            .flatten()
+                            .filter(|entry| entry.name == "bucket/object")
+                            .count();
+                        seen.lock().expect("seen mutex poisoned").push(matches);
+                    })
+                })),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("distinct fallback producers should restore listing quorum");
+
+        assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &[2]);
+    }
+
     #[tokio::test]
     async fn list_path_raw_returns_timeout_when_reader_stalls_before_completion() {
         let err = list_path_raw(

crates/ecstore/src/set_disk/read.rs

@@ -358,6 +358,14 @@ impl MetadataQuorumAccumulator {
         if self.requested_version_id.is_empty() {
             return None;
         }
+        if self.conflicting_metadata
+            || self.delete_marker_seen
+            || self.not_found_responses > 0
+            || self.version_not_found_responses > 0
+            || self.hard_errors > 0
+        {
+            return None;
+        }
         if self.matching_version_votes >= self.read_quorum_for_version() {
             return Some(MetadataEarlyStopDecision {
                 reason: GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM,
@@ -432,6 +440,7 @@ fn metadata_early_stop_candidate_matches(left: &FileInfo, right: &FileInfo) -> b
         && left.parts == right.parts
         && left.checksum == right.checksum
         && left.versioned == right.versioned
+        && left.data_dir == right.data_dir
         && left.erasure.algorithm == right.erasure.algorithm
         && left.erasure.data_blocks == right.erasure.data_blocks
         && left.erasure.parity_blocks == right.erasure.parity_blocks
@@ -728,7 +737,7 @@ async fn submit_read_repair_heal_with_submitter(
         bucket.to_string(),
         Some(object.to_string()),
         false,
-        Some(HealChannelPriority::Normal),
+        Some(HealChannelPriority::Low),
         Some(pool_index),
         Some(set_index),
     );
@@ -2468,10 +2477,13 @@ mod metadata_cache_tests {
     use super::*;
     use rustfs_common::heal_channel::HealAdmissionDropReason;
     use serial_test::serial;
+    use std::sync::Mutex;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
     static SLOW_READ_REPAIR_SUBMITTER_CALLS: AtomicUsize = AtomicUsize::new(0);
     static DROPPED_READ_REPAIR_SUBMITTER_CALLS: AtomicUsize = AtomicUsize::new(0);
+    static CAPTURED_READ_REPAIR_PRIORITY: Mutex<Option<HealChannelPriority>> = Mutex::new(None);
+    static CAPTURED_READ_REPAIR_CALLS: AtomicUsize = AtomicUsize::new(0);
 
     fn slow_read_repair_submitter(_request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
         SLOW_READ_REPAIR_SUBMITTER_CALLS.fetch_add(1, Ordering::Relaxed);
@@ -2488,6 +2500,15 @@ mod metadata_cache_tests {
         })
     }
 
+    fn capture_read_repair_submitter(request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
+        CAPTURED_READ_REPAIR_CALLS.fetch_add(1, Ordering::Relaxed);
+        *CAPTURED_READ_REPAIR_PRIORITY.lock().expect("capture mutex poisoned") = Some(request.priority);
+        Box::pin(async {
+            tokio::time::sleep(Duration::from_millis(1)).await;
+            ReadRepairAdmissionOutcome::Response(HealAdmissionResult::Accepted)
+        })
+    }
+
     async fn new_metadata_cache_test_set() -> Arc<SetDisks> {
         SetDisks::new(
             "metadata-cache-test".to_string(),
@@ -2671,6 +2692,41 @@ mod metadata_cache_tests {
         release_read_repair_heal_reservation(&released_key).await;
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn submit_read_repair_heal_uses_low_priority() {
+        CAPTURED_READ_REPAIR_CALLS.store(0, Ordering::Relaxed);
+        *CAPTURED_READ_REPAIR_PRIORITY.lock().expect("capture mutex poisoned") = None;
+        let bucket = format!("bucket-{}", Uuid::new_v4());
+
+        submit_read_repair_heal_with_submitter(
+            ReadRepairHealSubmission {
+                bucket: &bucket,
+                object: "object",
+                version_id: None,
+                pool_index: 0,
+                set_index: 0,
+                part_number: Some(1),
+                reason: "missing_shards",
+            },
+            capture_read_repair_submitter,
+        )
+        .await;
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while CAPTURED_READ_REPAIR_CALLS.load(Ordering::Relaxed) == 0 {
+                tokio::time::sleep(Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("background read-repair submitter should be called");
+
+        assert_eq!(
+            *CAPTURED_READ_REPAIR_PRIORITY.lock().expect("capture mutex poisoned"),
+            Some(HealChannelPriority::Low)
+        );
+    }
+
     #[test]
     fn resolved_read_repair_version_prefers_selected_fileinfo_version() {
         let mut fi = valid_test_fileinfo("object");
@@ -3165,6 +3221,21 @@ mod tests {
         assert_eq!(accumulator.final_miss_reason(), GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA);
     }
 
+    #[test]
+    fn metadata_quorum_accumulator_falls_back_on_split_data_dir() {
+        let mut accumulator = metadata_early_stop_accumulator();
+        let mut first = metadata_early_stop_candidate("object", 1);
+        let mut second = metadata_early_stop_candidate("object", 2);
+        first.data_dir = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000001").expect("static uuid should parse"));
+        second.data_dir = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000002").expect("static uuid should parse"));
+
+        accumulator.observe_file_info(&first);
+        accumulator.observe_file_info(&second);
+
+        assert!(accumulator.early_stop_decision().is_none());
+        assert_eq!(accumulator.final_miss_reason(), GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA);
+    }
+
     #[test]
     fn metadata_quorum_accumulator_falls_back_on_explicit_version_quorum() {
         let mut accumulator = MetadataQuorumAccumulator::new(4, 2, false);
@@ -3342,28 +3413,40 @@ mod tests {
     }
 
     #[test]
-    fn version_early_stop_tracks_matching_votes_independently_of_candidate() {
+    fn version_early_stop_falls_back_on_conflicting_metadata() {
         let requested_vid = Uuid::new_v4();
         let mut accumulator = version_early_stop_accumulator(&requested_vid.to_string());
 
         // Two valid responses with matching version_id but different erasure.index
-        // (so they conflict on the candidate path but still count for version votes)
+        // (so they conflict on the candidate path and must keep the fanout open)
         let mut fi1 = version_early_stop_candidate("object", 1, requested_vid);
         fi1.size = 100;
         let mut fi2 = version_early_stop_candidate("object", 2, requested_vid);
-        fi2.size = 200; // different size → conflicting metadata on candidate path
+        fi2.size = 200;
 
         accumulator.observe_file_info(&fi1);
         accumulator.observe_file_info(&fi2);
 
-        // Candidate path sees conflict, but version path sees quorum
         assert!(accumulator.early_stop_decision().is_none());
-        assert_eq!(
-            accumulator.version_early_stop_decision(),
-            Some(MetadataEarlyStopDecision {
-                reason: GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM
-            })
-        );
+        assert!(accumulator.version_early_stop_decision().is_none());
+        assert_eq!(accumulator.final_miss_reason(), GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA);
+    }
+
+    #[test]
+    fn version_early_stop_falls_back_on_split_data_dir() {
+        let requested_vid = Uuid::new_v4();
+        let mut accumulator = version_early_stop_accumulator(&requested_vid.to_string());
+        let mut first = version_early_stop_candidate("object", 1, requested_vid);
+        let mut second = version_early_stop_candidate("object", 2, requested_vid);
+        first.data_dir = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000001").expect("static uuid should parse"));
+        second.data_dir = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000002").expect("static uuid should parse"));
+
+        accumulator.observe_file_info(&first);
+        accumulator.observe_file_info(&second);
+
+        assert!(accumulator.early_stop_decision().is_none());
+        assert!(accumulator.version_early_stop_decision().is_none());
+        assert_eq!(accumulator.final_miss_reason(), GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA);
     }
 
     fn codec_streaming_test_object_info(fi: &FileInfo) -> ObjectInfo {