perf(ecstore): gate quorum-aware GET early stop (#6885)

* perf(ecstore): add gated two-phase GET metadata reads

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): require data-shard coverage for read plans

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): avoid inline overhead in read plan rollout

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): accept quorum-complete read candidates

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

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

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

@@ -49,10 +49,11 @@ use super::super::{
     capacity_scope_from_disks, coding, collect_inline_data_shard_fileinfos_by_index_or_reason, current_dirty_generation, debug,
     disk, file_info_is_valid_for_metadata, get_metadata_slowtail_fault_request, info, inline_erasure_shard_file_offset,
     inline_erasure_shard_size, is_err_object_not_found, is_err_version_not_found, is_get_metadata_data_read_early_stop_enabled,
-    is_get_metadata_early_stop_bounded_fanout_enabled, is_get_metadata_early_stop_enabled, is_object_dangling,
+    is_get_metadata_early_stop_bounded_fanout_enabled, is_get_metadata_early_stop_enabled,
-    is_version_early_stop_enabled, issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure,
+    is_get_metadata_two_phase_read_plan_enabled, is_object_dangling, is_version_early_stop_enabled, issue3031_diag_enabled,
-    merge_file_meta_versions, path_join_buf, record_global_dirty_scope, reduce_read_quorum_errs, reduce_write_quorum_errs,
+    join_all, join_errs, log_multipart_write_quorum_failure, merge_file_meta_versions, path_join_buf, record_global_dirty_scope,
-    send_heal_request_with_admission, should_prevent_write, to_object_err, try_read_inline_data_shards_direct, warn,
+    reduce_read_quorum_errs, 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;
@@ -1083,6 +1084,50 @@ fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Op
     None
 }
 
