fix(ecstore): keep commit-proof selection for system-bucket reads

Pool metadata's PREPARE/COMMIT recovery and identity epochs treat an
ordinary read of pool.bin as proof that the record committed. Under the
read-quorum rule, a COMMIT that reached only half of a 2+2 set was served
as committed, and the pool_meta_native_physical_* integration tests
failed.

Route reads through select_fileinfo_for_read: S3 buckets keep MinIO's
read-quorum rule, while is_meta_bucketname reads use the previous
algorithm unchanged as select_committed_fileinfo, so a split stays
unreadable until native recovery resolves it.

crates/ecstore/src/set_disk/metadata.rs

@@ -27,6 +27,13 @@ use rustfs_filemeta::metadata_keys;
 use rustfs_utils::http;
 use sha2::Digest;
 
+#[derive(Clone, Copy)]
+struct FileInfoIdentityGroup {
+    hash: [u8; 32],
+    count: usize,
+    mod_time: Option<OffsetDateTime>,
+}
+
 impl SetDisks {
     /// Read one key's complete version stack without enumerating its parent.
     pub(crate) async fn read_object_versions_for_lifecycle(
@@ -583,7 +590,135 @@ impl SetDisks {
         (new_disk, mod_time, None)
     }
 
-    /// Select the version a read serves with MinIO's rule (`getObjectFileInfo`
+    fn usable_fileinfo_count(parts_metadata: &[FileInfo], errs: &[Option<DiskError>]) -> (usize, bool) {
+        let mut has_read_error = false;
+        let mut usable_metadata = 0;
+        for (meta, err) in parts_metadata.iter().zip(errs.iter()) {
+            if err.is_some() {
+                has_read_error = true;
+                continue;
+            }
+
+            if file_info_is_valid_for_metadata(meta) {
+                usable_metadata += 1;
+            }
+        }
+
+        (usable_metadata, has_read_error)
+    }
+
+    pub(super) fn latest_fileinfo_selection_quorum(
+        version_id: &str,
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        read_quorum: usize,
+        write_quorum: usize,
+    ) -> usize {
+        if !version_id.is_empty() || write_quorum <= read_quorum {
+            return read_quorum;
+        }
+
+        let (usable_metadata, has_read_error) = Self::usable_fileinfo_count(parts_metadata, errs);
+
+        if usable_metadata < write_quorum {
+            return read_quorum;
+        }
+
+        if !has_read_error {
+            return write_quorum;
+        }
+
+        if let Some((first, _)) = parts_metadata
+            .iter()
+            .zip(errs.iter())
+            .find(|(meta, err)| err.is_none() && file_info_is_valid_for_metadata(meta))
+        {
+            let same_count = parts_metadata
+                .iter()
+                .zip(errs.iter())
+                .filter(|(meta, err)| {
+                    err.is_none()
+                        && file_info_is_valid_for_metadata(meta)
+                        && super::core::metadata_quorum::metadata_early_stop_candidate_matches(first, meta)
+                })
+                .count();
+            if same_count >= write_quorum {
+                return write_quorum;
+            }
+        }
+
+        let mut identity_counts = HashMap::with_capacity(usable_metadata);
+        for (meta, err) in parts_metadata.iter().zip(errs.iter()) {
+            if err.is_some() || !file_info_is_valid_for_metadata(meta) {
+                continue;
+            }
+
+            let key = Self::file_info_quorum_hash(meta);
+
+            let count = identity_counts.entry(key).or_insert(0);
+            *count += 1;
+            if *count >= write_quorum {
+                return write_quorum;
+            }
+        }
+
+        read_quorum
+    }
+
+    /// The commit-proof rule kept for internal metadata in the system bucket.
+    /// Pool metadata's PREPARE/COMMIT recovery and identity epochs treat an
+    /// ordinary read as proof that a record committed: when enough copies
+    /// answer to prove write quorum, the latest version must hold it, and a
+    /// degraded read never serves a newer version below write quorum. A split
+    /// write therefore stays unreadable until native recovery resolves it.
+    fn select_committed_fileinfo(
+        disks: &[Option<DiskStore>],
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        version_id: &str,
+        read_quorum: usize,
+        write_quorum: usize,
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo)> {
+        let selection_quorum =
+            Self::latest_fileinfo_selection_quorum(version_id, parts_metadata, errs, read_quorum, write_quorum);
+        let (usable_metadata, has_read_error) = Self::usable_fileinfo_count(parts_metadata, errs);
+
+        if version_id.is_empty()
+            && write_quorum > read_quorum
+            && has_read_error
+            && usable_metadata >= write_quorum
+            && selection_quorum == read_quorum
+        {
+            return Self::pick_degraded_latest_fileinfo(disks, parts_metadata, errs, read_quorum, write_quorum);
+        }
+
+        let (online_disks, mod_time, etag) = Self::list_online_disks(disks, parts_metadata, errs, selection_quorum);
+        let fi = Self::pick_valid_fileinfo(parts_metadata, mod_time, etag, selection_quorum)?;
+
+        Ok((online_disks, fi))
+    }
+
+    /// Select the version a read of `bucket` serves: S3 buckets use MinIO's
+    /// read-quorum rule (`select_valid_fileinfo`); the system bucket keeps the
+    /// commit-proof rule (`select_committed_fileinfo`).
+    #[allow(clippy::too_many_arguments)]
+    pub(super) fn select_fileinfo_for_read(
+        bucket: &str,
+        disks: &[Option<DiskStore>],
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        version_id: &str,
+        read_quorum: usize,
+        write_quorum: usize,
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo)> {
+        if crate::bucket::utils::is_meta_bucketname(bucket) {
+            Self::select_committed_fileinfo(disks, parts_metadata, errs, version_id, read_quorum, write_quorum)
+        } else {
+            Self::select_valid_fileinfo(disks, parts_metadata, errs, read_quorum)
+        }
+    }
+
+    /// Select the version an S3 read serves with MinIO's rule (`getObjectFileInfo`
     /// -> `listOnlineDisks` -> `pickValidFileInfo`): the modification time
     /// shared by the most copies wins when it reaches `read_quorum`, a tie
     /// goes to the newer time, and one content identity at that time must
@@ -853,6 +988,100 @@ impl SetDisks {
         }
     }
 
+    fn latest_fileinfo_identity_groups(parts_metadata: &[FileInfo], errs: &[Option<DiskError>]) -> Vec<FileInfoIdentityGroup> {
+        let mut groups: Vec<FileInfoIdentityGroup> = Vec::with_capacity(parts_metadata.len());
+        for (meta, err) in parts_metadata.iter().zip(errs.iter()) {
+            if err.is_some() || !file_info_is_valid_for_metadata(meta) {
+                continue;
+            }
+
+            let hash = Self::file_info_quorum_hash(meta);
+            if let Some(group) = groups.iter_mut().find(|group| group.hash == hash) {
+                group.count += 1;
+                continue;
+            }
+
+            groups.push(FileInfoIdentityGroup {
+                hash,
+                count: 1,
+                mod_time: meta.mod_time,
+            });
+        }
+
+        groups
+    }
+
+    fn pick_fileinfo_identity(
+        disks: &[Option<DiskStore>],
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        hash: [u8; 32],
+        quorum: usize,
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo)> {
+        let mut online_disks = vec![None; disks.len()];
+        let mut selected = None;
+        let mut count = 0;
+
+        for (i, ((meta, err), disk)) in parts_metadata.iter().zip(errs.iter()).zip(disks.iter()).enumerate() {
+            if err.is_some() || !file_info_is_valid_for_metadata(meta) || Self::file_info_quorum_hash(meta) != hash {
+                continue;
+            }
+
+            count += 1;
+            online_disks[i].clone_from(disk);
+            if selected.is_none() {
+                selected = Some(meta.clone());
+            }
+        }
+
+        if count < quorum {
+            return Err(DiskError::ErasureReadQuorum);
+        }
+
+        selected
+            .map(|mut fi| {
+                fi.is_latest = fi.successor_mod_time.is_none();
+                (online_disks, fi)
+            })
+            .ok_or(DiskError::ErasureReadQuorum)
+    }
+
+    fn pick_degraded_latest_fileinfo(
+        disks: &[Option<DiskStore>],
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        read_quorum: usize,
+        write_quorum: usize,
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo)> {
+        let mut groups = Self::latest_fileinfo_identity_groups(parts_metadata, errs);
+        if groups.is_empty() {
+            return Err(DiskError::ErasureReadQuorum);
+        }
+
+        groups.sort_by(|left, right| right.mod_time.cmp(&left.mod_time).then_with(|| right.count.cmp(&left.count)));
+        let latest_mod_time = groups[0].mod_time;
+
+        let mut older_start = 0;
+        while older_start < groups.len() && groups[older_start].mod_time == latest_mod_time {
+            if groups[older_start].count >= write_quorum {
+                return Self::pick_fileinfo_identity(disks, parts_metadata, errs, groups[older_start].hash, write_quorum);
+            }
+            older_start += 1;
+        }
+
+        if older_start > 1 {
+            return Err(DiskError::ErasureReadQuorum);
+        }
+
+        for group in groups.iter().skip(older_start) {
+            if group.count >= read_quorum {
+                return Self::pick_fileinfo_identity(disks, parts_metadata, errs, group.hash, read_quorum);
+            }
+        }
+
+        Err(DiskError::ErasureReadQuorum)
+    }
+
     pub(super) fn find_file_info_in_quorum(
         metas: &[FileInfo],
         mod_time: &Option<OffsetDateTime>,

crates/ecstore/src/set_disk/mod.rs

@@ -12153,6 +12153,47 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_system_bucket_reads_keep_the_commit_proof_rule() {
+        let old_mod_time = OffsetDateTime::now_utc();
+        let new_mod_time = old_mod_time + time::Duration::seconds(1);
+        let old_data_dir = Uuid::new_v4();
+        let new_data_dir = Uuid::new_v4();
+        let old = |index| quorum_test_fileinfo(old_mod_time, old_data_dir, "part-etag-old", index);
+        let new = |index| quorum_test_fileinfo(new_mod_time, new_data_dir, "part-etag-new", index);
+
+        // A PREPARE/COMMIT split of pool metadata must stay unreadable so that
+        // native recovery, not an ordinary read, decides whether it committed.
+        for (metas, errs) in [
+            (vec![old(1), old(2), new(3), new(4)], vec![None, None, None, None]),
+            (
+                vec![new(1), new(2), old(3), FileInfo::default()],
+                vec![None, None, None, Some(DiskError::DiskNotFound)],
+            ),
+        ] {
+            let disks = vec![None; metas.len()];
+            let (_, served) = SetDisks::select_fileinfo_for_read("bucket", &disks, &metas, &errs, "", 2, 3)
+                .expect("an S3 read serves the newer side at read quorum");
+            assert_eq!(served.data_dir, Some(new_data_dir));
+
+            let internal = SetDisks::select_fileinfo_for_read(crate::disk::RUSTFS_META_BUCKET, &disks, &metas, &errs, "", 2, 3);
+            assert!(matches!(internal, Err(DiskError::ErasureReadQuorum)));
+        }
+
+        let committed = vec![new(1), new(2), new(3), old(4)];
+        let (_, internal) = SetDisks::select_fileinfo_for_read(
+            crate::disk::RUSTFS_META_BUCKET,
+            &vec![None; committed.len()],
+            &committed,
+            &[None, None, None, None],
+            "",
+            2,
+            3,
+        )
+        .expect("a write-quorum version stays readable in the system bucket");
+        assert_eq!(internal.data_dir, Some(new_data_dir));
+    }
+
     #[test]
     fn test_latest_fileinfo_selection_picks_version_stack_shape_in_disk_order() {
         let mod_time = OffsetDateTime::now_utc();

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

@@ -9319,8 +9319,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             }
         };
 
-        let (online_disks, mut fi) = Self::select_valid_fileinfo(&disks, &metas, &errs, read_quorum)
-            .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
+        let src_version_id = src_opts.version_id.as_deref().unwrap_or_default();
+        let (online_disks, mut fi) =
+            Self::select_fileinfo_for_read(src_bucket, &disks, &metas, &errs, src_version_id, read_quorum, write_quorum)
+                .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
 
         if fi.deleted {
             if src_opts.version_id.is_none() {
@@ -11145,8 +11147,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             }
         };
 
-        let read_quorum = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) {
-            Ok((read_quorum, _)) => read_quorum,
+        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) {
+            Ok((read_quorum, write_quorum)) => (read_quorum, write_quorum),
             Err(mut err) => {
                 // A replication terminal CAS is a validation/write-back
                 // attempt, never a heal operation. In particular,
@@ -11176,9 +11178,13 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
 
         let read_quorum =
             usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
+        let write_quorum = usize::try_from(write_quorum)
+            .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
 
-        let (online_disks, mut fi) = Self::select_valid_fileinfo(&disks, &metas, &errs, read_quorum)
-            .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
+        let version_id = opts.version_id.as_deref().unwrap_or_default();
+        let (online_disks, mut fi) =
+            Self::select_fileinfo_for_read(bucket, &disks, &metas, &errs, version_id, read_quorum, write_quorum)
+                .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
 
         if fi.deleted {
             return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object]));

