chore: merge main into scanner feedback

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -5752,6 +5752,19 @@ impl SetDisks {
             return Ok(());
         }
 
+        // A metadata update that misses write quorum is not rolled back, so
+        // refuse it before writing when the target disks cannot reach quorum
+        // even if every write succeeds; otherwise the change a caller reports
+        // as failed would be served and spread by heal.
+        let write_quorum = fi.write_quorum(self.default_write_quorum());
+        let reachable = disks
+            .iter()
+            .map(|disk| disk.is_none().then_some(DiskError::DiskNotFound))
+            .collect::<Vec<_>>();
+        if let Some(err) = reduce_write_quorum_errs(&reachable, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+            return Err(err);
+        }
+
         self.invalidate_get_object_metadata_cache(bucket, object).await;
 
         let mut futures = Vec::with_capacity(disks.len());
@@ -5781,7 +5794,7 @@ impl SetDisks {
             }
         }
 
-        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, fi.write_quorum(self.default_write_quorum())) {
+        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
             return Err(err);
         }
 
@@ -9440,7 +9453,7 @@ mod tests {
             "lifecycle-inline-data-get-fanout-bucket",
             "lifecycle-inline-data-get-object",
             GET_OBJECT_PATH_LEGACY_DUPLEX,
-            None,
+            false,
         );
     }
 
@@ -9451,7 +9464,7 @@ mod tests {
             "lifecycle-bounded-inline-data-get-fanout-bucket",
             "lifecycle-bounded-inline-data-get-object",
             GET_OBJECT_PATH_LEGACY_DUPLEX,
-            Some("true"),
+            true,
         );
     }
 
@@ -9462,7 +9475,7 @@ mod tests {
             RUSTFS_META_BUCKET,
             "buckets/.usage-cache/lifecycle-inline-data-get-object",
             GET_OBJECT_PATH_INTERNAL_META,
-            None,
+            false,
         );
     }
 