+fn non_inline_data_read_candidate_is_safe(
+    candidate: &FileInfo,
+    parts_metadata: &[FileInfo],
+    disks: &[Option<DiskStore>],
+) -> bool {
+    if candidate.inline_data()
+        || candidate.is_compressed()
+        || candidate.is_remote()
+        || candidate
+            .metadata
+            .keys()
+            .any(|key| rustfs_utils::http::is_object_encryption_marker(key))
+        || candidate.parts.len() != 1
+    {
+        return false;
+    }
+    let Ok(erasure) = coding::Erasure::try_new_with_options(
+        candidate.erasure.data_blocks,
+        candidate.erasure.parity_blocks,
+        candidate.erasure.block_size,
+        candidate.uses_legacy_checksum,
+    ) else {
+        return false;
+    };
+    // The regular reader setup can reconstruct missing data shards from any
+    // `data_shards` matching metadata entries. Requiring every data slot here
+    // would unnecessarily wait for one slow data disk even when parity and
+    // the remaining data shards already form a read quorum.
+    let mut available_shards = vec![false; erasure.data_shards + erasure.parity_shards];
+    for ((file_info, disk), &erasure_index) in parts_metadata
+        .iter()
+        .zip(disks.iter())
+        .zip(candidate.erasure.distribution.iter())
+    {
+        if erasure_index == 0 || erasure_index > available_shards.len() || disk.is_none() {
+            continue;
+        }
+        if metadata_early_stop_candidate_matches(file_info, candidate) && file_info.erasure.index == erasure_index {
+            available_shards[erasure_index - 1] = true;
+        }
+    }
+    available_shards.into_iter().filter(|present| *present).count() >= erasure.data_shards
+}
+
 fn data_read_inline_missing_shards_are_pending(
     candidate: &FileInfo,
     parts_metadata: &[FileInfo],
@@ -2869,6 +2914,7 @@ impl SetDisks {
                 read_data,
                 healing,
                 incl_free_versions,
+                read_data && is_get_metadata_two_phase_read_plan_enabled(),
                 default_parity_count,
                 allow_coalescing,
             )
@@ -3008,6 +3054,7 @@ impl SetDisks {
         read_data: bool,
         healing: bool,
         incl_free_versions: bool,
+        allow_non_inline_data_read_early_stop: bool,
         default_parity_count: usize,
         allow_coalescing: bool,
     ) -> disk::error::Result<(Vec<FileInfo>, Vec<Option<DiskError>>, MetadataFanoutDiagnostics)> {
@@ -3096,6 +3143,8 @@ impl SetDisks {
                             && read_data
                             && !force_full_wait
                             && let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
+                            && !(allow_non_inline_data_read_early_stop
+                                && reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE)
                         {
                             force_full_wait = true;
                             final_miss_reason_override.get_or_insert(reason);
@@ -3126,6 +3175,12 @@ impl SetDisks {
             {
                 let should_return_early = if read_data {
                     match accumulator.candidate.as_ref() {
+                        Some(candidate)
+                            if allow_non_inline_data_read_early_stop
+                                && non_inline_data_read_candidate_is_safe(candidate, &ress, disks) =>
+                        {
+                            true
+                        }
                         Some(candidate) => match data_read_early_stop_inline_body_miss_reason(
                             bucket.as_ref(),
                             object.as_ref(),

crates/ecstore/src/set_disk/mod.rs

@@ -773,6 +773,13 @@ const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false;
 const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE";
 const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
 
+// Two-phase metadata/read-plan rollout (backlog#1309). The first phase reads
+// metadata without inline payloads and only fetches inline data from the
+// selected data-shard slots. Keep this opt-in until the Linux multi-node
+// slow-tail and small-inline cost gates are complete.
+const ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE: &str = "RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE";
+const DEFAULT_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE: bool = false;
+
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
 const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true;
 
@@ -1168,6 +1175,109 @@ mod prepared_get_object_metadata_tests {
         );
     }
 
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn two_phase_read_plan_uses_metadata_only_for_non_inline_get() {
+        let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+        let bucket = "two-phase-read-plan";
+        let object = object_with_initial_data_shards(bucket, "non-inline-object");
+        let payload = vec![0x5a; 2 * 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");
+
+        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);
+                reset_test_get_object_reader_path();
+                let mut reader = set_disks
+                    .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
+                    .await
+                    .expect("two-phase GET reader should open");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("two-phase GET body should stream");
+                assert_eq!(restored, payload);
+                assert!(
+                    calls.total(disk_call_counters::KIND_READ_VERSION) < 4,
+                    "non-inline two-phase GET should stop metadata fanout at a quorum"
+                );
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn two_phase_read_plan_preserves_inline_early_stop_path() {
+        let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+        let bucket = "two-phase-read-plan-inline";
+        let object = object_with_initial_data_shards(bucket, "inline-object");
+        let payload = b"two-phase inline payload".repeat(256);
+        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("inline object should be 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 inline GET reader should open");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("two-phase inline GET body should stream");
+                assert_eq!(restored, payload);
+                assert_eq!(
+                    test_get_object_reader_path_id(),
+                    3,
+                    "inline GET should retain the direct inline reader path"
+                );
+                assert!(calls.total(disk_call_counters::KIND_READ_VERSION) <= 4);
+            },
+        )
+        .await;
+    }
+
     #[test]
     #[serial_test::serial(body_cache_hook)]
     fn inline_data_read_early_stop_defaults_return_exact_body() {
@@ -1914,6 +2024,26 @@ fn is_get_metadata_data_read_early_stop_enabled() -> bool {
     }
 }
 
+fn is_get_metadata_two_phase_read_plan_enabled() -> bool {
+    #[cfg(test)]
+    {
+        rustfs_utils::get_env_bool(
+            ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE,
+            DEFAULT_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE,
+        )
+    }
+    #[cfg(not(test))]
+    {
+        static CACHED: OnceLock<bool> = OnceLock::new();
+        *CACHED.get_or_init(|| {
+            rustfs_utils::get_env_bool(
+                ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE,
+                DEFAULT_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE,
+            )
+        })
+    }
+}
+
 fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool {
     #[cfg(test)]
     {

crates/ecstore/src/set_disk/read.rs

@@ -89,6 +89,8 @@ use super::ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT;
 #[cfg(test)]
 use super::ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE;
 #[cfg(test)]
+use super::ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE;
+#[cfg(test)]
 use super::ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE;
 #[cfg(test)]
 use super::ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH;
@@ -126,6 +128,8 @@ use super::is_get_metadata_early_stop_bounded_fanout_enabled;
 #[cfg(test)]
 use super::is_get_metadata_early_stop_enabled;
 #[cfg(test)]
+use super::is_get_metadata_two_phase_read_plan_enabled;
+#[cfg(test)]
 use super::is_version_early_stop_enabled;
 #[cfg(test)]
 use super::load_get_codec_streaming_config;
@@ -4267,6 +4271,19 @@ mod tests {
         );
     }
 
+    #[test]
+    fn two_phase_read_plan_gate_defaults_off_and_honors_override() {
+        temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, None::<&str>, || {
+            assert!(!is_get_metadata_two_phase_read_plan_enabled());
+        });
+        temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true"), || {
+            assert!(is_get_metadata_two_phase_read_plan_enabled());
+        });
+        temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("false"), || {
+            assert!(!is_get_metadata_two_phase_read_plan_enabled());
+        });
+    }
+
     #[test]
     fn metadata_early_stop_rejects_healing_and_free_version_requests() {
         temp_env::with_vars(