crates/ecstore/src/set_disk/read.rs

@@ -926,7 +926,7 @@ impl SetDisks {
                 None
             }
         };
-        let (read_quorum, _) = match Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)
+        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)
             .map_err(|err| dangling_not_found(&err).unwrap_or(err))
             .map_err(|err| to_object_err(err.into(), vec![bucket, object, &vid]))
         {
@@ -944,6 +944,8 @@ impl SetDisks {
         };
         let read_quorum =
             usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
+        let write_quorum = usize::try_from(write_quorum)
+            .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
         if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) {
             if let Some(not_found) = dangling_not_found(&err) {
                 rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, "not_found");
@@ -964,14 +966,19 @@ impl SetDisks {
             return Err(to_object_err(err.into(), vec![bucket, object, &vid]));
         }
 
-        let (op_online_disks, mut fi) = match Self::select_valid_fileinfo(&disks, &parts_metadata, &errs, read_quorum) {
-            Ok(selection) => selection,
-            Err(err) => {
-                let err = StorageError::from(err);
-                rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, metadata_quorum_outcome(&err));
-                return Err(err);
-            }
-        };
+        let (op_online_disks, mut fi) =
+            match Self::select_fileinfo_for_read(bucket, &disks, &parts_metadata, &errs, vid.as_str(), read_quorum, write_quorum)
+            {
+                Ok(selection) => selection,
+                Err(err) => {
+                    let err = StorageError::from(err);
+                    rustfs_io_metrics::record_get_object_metadata_quorum_result(
+                        metadata_metrics_path,
+                        metadata_quorum_outcome(&err),
+                    );
+                    return Err(err);
+                }
+            };
         let include_part_checksums =
             opts.include_part_checksums || opts.part_number.is_some() || opts.data_movement || opts.raw_data_movement_read;
         if include_part_checksums {
@@ -2448,15 +2455,18 @@ impl SetDisks {
         let (mut parts_metadata, errs, diagnostics) = metadata_read.into_legacy();
         diagnostics.record(metrics_path);
 
-        let (read_quorum, _) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, expected.parity_blocks)
+        let (read_quorum, write_quorum) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, expected.parity_blocks)
             .map_err(|err| to_object_err(err.into(), vec![bucket, object]))?;
         let read_quorum =
             usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
+        let write_quorum = usize::try_from(write_quorum)
+            .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
         if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) {
             return Err(to_object_err(err.into(), vec![bucket, object]));
         }
 
-        let (mut online_disks, full_fi) = SetDisks::select_valid_fileinfo(fallback_disks, &parts_metadata, &errs, read_quorum)?;
+        let (mut online_disks, full_fi) =
+            SetDisks::select_fileinfo_for_read(bucket, fallback_disks, &parts_metadata, &errs, "", read_quorum, write_quorum)?;
         if !late_metadata_read_identity_matches(expected, &full_fi) {
             return Err(to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]));
         }