fix(ecstore): recover late parity after exact quorum (#6927)

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/erasure/coding/decode.rs

@@ -46,6 +46,7 @@ type ShardReadFuture<'a> = Pin<Box<dyn Future<Output = (usize, ShardReadCost, Re
 type OwnedShardReadFuture<'a, R> =
     Pin<Box<dyn Future<Output = (usize, ShardReadCost, Result<Vec<u8>, Error>, Option<BitrotReader<R>>, bool)> + Send + 'a>>;
 pub(crate) type DeferredReaderReopener<R> = Arc<dyn Fn(usize) -> Option<BitrotReader<R>> + Send + Sync>;
+pub(crate) type DecodeOutcome = (usize, Option<std::io::Error>, bool);
 
 type ShardIndexes = SmallVec<[usize; INLINE_SHARD_SLOTS]>;
 type ActiveReaders = SmallVec<[bool; INLINE_SHARD_SLOTS]>;
@@ -2190,8 +2191,10 @@ impl Erasure {
         W: AsyncWrite + Send + Sync + Unpin,
         R: crate::erasure::coding::ShardSource,
     {
-        self.decode_inner(writer, readers, offset, length, total_length, None, Vec::new(), Vec::new())
+        let (written, error, _) = self
-            .await
+            .decode_inner(writer, readers, offset, length, total_length, None, Vec::new(), Vec::new())
+            .await;
+        (written, error)
     }
 
     #[allow(dead_code, reason = "read-cost decode path asserted by this file's tests (backlog#1823)")]
@@ -2208,8 +2211,10 @@ impl Erasure {
         W: AsyncWrite + Send + Sync + Unpin,
         R: crate::erasure::coding::ShardSource,
     {
-        self.decode_inner(writer, readers, offset, length, total_length, Some(read_costs), Vec::new(), Vec::new())
+        let (written, error, _) = self
-            .await
+            .decode_inner(writer, readers, offset, length, total_length, Some(read_costs), Vec::new(), Vec::new())
+            .await;
+        (written, error)
     }
 
     /// GET decode entry point that also carries the deferred-parity stripe
@@ -2262,6 +2267,37 @@ impl Erasure {
         deferred_handles: Vec<Option<DeferredReaderStripeHandle>>,
         deferred_reopeners: Vec<Option<DeferredReaderReopener<R>>>,
     ) -> (usize, Option<std::io::Error>)
+    where
+        W: AsyncWrite + Send + Sync + Unpin,
+        R: crate::erasure::coding::ShardSource,
+    {
+        let (written, error, _) = self
+            .decode_inner(
+                writer,
+                readers,
+                offset,
+                length,
+                total_length,
+                read_costs,
+                deferred_handles,
+                deferred_reopeners,
+            )
+            .await;
+        (written, error)
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) async fn decode_with_stripe_handles_and_reopeners_with_diagnostics<W, R>(
+        &self,
+        writer: &mut W,
+        readers: Vec<Option<BitrotReader<R>>>,
+        offset: usize,
+        length: usize,
+        total_length: usize,
+        read_costs: Option<Vec<ShardReadCost>>,
+        deferred_handles: Vec<Option<DeferredReaderStripeHandle>>,
+        deferred_reopeners: Vec<Option<DeferredReaderReopener<R>>>,
+    ) -> DecodeOutcome
     where
         W: AsyncWrite + Send + Sync + Unpin,
         R: crate::erasure::coding::ShardSource,
@@ -2411,36 +2447,48 @@ impl Erasure {
         read_costs: Option<Vec<ShardReadCost>>,
         deferred_handles: Vec<Option<DeferredReaderStripeHandle>>,
         deferred_reopeners: Vec<Option<DeferredReaderReopener<R>>>,
-    ) -> (usize, Option<std::io::Error>)
+    ) -> DecodeOutcome
     where
         W: AsyncWrite + Send + Sync + Unpin,
         R: crate::erasure::coding::ShardSource,
     {
         if readers.len() != self.data_shards + self.parity_shards {
             record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid);
-            return (0, Some(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers")));
+            return (0, Some(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers")), false);
         }
 
         // block_size/data_shards come from on-disk metadata; a corrupt FileInfo with a
         // zero here must surface as an error, not a divide-by-zero panic on every GET.
         if self.block_size == 0 || self.data_shards == 0 {
             record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid);
-            return (0, Some(io::Error::new(ErrorKind::InvalidInput, "Invalid erasure coding parameters")));
+            return (
+                0,
+                Some(io::Error::new(ErrorKind::InvalidInput, "Invalid erasure coding parameters")),
+                false,
+            );
         }
 
         let Some(end_offset) = offset.checked_add(length) else {
             record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid);
-            return (0, Some(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length")));
+            return (
+                0,
+                Some(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length")),
+                false,
+            );
         };
         if end_offset > total_length {
             record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid);
-            return (0, Some(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length")));
+            return (
+                0,
+                Some(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length")),
+                false,
+            );
         }
 
         let mut ret_err = None;
 
         if length == 0 {
-            return (0, ret_err);
+            return (0, ret_err, false);
         }
 
         let mut written = 0;
@@ -2480,6 +2528,7 @@ impl Erasure {
             }
         };
 
+        let mut exact_quorum = false;
         if legacy_stripe_prefetch_enabled() {
             // Depth-1 stripe prefetch (backlog#930 HP-9 step 2): while the current
             // stripe is reconstructed and emitted, the next stripe's shard reads
@@ -2522,6 +2571,7 @@ impl Erasure {
                     let Some((mut shards, errs)) = current.take() else {
                         break;
                     };
+                    exact_quorum |= shards.iter().filter(|shard| shard.is_some()).count() == self.data_shards;
 
                     if idx + 1 < blocks.len() {
                         // Overlap: read stripe idx+1 while reconstructing/emitting idx.
@@ -2636,6 +2686,7 @@ impl Erasure {
                 let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
                 let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
                 let (mut shards, errs) = reader.read().await;
+                exact_quorum |= shards.iter().filter(|shard| shard.is_some()).count() == self.data_shards;
                 record_get_stage_duration_if_enabled(
                     GET_OBJECT_PATH_LEGACY_DUPLEX,
                     GET_STAGE_STRIPE_READ,
@@ -2665,14 +2716,14 @@ impl Erasure {
         }
 
         if ret_err.is_some() {
-            return (written, ret_err);
+            return (written, ret_err, exact_quorum);
         }
 
         if written < length {
             ret_err = Some(Error::LessData.into());
         }
 
-        (written, ret_err)
+        (written, ret_err, exact_quorum)
     }
 }
 

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

@@ -48,14 +48,15 @@ use super::super::{
     ObjectPartInfo, OffsetDateTime, RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, RawFileInfo, ReadMultipleReq,
     ReadMultipleResp, ReadOptions, Result, SLASH_SEPARATOR, STORAGE_FORMAT_FILE, SetDisks, SnapshotLeaseToken, StorageError,
     UpdateMetadataOpts, Uuid, build_inline_bitrot_readers_from_refs, can_try_inline_data_shards_direct,
-    capacity_scope_from_disks, coding, collect_inline_data_shard_fileinfos_by_index_or_reason, current_dirty_generation, debug,
+    capacity_scope_from_disks, codec_streaming_rollout_applies, coding, collect_inline_data_shard_fileinfos_by_index_or_reason,
-    disk, file_info_is_valid_for_metadata, get_metadata_slowtail_fault_request, info, inline_erasure_shard_file_offset,
+    current_dirty_generation, debug, disk, file_info_is_valid_for_metadata, get_metadata_slowtail_fault_request, info,
-    inline_erasure_shard_size, is_err_object_not_found, is_err_version_not_found, is_get_metadata_data_read_early_stop_enabled,
+    inline_erasure_shard_file_offset, inline_erasure_shard_size, is_err_object_not_found, is_err_version_not_found,
-    is_get_metadata_early_stop_bounded_fanout_enabled, is_get_metadata_early_stop_enabled,
+    is_get_metadata_data_read_early_stop_enabled, is_get_metadata_early_stop_bounded_fanout_enabled,
-    is_get_metadata_non_inline_data_read_early_stop_enabled, is_object_dangling, is_version_early_stop_enabled,
+    is_get_metadata_early_stop_enabled, is_get_metadata_non_inline_data_read_early_stop_enabled, is_object_dangling,
-    issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure, merge_file_meta_versions, path_join_buf,
+    is_version_early_stop_enabled, issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure,
-    record_global_dirty_scope, reduce_read_quorum_errs, reduce_write_quorum_errs, send_heal_request_with_admission,
+    merge_file_meta_versions, object_fits_single_block, path_join_buf, record_global_dirty_scope, reduce_read_quorum_errs,
-    should_prevent_write, to_object_err, try_read_inline_data_shards_direct, warn,
+    reduce_write_quorum_errs, send_heal_request_with_admission, should_prevent_write, to_object_err,
+    try_read_inline_data_shards_direct, warn,
 };
 #[cfg(test)]
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
@@ -1132,7 +1133,7 @@ fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Op
     None
 }
 
-fn non_inline_data_read_candidate_is_safe(candidate: &FileInfo) -> bool {
+pub(in crate::set_disk) fn non_inline_data_read_candidate_is_safe(candidate: &FileInfo) -> bool {
     if candidate.inline_data()
         || candidate.is_compressed()
         || candidate.is_remote()
@@ -1147,6 +1148,16 @@ fn non_inline_data_read_candidate_is_safe(candidate: &FileInfo) -> bool {
     candidate.has_valid_erasure_geometry()
 }
 
+pub(in crate::set_disk) fn late_materialization_candidate_is_safe(candidate: &FileInfo) -> bool {
+    non_inline_data_read_candidate_is_safe(candidate)
+        && candidate.size > 512 * 1024
+        && object_fits_single_block(candidate.size, candidate.erasure.block_size)
+}
+
+pub(in crate::set_disk) fn non_inline_data_read_early_stop_allowed(read_data: bool, bucket: &str, object: &str) -> bool {
+    read_data && is_get_metadata_non_inline_data_read_early_stop_enabled() && !codec_streaming_rollout_applies(bucket, object)
+}
+
 const NON_INLINE_SINGLE_PENDING_HEDGE_DELAY: Duration = Duration::from_millis(100);
 
 fn data_read_inline_missing_shards_are_pending(
@@ -2931,7 +2942,7 @@ impl SetDisks {
                 read_data,
                 healing,
                 incl_free_versions,
-                read_data && is_get_metadata_non_inline_data_read_early_stop_enabled(),
+                non_inline_data_read_early_stop_allowed(read_data, bucket, object),
                 default_parity_count,
                 allow_coalescing,
             )
@@ -7022,6 +7033,27 @@ mod tests {
     use tempfile::TempDir;
     use tokio::io::AsyncReadExt;
 
+    #[test]
+    #[serial_test::serial(codec_streaming_env)]
+    fn non_inline_early_stop_is_mutually_exclusive_with_codec_rollout() {
+        temp_env::with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true")),
+                ("RUSTFS_GET_CODEC_STREAMING_ROLLOUT", Some("on")),
+                ("RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED", Some("true")),
+                ("RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED", Some("true")),
+            ],
+            || assert!(!non_inline_data_read_early_stop_allowed(true, "bucket", "object")),
+        );
+        temp_env::with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true")),
+                ("RUSTFS_GET_CODEC_STREAMING_ROLLOUT", Some("off")),
+            ],
+            || assert!(non_inline_data_read_early_stop_allowed(true, "bucket", "object")),
+        );
+    }
+
     #[tokio::test]
     async fn scanner_delete_owner_survives_waiter_cancellation() {
         let movement_gate = Arc::new(tokio::sync::RwLock::new(()));

crates/ecstore/src/set_disk/mod.rs

@@ -894,8 +894,11 @@ struct OwnedGetObjectFileInfo {
     fi: FileInfo,
     parts_metadata: Vec<FileInfo>,
     online_disks: Vec<Option<DiskStore>>,
+    late_metadata_fanout_disks: Option<Vec<Option<DiskStore>>>,
 }
 
+type OwnedGetObjectFileInfoParts = (FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>, Option<Vec<Option<DiskStore>>>);
+
 impl GetObjectFileInfo {
     fn owned(fi: FileInfo, parts_metadata: Vec<FileInfo>, online_disks: Vec<Option<DiskStore>>) -> Self {
         Self {
@@ -903,6 +906,24 @@ impl GetObjectFileInfo {
                 fi,
                 parts_metadata,
                 online_disks,
+                late_metadata_fanout_disks: None,
+            }),
+            shared: None,
+        }
+    }
+
+    fn owned_with_late_metadata_fanout(
+        fi: FileInfo,
+        parts_metadata: Vec<FileInfo>,
+        online_disks: Vec<Option<DiskStore>>,
+        late_metadata_fanout_disks: Vec<Option<DiskStore>>,
+    ) -> Self {
+        Self {
+            owned: Some(OwnedGetObjectFileInfo {
+                fi,
+                parts_metadata,
+                online_disks,
+                late_metadata_fanout_disks: Some(late_metadata_fanout_disks),
             }),
             shared: None,
         }
@@ -939,19 +960,28 @@ impl GetObjectFileInfo {
         }
     }
 
+    fn has_late_metadata_fanout(&self) -> bool {
+        self.owned
+            .as_ref()
+            .is_some_and(|snapshot| snapshot.late_metadata_fanout_disks.is_some())
+    }
+
     fn into_owned(self) -> (FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>) {
+        let (fi, parts_metadata, online_disks, _) = self.into_owned_with_late_metadata_fanout();
+        (fi, parts_metadata, online_disks)
+    }
+
+    fn into_owned_with_late_metadata_fanout(self) -> OwnedGetObjectFileInfoParts {
         match (self.owned, self.shared) {
-            (Some(snapshot), None) => {
+            (Some(snapshot), None) => (
-                let OwnedGetObjectFileInfo {
+                snapshot.fi,
-                    fi,
+                snapshot.parts_metadata,
-                    parts_metadata,
+                snapshot.online_disks,
-                    online_disks,
+                snapshot.late_metadata_fanout_disks,
-                } = snapshot;
+            ),
-                (fi, parts_metadata, online_disks)
-            }
             (None, Some(entry)) => match Arc::try_unwrap(entry) {
-                Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks),
+                Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks, None),
-                Err(entry) => (entry.fi.clone(), entry.parts_metadata.clone(), entry.online_disks.clone()),
+                Err(entry) => (entry.fi.clone(), entry.parts_metadata.clone(), entry.online_disks.clone(), None),
             },
             _ => unreachable!("GET metadata snapshot representation must be exclusive"),
         }
@@ -1232,6 +1262,278 @@ mod prepared_get_object_metadata_tests {
         .await;
     }
 
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn non_inline_two_phase_read_fetches_late_parity_after_two_selected_shards_fail() {
+        let (dirs, set_disks) = make_local_set_disks(4, 2).await;
+        let bucket = "non-inline-read-late-parity";
+        let object = object_with_initial_data_shards(bucket, "late-parity-object");
+        let payload = vec![0x5a; 1024 * 1024];
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        let mut put_reader = PutObjReader::from_vec(payload.clone());
+        set_disks
+            .put_object(bucket, &object, &mut put_reader, &opts)
+            .await
+            .expect("object should be written");
+
+        let order = bounded_metadata_fanout_order(bucket, &object, 4, 2);
+        let distribution = FileInfo::new(&[bucket, object.as_str()].join("/"), 2, 2).erasure.distribution;
+        assert!(
+            order.iter().take(2).all(|disk_index| distribution[*disk_index] <= 2),
+            "the two failed selected shards must be data shards"
+        );
+        assert!(
+            distribution[order[3]] > 2,
+            "the metadata shard omitted by the plan must be healthy parity"
+        );
+        for disk_index in order.iter().take(2) {
+            let object_dir = dirs[*disk_index].path().join(bucket).join(&object);
+            let data_dir = std::fs::read_dir(&object_dir)
+                .expect("object directory should be readable")
+                .find_map(|entry| {
+                    let entry = entry.expect("object directory entry should be readable");
+                    entry
+                        .file_type()
+                        .expect("object directory entry type should be readable")
+                        .is_dir()
+                        .then(|| entry.path())
+                })
+                .expect("object data directory should exist");
+            let part_path = data_dir.join("part.1");
+            let mut shard = std::fs::read(&part_path).expect("selected data shard should be readable before corruption");
+            shard[0] ^= 0xff;
+            std::fs::write(part_path, shard).expect("selected data shard should be corrupted after metadata was written");
+        }
+
+        temp_env::async_with_vars(
+            [
+                ("RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", Some("true")),
+                ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+            ],
+            async {
+                let calls = disk_call_counters::observe(&object);
+                let mut reader = set_disks
+                    .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
+                    .await
+                    .expect("two-phase GET should recover using late parity");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("late parity should restore the exact GET body");
+                assert_eq!(restored, payload);
+                assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 7);
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn non_inline_two_phase_read_fetches_late_parity_when_selected_parts_are_missing() {
+        let (dirs, set_disks) = make_local_set_disks(4, 2).await;
+        let bucket = "non-inline-read-late-parity-missing";
+        let object = object_with_initial_data_shards(bucket, "late-parity-missing-object");
+        let payload = vec![0x3c; 1024 * 1024];
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        let mut put_reader = PutObjReader::from_vec(payload.clone());
+        set_disks
+            .put_object(bucket, &object, &mut put_reader, &opts)
+            .await
+            .expect("object should be written");
+
+        let order = bounded_metadata_fanout_order(bucket, &object, 4, 2);
+        for disk_index in order.iter().take(2) {
+            let object_dir = dirs[*disk_index].path().join(bucket).join(&object);
+            let data_dir = std::fs::read_dir(&object_dir)
+                .expect("object directory should be readable")
+                .find_map(|entry| {
+                    let entry = entry.expect("object directory entry should be readable");
+                    entry
+                        .file_type()
+                        .expect("entry type should be readable")
+                        .is_dir()
+                        .then(|| entry.path())
+                })
+                .expect("object data directory should exist");
+            std::fs::remove_file(data_dir.join("part.1")).expect("selected data shard should be removed");
+        }
+
+        temp_env::async_with_vars(
+            [
+                ("RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", Some("true")),
+                ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+            ],
+            async {
+                let calls = disk_call_counters::observe(&object);
+                let mut reader = set_disks
+                    .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
+                    .await
+                    .expect("two-phase GET should recover using late parity");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("late parity should restore the exact GET body");
+                assert_eq!(restored, payload);
+                assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 7);
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn four_data_two_parity_two_phase_read_recovers_one_failed_data_shard() {
+        let (dirs, set_disks) = make_local_set_disks(6, 2).await;
+        let bucket = "four-data-two-parity-late-read";
+        let object = object_with_initial_data_shards_for_geometry(bucket, "one-failed-data", 4, 2);
+        let payload = vec![0x7a; 1024 * 1024];
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        let mut put_reader = PutObjReader::from_vec(payload.clone());
+        set_disks
+            .put_object(bucket, &object, &mut put_reader, &opts)
+            .await
+            .expect("object should be written");
+
+        let order = bounded_metadata_fanout_order(bucket, &object, 6, 2);
+        let distribution = FileInfo::new(&[bucket, object.as_str()].join("/"), 4, 2).erasure.distribution;
+        let failed_disk = *order
+            .iter()
+            .take(4)
+            .find(|disk_index| distribution[**disk_index] <= 4)
+            .expect("initial fanout should include a data shard");
+        assert!(
+            order.iter().take(4).all(|disk_index| distribution[*disk_index] <= 4),
+            "initial fanout should cover all four data shards"
+        );
+        assert!(distribution[order[5]] > 4, "the final deferred metadata shard should be parity");
+
+        let object_dir = dirs[failed_disk].path().join(bucket).join(&object);
+        let data_dir = std::fs::read_dir(&object_dir)
+            .expect("object directory should be readable")
+            .find_map(|entry| {
+                let entry = entry.expect("object directory entry should be readable");
+                entry
+                    .file_type()
+                    .expect("object directory entry type should be readable")
+                    .is_dir()
+                    .then(|| entry.path())
+            })
+            .expect("object data directory should exist");
+        let part_path = data_dir.join("part.1");
+        let mut shard = std::fs::read(&part_path).expect("selected data shard should be readable before corruption");
+        shard[0] ^= 0xff;
+        std::fs::write(part_path, shard).expect("selected data shard should be corrupted after metadata was written");
+
+        temp_env::async_with_vars(
+            [
+                ("RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", Some("true")),
+                ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+            ],
+            async {
+                let calls = disk_call_counters::observe(&object);
+                let mut reader = set_disks
+                    .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
+                    .await
+                    .expect("two-phase GET should recover with one failed data shard");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("late parity should restore the exact GET body");
+                assert_eq!(restored, payload);
+                assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 11);
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn four_data_two_parity_two_phase_read_rejects_below_read_quorum() {
+        let (dirs, set_disks) = make_local_set_disks(6, 2).await;
+        let bucket = "four-data-two-parity-quorum-minus-one";
+        let object = object_with_initial_data_shards_for_geometry(bucket, "quorum-minus-one", 4, 2);
+        let payload = vec![0x4b; 1024 * 1024];
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        let mut put_reader = PutObjReader::from_vec(payload);
+        set_disks
+            .put_object(bucket, &object, &mut put_reader, &opts)
+            .await
+            .expect("object should be written");
+
+        let order = bounded_metadata_fanout_order(bucket, &object, 6, 2);
+        for disk_index in order.iter().take(3) {
+            let object_dir = dirs[*disk_index].path().join(bucket).join(&object);
+            let data_dir = std::fs::read_dir(&object_dir)
+                .expect("object directory should be readable")
+                .find_map(|entry| {
+                    let entry = entry.expect("object directory entry should be readable");
+                    entry
+                        .file_type()
+                        .expect("entry type should be readable")
+                        .is_dir()
+                        .then(|| entry.path())
+                })
+                .expect("object data directory should exist");
+            std::fs::remove_file(data_dir.join("part.1")).expect("selected shard should be removed");
+        }
+
+        temp_env::async_with_vars(
+            [
+                ("RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", Some("true")),
+                ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+            ],
+            async {
+                let result = set_disks
+                    .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
+                    .await;
+                assert!(result.is_err(), "quorum-minus-one read must fail closed without exposing a body");
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     #[serial_test::serial(body_cache_hook)]
     async fn non_inline_data_read_early_stop_keeps_reserve_on_unequal_layout() {
@@ -2340,6 +2642,15 @@ fn should_use_codec_streaming(config: GetCodecStreamingConfig, bucket: &str, obj
     is_optimization_enabled_for_request(config.enabled, config.rollout_pct, bucket, object)
 }
 
+pub(in crate::set_disk) fn codec_streaming_rollout_applies(bucket: &str, object: &str) -> bool {
+    let config = get_codec_streaming_config();
+    config.enabled
+        && config.body_compat_confirmed
+        && config.header_compat_confirmed
+        && config.rollout.is_opted_in()
+        && should_use_codec_streaming(config, bucket, object)
+}
+
 /// Should this specific request use metadata early-stop?
 #[allow(
     dead_code,
@@ -12250,6 +12561,7 @@ mod tests {
                 0,
                 true,
                 false,
+                false,
                 GET_OBJECT_PATH_LEGACY_DUPLEX,
                 GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
                 metrics_size_bucket,
@@ -12362,6 +12674,7 @@ mod tests {
                 0,
                 true,
                 false,
+                false,
                 GET_OBJECT_PATH_LEGACY_DUPLEX,
                 GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
                 metrics_size_bucket,

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

@@ -1379,6 +1379,14 @@ fn data_read_metadata_early_stop_request_shape_allowed(range: &Option<HTTPRangeS
         && !crate::object_api::restore_request_active(opts)
 }
 
+fn prepare_late_materialized_retry(initial_result: &Result<()>, output: &mut Vec<u8>, expected_size: usize) -> bool {
+    if initial_result.is_ok() && output.len() == expected_size {
+        return false;
+    }
+    output.clear();
+    true
+}
+
 #[cfg(test)]
 mod data_read_metadata_early_stop_request_shape_tests {
     use super::*;
@@ -2012,6 +2020,88 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             }
         }
 
+        if snapshot.has_late_metadata_fanout() {
+            // Keep refresh plus the second decode off the default GET poll stack.
+            // The allocation is limited to the opt-in late-materialization path.
+            return Box::pin(async move {
+                let object_size = usize::try_from(object_info.size)
+                    .map_err(|_| to_object_err(Error::other("two-phase GET object size is invalid"), vec![bucket, object]))?;
+                let mut output = Vec::with_capacity(object_size);
+                let (fi, files, disks, late_metadata_fanout_disks) = snapshot.into_owned_with_late_metadata_fanout();
+                let expected_identity = super::super::read::LateMetadataIdentity::from_file_info(&fi);
+                let late_metadata_fanout_disks = late_metadata_fanout_disks.ok_or_else(|| {
+                    to_object_err(Error::other("two-phase GET fallback context is missing"), vec![bucket, object])
+                })?;
+                let initial_result = Self::get_object_with_fileinfo(
+                    bucket,
+                    object,
+                    Arc::clone(&self.erasure_cache),
+                    0,
+                    object_info.size,
+                    &mut output,
+                    fi,
+                    files,
+                    &disks,
+                    self.set_index,
+                    self.pool_index,
+                    opts.skip_verify_bitrot,
+                    true,
+                    true,
+                    GET_OBJECT_PATH_LEGACY_DUPLEX,
+                    object_class.as_str(),
+                    size_bucket,
+                )
+                .await;
+                if prepare_late_materialized_retry(&initial_result, &mut output, object_size) {
+                    let (full_fi, full_parts_metadata, full_online_disks) = Self::refresh_late_metadata_fanout(
+                        &late_metadata_fanout_disks,
+                        bucket,
+                        object,
+                        &expected_identity,
+                        GET_OBJECT_PATH_LEGACY_DUPLEX,
+                    )
+                    .await?;
+                    Self::get_object_with_fileinfo(
+                        bucket,
+                        object,
+                        Arc::clone(&self.erasure_cache),
+                        0,
+                        object_info.size,
+                        &mut output,
+                        full_fi,
+                        full_parts_metadata,
+                        &full_online_disks,
+                        self.set_index,
+                        self.pool_index,
+                        opts.skip_verify_bitrot,
+                        true,
+                        false,
+                        GET_OBJECT_PATH_LEGACY_DUPLEX,
+                        object_class.as_str(),
+                        size_bucket,
+                    )
+                    .await?;
+                }
+                if output.len() != object_size {
+                    return Err(to_object_err(Error::other("two-phase GET decoded length mismatch"), vec![bucket, object]));
+                }
+
+                record_get_object_reader_path_observation(GET_OBJECT_PATH_LEGACY_DUPLEX, object_class, size_bucket);
+                let body = Bytes::from(output);
+                let reader = GetObjectReader {
+                    stream: Box::new(Cursor::new(body.clone())),
+                    object_info,
+                    buffered_body: Some(body),
+                    body_source,
+                };
+                if lock_optimization_enabled {
+                    release_materialized_read_lock(bucket, object, read_lock_guard.take());
+                }
+                Ok(reader)
+            })
+            .await;
+        }
+
         let direct_memory_decision = get_small_object_direct_memory_decision_with_threshold_and_plan(
             &range,
             &object_info,
@@ -2073,6 +2163,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 self.pool_index,
                 opts.skip_verify_bitrot,
                 true,
+                false,
                 GET_OBJECT_PATH_DIRECT_MEMORY,
                 object_class.as_str(),
                 size_bucket,
@@ -2272,6 +2363,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                         pool_index,
                         skip_verify,
                         false,
+                        false,
                         GET_OBJECT_PATH_LEGACY_DUPLEX,
                         object_class.as_str(),
                         size_bucket,
@@ -7821,6 +7913,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 pool_index,
                 skip_verify,
                 false,
+                false,
                 GET_OBJECT_PATH_LEGACY_DUPLEX,
                 GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
                 metrics_size_bucket,
@@ -9791,6 +9884,9 @@ mod inline_put_commit_path_tests {
     use super::*;
     use crate::config::storageclass::{INLINE_BLOCK_ENV, lookup_config_for_pools, lookup_config_for_pools_without_env};
     use crate::disk::ReadOptions;
+    use crate::ecstore_validation_blackbox::make_local_set_disks;
+    use crate::set_disk::disk_call_counters;
+    use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
     use rustfs_config::server_config::KVS;
     use serial_test::serial;
     use tokio::io::AsyncReadExt;
@@ -9963,6 +10059,69 @@ mod inline_put_commit_path_tests {
         .await;
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn get_object_reader_codec_rollout_excludes_late_metadata_refresh() {
+        let (_temp_dirs, set_disks) = make_local_set_disks(4, 2).await;
+        let bucket = "one-mib-codec-reader";
+        let object = "object.bin";
+        let payload = vec![0x6b; 1024 * 1024];
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("codec bucket should be created");
+        let storage_class = temp_env::with_var(INLINE_BLOCK_ENV, Some("1KiB"), || lookup_config_for_pools(&KVS::new(), &[4]))
+            .expect("test storage class should resolve");
+        set_disks.set_test_storage_class_config(storage_class);
+
+        let mut writer = PutObjReader::from_vec(payload.clone());
+        temp_env::async_with_vars(
+            [
+                ("RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", Some("true")),
+                (ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE, Some("false")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some("legacy")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE, Some("true")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE, Some("false")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("on")),
+                (rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("true")),
+            ],
+            async {
+                set_disks
+                    .put_object(bucket, object, &mut writer, &ObjectOptions::default())
+                    .await
+                    .expect("codec fixture should commit");
+                crate::set_disk::reset_test_get_object_reader_path();
+                let calls = disk_call_counters::observe(object);
+                let mut reader = set_disks
+                    .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("codec GET should succeed");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("codec GET should stream");
+                assert_eq!(restored, payload);
+                assert_eq!(
+                    crate::set_disk::test_get_object_reader_path_id(),
+                    5,
+                    "codec path must win over late refresh"
+                );
+                assert_eq!(
+                    calls.total(disk_call_counters::KIND_READ_VERSION),
+                    4,
+                    "codec path must use full metadata fanout and must not trigger a second late refresh"
+                );
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     async fn repeated_gets_reuse_the_set_erasure_shell() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;

crates/ecstore/src/set_disk/read.rs

@@ -609,7 +609,20 @@ impl SetDisks {
 
         // let online_disks: Vec<Option<DiskStore>> = op_online_disks.iter().filter(|v| v.is_some()).cloned().collect();
 
-        Ok(GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks))
+        if !metadata_fanout_complete
+            && allow_early_stop
+            && non_inline_data_read_early_stop_allowed(read_data, bucket, object)
+            && late_materialization_candidate_is_safe(&fi)
+        {
+            Ok(GetObjectFileInfo::owned_with_late_metadata_fanout(
+                fi,
+                parts_metadata,
+                op_online_disks,
+                disks,
+            ))
+        } else {
+            Ok(GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks))
+        }
     }
 
     #[hotpath::measure(impl_type = "SetDisks")]
@@ -819,6 +832,7 @@ impl SetDisks {
         pool_index: usize,
         skip_verify_bitrot: bool,
         prefer_data_blocks_first_reader_setup: bool,
+        require_reconstruction_surplus: bool,
         metrics_path: &'static str,
         metrics_object_class: &'static str,
         metrics_size_bucket: &'static str,
@@ -1083,6 +1097,9 @@ impl SetDisks {
             }
 
             let nil_count = reader_setup.available_shards();
+            if require_reconstruction_surplus && nil_count <= erasure.data_shards {
+                return Err(Error::other("insufficient reconstruction surplus for two-phase read"));
+            }
             if nil_count < erasure.data_shards {
                 if let Some(read_err) = reduce_read_quorum_errs(&reader_setup.errors, OBJECT_OP_IGNORED_ERRS, erasure.data_shards)
                 {
@@ -1190,18 +1207,34 @@ impl SetDisks {
             let readers = reader_setup.readers;
             let deferred_stripe_handles = reader_setup.deferred_stripe_handles;
             let deferred_reopeners = reader_setup.deferred_reopeners;
-            let (written, err) = erasure
+            let (written, err, exact_quorum) = if require_reconstruction_surplus {
-                .decode_with_stripe_handles_and_reopeners(
+                erasure
-                    writer,
+                    .decode_with_stripe_handles_and_reopeners_with_diagnostics(
-                    readers,
+                        writer,
-                    part_offset,
+                        readers,
-                    part_length,
+                        part_offset,
-                    part_size,
+                        part_length,
-                    read_costs,
+                        part_size,
-                    deferred_stripe_handles,
+                        read_costs,
-                    deferred_reopeners,
+                        deferred_stripe_handles,
-                )
+                        deferred_reopeners,
-                .await;
+                    )
+                    .await
+            } else {
+                let (written, err) = erasure
+                    .decode_with_stripe_handles_and_reopeners(
+                        writer,
+                        readers,
+                        part_offset,
+                        part_length,
+                        part_size,
+                        read_costs,
+                        deferred_stripe_handles,
+                        deferred_reopeners,
+                    )
+                    .await;
+                (written, err, false)
+            };
             let decode_elapsed = decode_stage_start.elapsed();
             rustfs_io_metrics::record_get_object_decode_duration(decode_elapsed.as_secs_f64());
             rustfs_io_metrics::record_get_object_stage_duration_by_size(
@@ -1211,6 +1244,9 @@ impl SetDisks {
                 metrics_size_bucket,
                 decode_elapsed.as_secs_f64(),
             );
+            if exact_quorum && err.is_none() {
+                return Err(Error::other("two-phase read completed with exact reconstruction quorum"));
+            }
             if decode_elapsed >= SLOW_OBJECT_READ_LOG_THRESHOLD && err.is_none() {
                 warn!(
                     event = EVENT_SET_DISK_READ,
@@ -1758,6 +1794,102 @@ fn multipart_reader_setup_prefetch_enabled(policy: GetObjectReadPolicy) -> bool
     policy.allows_multipart_setup_prefetch() && is_multipart_reader_setup_prefetch_enabled()
 }
 
+pub(super) struct LateMetadataIdentity {
+    volume: String,
+    name: String,
+    algorithm: String,
+    block_size: usize,
+    uses_legacy_checksum: bool,
+    quorum_hash: [u8; 32],
+    distribution: Vec<usize>,
+    parity_blocks: usize,
+}
+
+impl LateMetadataIdentity {
+    pub(super) fn from_file_info(file_info: &FileInfo) -> Self {
+        Self {
+            volume: file_info.volume.clone(),
+            name: file_info.name.clone(),
+            algorithm: file_info.erasure.algorithm.clone(),
+            block_size: file_info.erasure.block_size,
+            uses_legacy_checksum: file_info.uses_legacy_checksum,
+            quorum_hash: SetDisks::file_info_quorum_hash(file_info),
+            distribution: file_info.erasure.distribution.clone(),
+            parity_blocks: file_info.erasure.parity_blocks,
+        }
+    }
+}
+
+fn late_metadata_read_identity_matches(expected: &LateMetadataIdentity, actual: &FileInfo) -> bool {
+    expected.volume == actual.volume
+        && expected.name == actual.name
+        && expected.algorithm == actual.erasure.algorithm
+        && expected.block_size == actual.erasure.block_size
+        && expected.uses_legacy_checksum == actual.uses_legacy_checksum
+        && expected.quorum_hash == SetDisks::file_info_quorum_hash(actual)
+}
+
+fn late_metadata_shard_matches(expected: &LateMetadataIdentity, actual: &FileInfo, disk_index: usize) -> bool {
+    expected
+        .distribution
+        .get(disk_index)
+        .is_some_and(|mapped_index| *mapped_index == actual.erasure.index)
+        && late_metadata_read_identity_matches(expected, actual)
+}
+
+impl SetDisks {
+    pub(super) async fn refresh_late_metadata_fanout(
+        fallback_disks: &[Option<DiskStore>],
+        bucket: &str,
+        object: &str,
+        expected: &LateMetadataIdentity,
+        metrics_path: &'static str,
+    ) -> Result<(FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>)> {
+        let (mut parts_metadata, errs, diagnostics) = SetDisks::read_all_fileinfo_observed(
+            fallback_disks,
+            "",
+            bucket,
+            object,
+            "",
+            true,
+            false,
+            false,
+            false,
+            expected.parity_blocks,
+        )
+        .await?;
+        diagnostics.record(metrics_path);
+
+        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, write_quorum)?;
+        if !late_metadata_read_identity_matches(expected, &full_fi) {
+            return Err(to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]));
+        }
+
+        for (disk_index, (metadata, disk)) in parts_metadata.iter_mut().zip(online_disks.iter_mut()).enumerate() {
+            if !late_metadata_shard_matches(expected, metadata, disk_index) {
+                *metadata = FileInfo::default();
+                *disk = None;
+            }
+        }
+        if online_disks.iter().filter(|disk| disk.is_some()).count() < read_quorum {
+            return Err(to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]));
+        }
+
+        Ok((full_fi, parts_metadata, online_disks))
+    }
+}
+
 /// Run one part's bitrot reader setup and measure its wall-clock duration.
 ///
 /// Shared by the synchronous path and the prefetch task in
@@ -2349,6 +2481,7 @@ mod metadata_cache_tests {
             0,
             false,
             false,
+            false,
             GET_OBJECT_PATH_SET_DISK,
             "plain",
             "small",
@@ -2380,6 +2513,7 @@ mod metadata_cache_tests {
             0,
             false,
             false,
+            false,
             GET_OBJECT_PATH_SET_DISK,
             "plain",
             "small",
@@ -2404,6 +2538,7 @@ mod metadata_cache_tests {
             0,
             false,
             false,
+            false,
             GET_OBJECT_PATH_SET_DISK,
             "plain",
             "small",
@@ -2426,6 +2561,7 @@ mod metadata_cache_tests {
             0,
             false,
             false,
+            false,
             GET_OBJECT_PATH_SET_DISK,
             "plain",
             "small",
@@ -2450,6 +2586,7 @@ mod metadata_cache_tests {
             0,
             false,
             false,
+            false,
             GET_OBJECT_PATH_SET_DISK,
             "plain",
             "small",
@@ -2488,6 +2625,7 @@ mod metadata_cache_tests {
             0,
             false,
             false,
+            false,
             GET_OBJECT_PATH_SET_DISK,
             "plain",
             "empty",
@@ -2521,6 +2659,7 @@ mod metadata_cache_tests {
             0,
             false,
             false,
+            false,
             GET_OBJECT_PATH_SET_DISK,
             "plain",
             "small",
@@ -4882,6 +5021,7 @@ mod tests {
             0,
             false,
             false,
+            false,
             GET_OBJECT_PATH_SET_DISK,
             "test-object-class",
             "test-size-bucket",