@@ -9563,7 +9576,7 @@ mod tests {
         bucket: &'static str,
         object_prefix: &str,
         expected_path: &'static str,
-        bounded_fanout_env: Option<&'static str>,
+        bounded_fanout: bool,
     ) {
         const DISKS: usize = 4;
         let object = object_with_initial_data_shards(bucket, object_prefix, 2, 3);
@@ -9583,16 +9596,26 @@ mod tests {
                     [
                         ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
                         ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
-                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", bounded_fanout_env),
+                        (
+                            "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT",
+                            Some(if bounded_fanout { "true" } else { "false" }),
+                        ),
                     ],
                     async {
-                        let barrier_disk = if bounded_fanout_env.is_some() {
-                            bounded_spare_disk_index(bucket, &object, 2, 2)
-                        } else {
-                            3
-                        };
+                        let barrier_disk = bounded_spare_disk_index(bucket, &object, 2, 2);
                         let barrier =
                             rename_fanout_barrier::arm(&object, barrier_disk, rename_fanout_barrier::PHASE_READ_VERSION);
+                        // Keep one initial bounded read pending so its hedge is scheduled;
+                        // eager fanout needs all quorum reads parked until the spare starts.
+                        let quorum_disks = if bounded_fanout {
+                            vec![bounded_metadata_fanout_order(bucket, &object, DISKS, 2)[0]]
+                        } else {
+                            (0..DISKS).filter(|&index| index != barrier_disk).collect()
+                        };
+                        let quorum_barriers = quorum_disks
+                            .into_iter()
+                            .map(|index| rename_fanout_barrier::arm(&object, index, rename_fanout_barrier::PHASE_READ_VERSION))
+                            .collect::<Vec<_>>();
                         let disks_for_read = disks.clone();
                         let object_for_read = object.clone();
                         let mut read = tokio::spawn(async move {
@@ -9614,6 +9637,9 @@ mod tests {
                         tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
                             .await
                             .expect("metadata fanout should pause the spare metadata task");
+                        for quorum_barrier in &quorum_barriers {
+                            quorum_barrier.release();
+                        }
                         let (parts_metadata, errs, diagnostics) = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read)
                             .await
                             .expect("verified inline quorum should abort the paused eager fanout")

crates/ecstore/src/set_disk/core/metadata_quorum.rs

@@ -597,10 +597,8 @@ mod tests {
                         .iter()
                         .map(|slot| slot.file_info().expect("every disk must have arrived").clone())
                         .collect::<Vec<_>>();
-                    let (_, selected, selection_quorum) =
-                        SetDisks::select_valid_fileinfo(&vec![None; 4], &completed_metadata, &vec![None; 4], "", 2, 3)
-                            .expect("three matching B disks must select the current identity");
-                    assert_eq!(selection_quorum, 3);
+                    let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; 4], &completed_metadata, &vec![None; 4], 2)
+                        .expect("three matching B disks must select the current identity");
                     assert_eq!(selected.version_id, expected_current.version_id);
                     assert_eq!(selected.mod_time, expected_current.mod_time);
                     assert_eq!(selected.data_dir, expected_current.data_dir);

crates/ecstore/src/set_disk/metadata.rs

@@ -665,14 +665,20 @@ impl SetDisks {
         read_quorum
     }
 
-    pub(super) fn select_valid_fileinfo(
+    /// 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, 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);
@@ -683,14 +689,71 @@ impl SetDisks {
             && usable_metadata >= write_quorum
             && selection_quorum == read_quorum
         {
-            let (online_disks, fi) = Self::pick_degraded_latest_fileinfo(disks, parts_metadata, errs, read_quorum, write_quorum)?;
-            return Ok((online_disks, fi, 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, 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
+    /// itself reach `read_quorum`. Heal picks the latest version the same
+    /// way, so a read returns the version heal converges the set to; a write
+    /// that failed after reaching exactly `read_quorum` copies is readable
+    /// instead of failing until heal runs, as listing already shows it.
+    ///
+    /// `read_quorum` follows the parity most copies agree on, which delete
+    /// markers and empty versions can lower below the selected version's own
+    /// data blocks; like MinIO's `getLatestFileInfo`, a version with local
+    /// shards also needs that many copies, or HEAD would report a version
+    /// GET cannot decode.
+    pub(super) fn select_valid_fileinfo(
+        disks: &[Option<DiskStore>],
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        read_quorum: usize,
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo)> {
+        let (online_disks, mod_time, etag) = Self::list_online_disks(disks, parts_metadata, errs, read_quorum);
+        let (fi, copies) = Self::find_file_info_in_quorum_counted(parts_metadata, &mod_time, &etag, read_quorum)?;
+        if !fi.deleted && !fi.is_remote() && fi.size > 0 && copies < fi.erasure.data_blocks {
+            // The shortcut counts byte-identical copies only; count every copy
+            // of the content identity before refusing the version.
+            let identity = Self::file_info_quorum_hash(&fi);
+            let copies = parts_metadata
+                .iter()
+                .filter(|meta| file_info_is_valid_for_metadata(meta) && Self::file_info_quorum_hash(meta) == identity)
+                .count();
+            if copies < fi.erasure.data_blocks {
+                return Err(DiskError::ErasureReadQuorum);
+            }
+        }
+
+        Ok((online_disks, fi))
     }
 
     pub(super) fn pick_valid_fileinfo(
@@ -1025,6 +1088,23 @@ impl SetDisks {
         etag: &Option<String>,
         quorum: usize,
     ) -> disk::error::Result<FileInfo> {
+        Self::find_file_info_in_quorum_counted(metas, mod_time, etag, quorum).map(|(fi, _)| fi)
+    }
+
+    /// `find_file_info_in_quorum`, also returning how many copies carry the
+    /// selected content identity.
+    ///
+    /// Two identities can share the top count only in a data = parity set
+    /// split exactly in half at one modification time, such as a metadata
+    /// update that failed after reaching half the drives. The identity held
+    /// by the lowest-index disk wins, so every read and heal picks the same
+    /// side and a retried metadata write can repair the split.
+    fn find_file_info_in_quorum_counted(
+        metas: &[FileInfo],
+        mod_time: &Option<OffsetDateTime>,
+        etag: &Option<String>,
+        quorum: usize,
+    ) -> disk::error::Result<(FileInfo, usize)> {
         if quorum < 1 {
             warn!("find_file_info_in_quorum: quorum < 1");
             return Err(DiskError::ErasureReadQuorum);
@@ -1053,7 +1133,7 @@ impl SetDisks {
             if same_count >= quorum && same_count > total_valid - same_count {
                 let mut fi = first.clone();
                 fi.is_latest = fi.successor_mod_time.is_none();
-                return Ok(fi);
+                return Ok((fi, same_count));
             }
         }
 
@@ -1106,15 +1186,12 @@ impl SetDisks {
             *count_map.entry(hash).or_insert(0) += 1;
         }
 
-        let mut max_val = None;
-        let mut max_count = 0;
-
-        for (&val, &count) in &count_map {
-            if count > max_count {
-                max_val = Some(val);
-                max_count = count;
-            }
-        }
+        let max_count = count_map.values().copied().max().unwrap_or(0);
+        let max_val = meta_hashes
+            .iter()
+            .flatten()
+            .find(|hash| count_map.get(*hash) == Some(&max_count))
+            .copied();
 
         if max_count < quorum {
             warn!(
@@ -1155,17 +1232,24 @@ impl SetDisks {
         if found {
             let mut fi = found_fi.expect("operation should succeed");
 
-            for (val, &count) in &valid_obj_map {
-                if count >= quorum {
-                    fi.successor_mod_time = val.successor_mod_time;
-                    fi.num_versions = val.num_versions;
-                    fi.is_latest = val.successor_mod_time.is_none();
-
-                    break;
-                }
+            // Scan in disk order so that two version-stack shapes with equal
+            // counts resolve the same way on every read.
+            let quorum_props = metas
+                .iter()
+                .zip(meta_hashes.iter())
+                .filter(|(meta, hash)| hash.is_some() && **hash == max_val && file_info_is_valid_for_metadata(meta))
+                .map(|(meta, _)| ObjProps {
+                    successor_mod_time: meta.successor_mod_time,
+                    num_versions: meta.num_versions,
+                })
+                .find(|props| valid_obj_map.get(props).is_some_and(|&count| count >= quorum));
+            if let Some(props) = quorum_props {
+                fi.successor_mod_time = props.successor_mod_time;
+                fi.num_versions = props.num_versions;
+                fi.is_latest = props.successor_mod_time.is_none();
             }
 
-            return Ok(fi);
+            return Ok((fi, max_count));
         }
 
         warn!("find_file_info_in_quorum: fileinfo not found");

crates/ecstore/src/set_disk/mod.rs

@@ -1790,7 +1790,7 @@ mod prepared_get_object_metadata_tests {
         assert!(original.is_complete());
         let (original_metadata, original_errors, _) = original.into_legacy();
         assert!(original_errors.iter().all(Option::is_none));
-        let (_, generation_a, _) = SetDisks::select_valid_fileinfo(&disks, &original_metadata, &original_errors, "", 2, 3)
+        let (_, generation_a) = SetDisks::select_valid_fileinfo(&disks, &original_metadata, &original_errors, 2)
             .expect("generation A must have real metadata quorum");
         assert!(!generation_a.inline_data(), "fixture must use non-inline shard files");
         let order = bounded_metadata_fanout_order(bucket, &object, 4, 2);
@@ -1851,8 +1851,8 @@ mod prepared_get_object_metadata_tests {
                 assert!(replacement.is_complete());
                 let (replacement_metadata, replacement_errors, _) = replacement.into_legacy();
                 assert!(replacement_errors.iter().all(Option::is_none));
-                let (online, generation_b, _) =
-                    SetDisks::select_valid_fileinfo(&disks, &replacement_metadata, &replacement_errors, "", 2, 3)
+                let (online, generation_b) =
+                    SetDisks::select_valid_fileinfo(&disks, &replacement_metadata, &replacement_errors, 2)
                         .expect("replacement must have healthy real metadata quorum");
                 assert_eq!(online.iter().flatten().count(), 4);
                 assert_ne!(generation_a.data_dir, generation_b.data_dir);
@@ -6829,7 +6829,7 @@ impl SetDisks {
         let (metadata, errors) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false).await?;
         // Match delete_object_version_with_purge, independently of data parity or online inventory size.
         let quorum = self.set_drive_count / 2 + 1;
-        let (_, marker, _) = Self::select_valid_fileinfo(&disks, &metadata, &errors, version_id, quorum, quorum)
+        let (_, marker) = Self::select_valid_fileinfo(&disks, &metadata, &errors, quorum)
             .map_err(|error| to_object_err(error.into(), vec![bucket, object, version_id]))?;
         if !marker.deleted || marker.tier_free_version() || marker.version_id.unwrap_or_else(Uuid::nil) != requested_version {
             return Err(Error::DataMovementOverwriteErr(
@@ -11969,41 +11969,121 @@ mod tests {
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_quorum_requires_write_quorum_when_full_metadata_is_available() {
+    fn test_latest_fileinfo_selection_breaks_same_time_tie_in_disk_order() {
         let mod_time = OffsetDateTime::now_utc();
         let data_dir = Uuid::new_v4();
-        let metas = vec![
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 1),
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 2),
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-b", 3),
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-b", 4),
-        ];
+        let tagged = |index, tag: &str| {
+            let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+            fi.metadata.insert("x-amz-tagging".to_string(), tag.to_string());
+            fi
+        };
         let errs = vec![None, None, None, None];
 
-        let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3);
+        // A tagging write that failed after reaching half the drives.
+        for (metas, expected) in [
+            (vec![tagged(1, "a=1"), tagged(2, "a=1"), tagged(3, "a=2"), tagged(4, "a=2")], "a=1"),
+            (vec![tagged(1, "a=2"), tagged(2, "a=1"), tagged(3, "a=1"), tagged(4, "a=2")], "a=2"),
+        ] {
+            // Each call hashes into fresh maps; a map-order choice would flip.
+            for _ in 0..32 {
+                let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                    .expect("a same-time half split serves the side on the lowest-index disk");
+                let healed =
+                    SetDisks::find_file_info_in_quorum(&metas, &Some(mod_time), &None, 2).expect("heal resolves the same split");
 
-        assert_eq!(quorum, 3);
+                assert_eq!(selected.metadata.get("x-amz-tagging").map(String::as_str), Some(expected));
+                assert_eq!(healed.metadata.get("x-amz-tagging").map(String::as_str), Some(expected));
+            }
+        }
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_quorum_preserves_read_quorum_for_version_or_degraded_reads() {
-        let mod_time = OffsetDateTime::now_utc();
-        let data_dir = Uuid::new_v4();
-        let metas = vec![
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 1),
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 2),
-            FileInfo::default(),
-            FileInfo::default(),
-        ];
-        let degraded_errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)];
-        let clean_errs = vec![None, None, None, None];
+    fn test_latest_fileinfo_selection_requires_the_versions_own_data_blocks() {
+        let old_mod_time = OffsetDateTime::now_utc();
+        let new_mod_time = old_mod_time + time::Duration::seconds(1);
+        let old = |index| quorum_test_fileinfo(old_mod_time, Uuid::from_u128(1), "part-etag-old", index);
+        let new = |index| {
+            let mut fi = quorum_test_fileinfo(new_mod_time, Uuid::from_u128(2), "part-etag-new", index);
+            fi.erasure.data_blocks = 3;
+            fi.erasure.parity_blocks = 1;
+            fi
+        };
+        let errs = vec![None, None, None, None];
 
-        assert_eq!(SetDisks::latest_fileinfo_selection_quorum("", &metas, &degraded_errs, 2, 3), 2);
-        assert_eq!(SetDisks::latest_fileinfo_selection_quorum("version-id", &metas, &clean_errs, 2, 3), 2);
+        // Read quorum two comes from the 2+2 copies; the newer 3+1 version
+        // wins the time tie but cannot be decoded from two shards.
+        let metas = vec![old(1), old(2), new(3), new(4)];
+        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2);
+        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
+
+        let metas = vec![old(1), new(2), new(3), new(4)];
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("three copies decode a 3+1 version");
+        assert_eq!(selected.mod_time, Some(new_mod_time));
+
+        // Replication-status drift does not change the content identity, so
+        // the drifted copy still counts toward the version's data blocks.
+        let replication_status_key = format!(
+            "{}{}",
+            rustfs_utils::http::RUSTFS_INTERNAL_PREFIX,
+            rustfs_utils::http::SUFFIX_REPLICATION_STATUS
+        );
+        let mut drifted = new(4);
+        drifted
+            .metadata
+            .insert(replication_status_key, "target-a=PENDING;".to_string());
+        let metas = vec![old(1), new(2), new(3), drifted];
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("a replication-status drift still leaves three copies of the version");
+        assert_eq!(selected.mod_time, Some(new_mod_time));
+    }
+
+    #[tokio::test]
+    async fn update_object_meta_below_write_quorum_writes_nothing() {
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+        use crate::storage_api_contracts::object::ObjectIO as _;
+
+        let (_dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks(4, 2).await;
+        let (bucket, object) = ("metadata-write-quorum", "object");
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        set.put_object(
+            bucket,
+            object,
+            &mut crate::object_api::PutObjReader::from_vec(b"body".to_vec()),
+            &ObjectOptions::default(),
+        )
+        .await
+        .expect("object should commit on every drive");
+        let disks = set.get_disks_internal().await;
+        let (metadata, _) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", false, false, false)
+            .await
+            .expect("committed metadata should be readable");
+        let mut update = metadata[0].clone();
+        update.metadata.insert("x-amz-meta-probe".to_string(), "written".to_string());
+
+        // A read that selected one side of a 2+2 split returns two disks.
+        let mut targets = disks.clone();
+        targets[2] = None;
+        targets[3] = None;
+        let err = set
+            .update_object_meta(bucket, object, update, &targets)
+            .await
+            .expect_err("two of four disks cannot reach write quorum");
+
+        assert_eq!(err, DiskError::ErasureWriteQuorum);
+        let (after, _) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", false, false, false)
+            .await
+            .expect("metadata should stay readable");
+        assert!(
+            after.iter().all(|fi| !fi.metadata.contains_key("x-amz-meta-probe")),
+            "a refused update must not reach any disk"
+        );
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_quorum_keeps_read_quorum_for_partial_overwrite_with_read_error() {
+    fn test_latest_fileinfo_selection_keeps_older_majority_over_partial_overwrite() {
         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();
@@ -12016,26 +12096,15 @@ mod tests {
         ];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3);
-        let (online_disks, mod_time, etag) = SetDisks::list_online_disks(&vec![None; metas.len()], &metas, &errs, quorum);
-        let fi = SetDisks::pick_valid_fileinfo(&metas, mod_time, etag, quorum)
-            .expect("old metadata should remain readable with read quorum");
-
-        assert_eq!(quorum, 2);
-        assert_eq!(online_disks.len(), metas.len());
-        assert_eq!(fi.data_dir, Some(old_data_dir));
-        assert_eq!(fi.parts[0].etag, "part-etag-old");
-
-        let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
             .expect("old metadata should remain selectable with read quorum");
-        assert_eq!(selected_quorum, 2);
         assert_eq!(selected.data_dir, Some(old_data_dir));
         assert_eq!(selected.parts[0].etag, "part-etag-old");
         assert!(selected.is_latest);
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_rejects_partial_latest_read_quorum_with_read_error() {
+    fn test_latest_fileinfo_selection_serves_newer_read_quorum_with_read_error() {
         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();
@@ -12048,9 +12117,120 @@ mod tests {
         ];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3);
+        // The new version may hold write quorum through the missing disk; the
+        // old one cannot, and heal converges the set to the new version.
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("the newer version reaches read quorum");
 
-        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
+        assert_eq!(selected.data_dir, Some(new_data_dir));
+        assert_eq!(selected.mod_time, Some(new_mod_time));
+    }
+
+    #[test]
+    fn test_latest_fileinfo_selection_serves_newer_side_of_complete_split() {
+        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);
+        let errs = vec![None, None, None, None];
+
+        for metas in [vec![old(1), old(2), new(3), new(4)], vec![new(1), old(2), new(3), old(4)]] {
+            let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                .expect("a 2+2 split with every disk answering serves the newer version");
+
+            assert_eq!(selected.data_dir, Some(new_data_dir));
+            assert_eq!(selected.parts[0].etag, "part-etag-new");
+            assert!(selected.is_latest);
+        }
+    }
+
+    #[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();
+        let data_dir = Uuid::new_v4();
+        let successor = mod_time + time::Duration::seconds(1);
+        let shape = |index, successor_mod_time: Option<OffsetDateTime>, num_versions| {
+            let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+            fi.successor_mod_time = successor_mod_time;
+            fi.is_latest = successor_mod_time.is_none();
+            fi.num_versions = num_versions;
+            fi
+        };
+        let errs = vec![None, None, None, None];
+
+        for (metas, expected) in [
+            (
+                vec![
+                    shape(1, Some(successor), 2),
+                    shape(2, None, 1),
+                    shape(3, Some(successor), 2),
+                    shape(4, None, 1),
+                ],
+                Some(successor),
+            ),
+            (
+                vec![
+                    shape(1, None, 1),
+                    shape(2, Some(successor), 2),
+                    shape(3, None, 1),
+                    shape(4, Some(successor), 2),
+                ],
+                None,
+            ),
+        ] {
+            // Each call hashes into fresh maps; a map-order choice would flip.
+            for _ in 0..32 {
+                let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                    .expect("one content identity reaches read quorum");
+
+                assert_eq!(selected.successor_mod_time, expected);
+                assert_eq!(selected.is_latest, expected.is_none());
+                assert_eq!(selected.num_versions, if expected.is_none() { 1 } else { 2 });
+            }
+        }
     }
 
     #[test]
@@ -12068,11 +12248,9 @@ mod tests {
         let metas = vec![first, second, FileInfo::default(), FileInfo::default()];
         let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)];
 
-        let (_, mut selected, selected_quorum) =
-            SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
-                .expect("read quorum should remain enough when no competing latest is visible");
+        let (_, mut selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("read quorum should remain enough when no competing latest is visible");
 
-        assert_eq!(selected_quorum, 2);
         assert_eq!(selected.data_dir, Some(data_dir));
         assert_eq!(selected.parts[0].etag, "part-etag-old");
         assert!(selected.parts[0].checksums.is_none());
@@ -12099,7 +12277,7 @@ mod tests {
         let metas = vec![first, second, FileInfo::default(), FileInfo::default()];
         let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)];
 
-        let (_, mut selected, _) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
+        let (_, mut selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
             .expect("winner selection should defer sidecar decoding");
         let err = SetDisks::hydrate_selected_fileinfo_part_checksums(&mut selected)
             .expect_err("a malformed degraded winner must fail closed when checksums are requested");
@@ -12166,10 +12344,9 @@ mod tests {
         let metas = vec![latest_meta, stale_stack_meta, newer_stack_meta, FileInfo::default()];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
             .expect("same object version should stay readable despite derived version stack drift");
 
-        assert_eq!(selected_quorum, 3);
         assert_eq!(selected.data_dir, Some(data_dir));
         assert_eq!(selected.parts[0].etag, "part-etag");
         assert_eq!(selected.mod_time, Some(mod_time));
@@ -12262,81 +12439,77 @@ mod tests {
         let metas = vec![meta_a, meta_b, meta_c, FileInfo::default()];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
             .expect("replication status drift should not split readable object identity");
 
-        assert_eq!(selected_quorum, 3);
         assert_eq!(selected.data_dir, Some(data_dir));
         assert_eq!(selected.parts[0].etag, "part-etag");
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_rejects_same_modtime_metadata_split_without_write_quorum() {
+    fn test_latest_fileinfo_selection_serves_same_time_metadata_majority() {
         let mod_time = OffsetDateTime::now_utc();
         let data_dir = Uuid::new_v4();
-        let mut old_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
-        let mut old_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2);
-        let mut partial_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3);
-        old_meta_a.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string());
-        old_meta_b.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string());
-        partial_meta
-            .metadata
-            .insert("x-amz-meta-color".to_string(), "red".to_string());
-        let metas = vec![old_meta_a, old_meta_b, partial_meta, FileInfo::default()];
+        let colored = |index, color: &str| {
+            let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+            fi.metadata.insert("x-amz-meta-color".to_string(), color.to_string());
+            fi
+        };
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3);
-        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3);
+        // With one disk missing, only the two-copy side can hold write quorum,
+        // whichever disk order the copies arrive in.
+        for (metas, expected) in [
+            (
+                vec![colored(1, "blue"), colored(2, "blue"), colored(3, "red"), FileInfo::default()],
+                "blue",
+            ),
+            (vec![colored(1, "blue"), colored(2, "red"), colored(3, "red"), FileInfo::default()], "red"),
+        ] {
+            let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                .expect("the read-quorum side of a same-time metadata split is served");
 
-        assert_eq!(quorum, 2);
-        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
+            assert_eq!(selected.metadata.get("x-amz-meta-color").map(String::as_str), Some(expected));
+        }
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_rejects_same_modtime_partial_metadata_read_quorum() {
+    fn test_latest_fileinfo_selection_serves_same_time_transition_majority() {
         let mod_time = OffsetDateTime::now_utc();
         let data_dir = Uuid::new_v4();
-        let mut old_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
-        let mut partial_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2);
-        let mut partial_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3);
-        old_meta.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string());
-        partial_meta_a
-            .metadata
-            .insert("x-amz-meta-color".to_string(), "red".to_string());
-        partial_meta_b
-            .metadata
-            .insert("x-amz-meta-color".to_string(), "red".to_string());
-        let metas = vec![old_meta, partial_meta_a, partial_meta_b, FileInfo::default()];
+        let local = |index| quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+        let remote = |index| {
+            let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+            fi.transition_status = TRANSITION_COMPLETE.to_string();
+            fi.transition_tier = "WARM".to_string();
+            fi.transitioned_objname = "remote/object".to_string();
+            fi.transition_version_id = Some(Uuid::from_u128(7));
+            fi
+        };
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3);
+        for (metas, expect_remote) in [
+            (vec![local(1), local(2), remote(3), FileInfo::default()], false),
+            (vec![local(1), remote(2), remote(3), FileInfo::default()], true),
+        ] {
+            let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                .expect("the read-quorum side of a same-time transition split is served");
 
-        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
+            assert_eq!(selected.is_remote(), expect_remote);
+        }
+
+        for (tied, expect_remote) in [
+            (vec![local(1), local(2), remote(3), remote(4)], false),
+            (vec![remote(1), local(2), local(3), remote(4)], true),
+        ] {
+            let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; tied.len()], &tied, &[None, None, None, None], 2)
+                .expect("a same-time half split serves the side on the lowest-index disk");
+            assert_eq!(selected.is_remote(), expect_remote);
+        }
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_rejects_same_modtime_transition_split_without_write_quorum() {
-        let mod_time = OffsetDateTime::now_utc();
-        let data_dir = Uuid::new_v4();
-        let old_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
-        let old_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2);
-        let mut partial_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3);
-        partial_meta.transition_status = TRANSITION_COMPLETE.to_string();
-        partial_meta.transition_tier = "WARM".to_string();
-        partial_meta.transitioned_objname = "remote/object".to_string();
-        partial_meta.transition_version_id = Some(Uuid::new_v4());
-        let metas = vec![old_meta_a, old_meta_b, partial_meta, FileInfo::default()];
-        let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
-
-        let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3);
-        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3);
-
-        assert_eq!(quorum, 2);
-        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
-    }
-
-    #[test]
-    fn test_latest_fileinfo_selection_quorum_uses_write_quorum_for_degraded_committed_identity() {
+    fn test_latest_fileinfo_selection_serves_degraded_committed_identity() {
         let mod_time = OffsetDateTime::now_utc();
         let data_dir = Uuid::new_v4();
         let metas = vec![
@@ -12347,7 +12520,10 @@ mod tests {
         ];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        assert_eq!(SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3), 3);
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("a committed identity stays readable with one disk missing");
+
+        assert_eq!(selected.parts[0].etag, "part-etag-a");
     }
 
     #[test]

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

@@ -9320,8 +9320,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         };
 
         let src_version_id = src_opts.version_id.as_deref().unwrap_or_default();
-        let (online_disks, mut fi, _) =
-            Self::select_valid_fileinfo(&disks, &metas, &errs, src_version_id, read_quorum, write_quorum)
+        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 {
@@ -11182,8 +11182,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
 
         let version_id = opts.version_id.as_deref().unwrap_or_default();
-        let (online_disks, mut fi, _) = Self::select_valid_fileinfo(&disks, &metas, &errs, version_id, read_quorum, write_quorum)
-            .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
+        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

@@ -966,8 +966,9 @@ impl SetDisks {
             return Err(to_object_err(err.into(), vec![bucket, object, &vid]));
         }
 
-        let (op_online_disks, mut fi, fileinfo_selection_quorum) =
-            match Self::select_valid_fileinfo(&disks, &parts_metadata, &errs, vid.as_str(), read_quorum, write_quorum) {
+        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);
@@ -987,7 +988,7 @@ impl SetDisks {
                 rustfs_utils::http::remove_str(&mut metadata.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
             }
         }
-        metadata_fanout_diagnostics.record_quorum_candidate_latency(metadata_metrics_path, fileinfo_selection_quorum);
+        metadata_fanout_diagnostics.record_quorum_candidate_latency(metadata_metrics_path, read_quorum);
         let needs_late_metadata_fanout = !metadata_fanout_complete
             && allow_early_stop
             && non_inline_data_read_early_stop_allowed(read_data, bucket, object)
@@ -1014,7 +1015,6 @@ impl SetDisks {
                     && allow_early_stop
                     && metadata_fanout_diagnostics.valid_responses() >= read_quorum
                     && errs.iter().all(Option::is_none)
-                    && fileinfo_selection_quorum >= read_quorum
                     && op_online_disks.iter().filter(|disk| disk.is_some()).count() >= read_quorum))
         {
             #[cfg(test)]
@@ -2465,8 +2465,8 @@ impl SetDisks {
             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, write_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]));
         }
@@ -4052,8 +4052,7 @@ mod metadata_cache_tests {
         assert!(original.is_complete());
         let (baseline, errors, _) = original.into_legacy();
         assert!(errors.iter().all(Option::is_none), "original real shards must all be readable");
-        let (_, selected, _) =
-            SetDisks::select_valid_fileinfo(&disks, &baseline, &errors, "", 2, 3).expect("original metadata quorum");
+        let (_, selected) = SetDisks::select_valid_fileinfo(&disks, &baseline, &errors, 2).expect("original metadata quorum");
         assert!(!selected.inline_data(), "late fixture must retain real non-inline shard files");
         let expected = LateMetadataIdentity::from_file_info(&selected);
         let (_, _, original_online) =
@@ -4086,7 +4085,7 @@ mod metadata_cache_tests {
                 .expect("read changed real disk metadata");
             assert!(changed.is_complete(), "late fixture must reach complete metadata observation");
             let (metadata, errors, _) = changed.into_legacy();
-            let (online, selected, _) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errors, "", 2, 3)
+            let (online, selected) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errors, 2)
                 .expect("changed metadata still reaches selection quorum");
             assert_eq!(online.iter().flatten().count(), 4, "the late identity guard must decide this failure");
             if changed_generation {
@@ -5135,7 +5134,7 @@ mod metadata_cache_tests {
         // This seam prepares a real, independently validated D1/P1/P2 quorum.
         // The hedge test separately exercises its production publication timing.
         let errs: Vec<Option<DiskError>> = (0..4).map(|_| None).collect();
-        let (online_disks, fi, _) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errs, "", 2, 3)
+        let (online_disks, fi) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errs, 2)
             .expect("three real members satisfy latest-generation selection");
         assert_eq!(online_disks.iter().flatten().count(), 3);
         assert_eq!(fi.metadata.get("etag"), committed.etag.as_ref());
@@ -6240,16 +6239,17 @@ mod tests {
         let parts_metadata = vec![first.clone(), second, third, FileInfo::default()];
         let errs = vec![None, None, None, None];
 
-        let (read_quorum, write_quorum) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 2)
+        let (read_quorum, _) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 2)
             .expect("partial early-stop metadata should preserve read quorum");
         let read_quorum = usize::try_from(read_quorum).expect("read quorum should be non-negative");
-        let write_quorum = usize::try_from(write_quorum).expect("write quorum should be non-negative");
-        let selection_quorum = SetDisks::latest_fileinfo_selection_quorum("", &parts_metadata, &errs, read_quorum, write_quorum);
+        let (_, latest) = SetDisks::select_valid_fileinfo(&vec![None; parts_metadata.len()], &parts_metadata, &errs, read_quorum)
+            .expect("partial early-stop metadata should select the candidate");
         let selected = SetDisks::pick_valid_fileinfo(&parts_metadata, None, Some("etag-1".to_string()), read_quorum)
             .expect("partial early-stop metadata should preserve selected FileInfo");
 
         assert_eq!(read_quorum, 2);
-        assert_eq!(selection_quorum, 3);
+        assert_eq!(latest.name, first.name);
+        assert_eq!(latest.get_etag(), first.get_etag());
         assert_eq!(selected.name, first.name);
         assert_eq!(selected.get_etag(), first.get_etag());
     }

crates/ecstore/src/store/list_objects.rs

@@ -6481,6 +6481,147 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    async fn latest_get_follows_read_quorum_on_torn_writes() {
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+        use crate::storage_api_contracts::object::ObjectOperations as _;
+
+        // HEAD and GET serve the version with the most copies once it reaches
+        // the object's data blocks, and a tie goes to the newer version, as in
+        // MinIO's getObjectFileInfo and as heal converges the set. GET used to
+        // demand write quorum whenever enough metadata answered, so a 2+2 or
+        // 8+8 split, or a newer side that could still hold write quorum
+        // through an offline drive, listed but answered 503.
+        let opts = ObjectOptions {
+            suppress_read_repair: true,
+            ..Default::default()
+        };
+        for (drives, parity) in [(4_usize, 2_usize), (16, 8), (16, 4)] {
+            let data = drives - parity;
+            let bucket = format!("torn-write-get-{drives}-{parity}");
+            let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks(drives, parity).await;
+            set.make_bucket(&bucket, &MakeBucketOptions::default())
+                .await
+                .expect("bucket should be created");
+            // Two committed generations of one key give every drive its own
+            // inline shard of each generation; the cases copy them per drive.
+            let mut generations = Vec::new();
+            for body in [b"old generation".as_slice(), b"new generation".as_slice()] {
+                let info = set
+                    .put_object(
+                        &bucket,
+                        "template",
+                        &mut PutObjReader::from_vec(body.to_vec()),
+                        &ObjectOptions {
+                            no_lock: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("generation PUT should commit on every drive");
+                let mut metadata = Vec::with_capacity(drives);
+                for dir in &dirs {
+                    metadata.push(
+                        tokio::fs::read(dir.path().join(&bucket).join("template").join(STORAGE_FORMAT_FILE))
+                            .await
+                            .expect("generation metadata should carry the inline shard"),
+                    );
+                }
+                generations.push((body, info.etag.expect("PUT should report an etag"), metadata));
+            }
+            let (new_body, new_etag, new) = generations.pop().expect("new generation");
+            let (old_body, old_etag, old) = generations.pop().expect("old generation");
+
+            // New copies occupy the first drives, so the offline drive always
+            // removes one new copy. Each phase reads its own keys so the GET
+            // metadata cache never answers for the other phase.
+            let mut cases = Vec::new();
+            for offline in [false, true] {
+                for copies in 1..drives {
+                    for overwrite in [false, true] {
+                        let name = format!(
+                            "{}-{}-{copies:02}",
+                            if offline { "offline" } else { "online" },
+                            if overwrite { "overwrite" } else { "new" }
+                        );
+                        for slot in 0..drives {
+                            let bytes = if slot < copies {
+                                &new[slot]
+                            } else if overwrite {
+                                &old[slot]
+                            } else {
+                                continue;
+                            };
+                            let path = dirs[slot].path().join(&bucket).join(&name);
+                            tokio::fs::create_dir_all(&path).await.expect("create object directory");
+                            tokio::fs::write(path.join(STORAGE_FORMAT_FILE), bytes)
+                                .await
+                                .expect("write generation metadata");
+                        }
+                        cases.push((offline, name, copies, overwrite));
+                    }
+                }
+            }
+
+            for offline in [false, true] {
+                let parked = if offline { set.disks.write().await[0].take() } else { None };
+                for (_, name, copies, overwrite) in cases.iter().filter(|case| case.0 == offline) {
+                    let new_copies = copies - usize::from(offline);
+                    let old_copies = if *overwrite { drives - copies } else { 0 };
+                    let expected = if new_copies >= data && new_copies >= old_copies {
+                        Some((&new_etag, new_body))
+                    } else if old_copies >= data {
+                        Some((&old_etag, old_body))
+                    } else {
+                        None
+                    };
+                    let context = format!("{drives} drives EC {data}+{parity}, {name}: {new_copies} new, {old_copies} old");
+                    let head = set.get_object_info(&bucket, name, &opts).await;
+                    let get = set.get_object_reader(&bucket, name, None, HeaderMap::new(), &opts).await;
+                    match (expected, head, get) {
+                        (Some((etag, body)), Ok(info), Ok(mut reader)) => {
+                            assert_eq!(info.etag.as_ref(), Some(etag), "{context}: HEAD");
+                            assert_eq!(reader.object_info.etag.as_ref(), Some(etag), "{context}: GET");
+                            let mut streamed = Vec::new();
+                            reader
+                                .stream
+                                .read_to_end(&mut streamed)
+                                .await
+                                .unwrap_or_else(|err| panic!("{context}: GET body: {err:?}"));
+                            assert_eq!(streamed, body, "{context}: GET body");
+                        }
+                        (Some((etag, _)), head, get) => panic!(
+                            "{context}: expected {etag}, HEAD {:?}, GET {:?}",
+                            head.map(|info| info.etag),
+                            get.map(|reader| reader.object_info.etag)
+                        ),
+                        (None, head, get) => {
+                            for (method, result) in [
+                                ("HEAD", head.map(|info| info.etag)),
+                                ("GET", get.map(|reader| reader.object_info.etag)),
+                            ] {
+                                let err = result.expect_err(&format!("{context}: {method} expected 404 or unavailable"));
+                                assert!(
+                                    matches!(
+                                        err,
+                                        StorageError::ObjectNotFound(..)
+                                            | StorageError::FileNotFound
+                                            | StorageError::ErasureReadQuorum
+                                            | StorageError::InsufficientReadQuorum(..)
+                                    ),
+                                    "{context}: {method} expected 404 or unavailable, got {err:?}"
+                                );
+                            }
+                        }
+                    }
+                }
+                if offline {
+                    set.disks.write().await[0] = parked;
+                }
+            }
+        }
+    }
+
     #[tokio::test]
     async fn latest_listing_and_get_agree_on_hermetic_set_states() {
         #[derive(Clone, Copy)]
@@ -6553,6 +6694,10 @@ mod test {
             ("d-torn-two", [Some(Generation::New), Some(Generation::New), None, None]),
             ("e-torn-one", [Some(Generation::New), None, None, None]),
             ("f-stale-one", [None, None, None, Some(Generation::Old)]),
+            (
+                "fa-overwrite-two-of-three",
+                [Some(Generation::New), Some(Generation::New), Some(Generation::Old), None],
+            ),
             (
                 "g-overwrite-torn-one",
                 [
@@ -6597,6 +6742,7 @@ mod test {
             ("b-copied", &new_etag),
             ("c-split", &new_etag),
             ("d-torn-two", &new_etag),
+            ("fa-overwrite-two-of-three", &new_etag),
             ("g-overwrite-torn-one", &old_etag),
         ]
         .into_iter()
@@ -6623,12 +6769,6 @@ mod test {
                         "{name}: GET must serve the version LIST selected"
                     );
                 }
-                // Known divergence: when every drive answers, GET of the latest
-                // version demands write quorum (three of four), so a 2+2 split
-                // answers 503 although LIST shows the newer generation, as
-                // MinIO's listing does. F.0's dangling classification does not
-                // cover a split; a GET change is a separate decision.
-                (Some(_), Err(StorageError::InsufficientReadQuorum(..))) if name == "c-split" => {}
                 (Some(_), Err(err)) => assert!(
                     matches!(
                         err,

docs/architecture/erasure-coding.md

@@ -313,7 +313,7 @@ Small objects store their payload inline after the container CRC ([filemeta_inli
 ## 8. Read path and read quorum
 
 - **INVARIANT — read quorum = `data_blocks`.** `object_quorum_from_meta` returns `(read_quorum = data_blocks, write_quorum)` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)); `parity_blocks = common_parity(...)` is the parity value held by the most disks that still reaches its own read quorum. When `default_parity_count == 0`, read = write = all shards.
-- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. Latest-version reads may escalate to write_quorum to avoid resurrecting a partially-overwritten version.
+- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. Reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set.
 - **INVARIANT — decode needs ≥ `data_blocks` shards.** The stripe reader requires `available_shards ≥ data_shards`; below that the read fails closed with a read-quorum error (never silent truncation) ([set_disk/read.rs](../../crates/ecstore/src/set_disk/read.rs), [set_disk/shard_source.rs](../../crates/ecstore/src/set_disk/shard_source.rs)). Codec geometry is validated at construction: `Erasure::try_new` / `try_new_with_options` ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)) return `ErasureConstructionError` for `data_shards == 0`, `block_size == 0`, shard-count overflow, or an unsupported shard configuration, so a read never reaches a `block_size` / `data_shards` division with invalid geometry (§13).
 - If `available ≥ data_blocks` but some shards are missing, the read is served **and** a background read-repair heal is enqueued.
 - **INVARIANT — cross-stripe read verification.** When a data shard is missing and `available > data_blocks`, reconstruction regenerates parity and compares it to the surviving parity; a mismatch is `InvalidData "inconsistent read source shards"` (backlog#832), catching corruption that passed per-shard bitrot but disagrees across the stripe ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)).