perf(ecstore): defer pending inline data shards (#6137)

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

@@ -749,6 +749,63 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
     }
 }
 
+fn data_read_inline_missing_shards_are_pending(
+    candidate: &FileInfo,
+    parts_metadata: &[FileInfo],
+    errors: &[Option<DiskError>],
+    disks: &[Option<DiskStore>],
+    fanout_order: &[usize],
+    scheduled_fanout_len: usize,
+) -> bool {
+    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;
+    };
+    let distribution = &candidate.erasure.distribution;
+    let mut data_shards_seen_or_pending = vec![false; erasure.data_shards];
+    let mut missing_pending_data_shards = 0usize;
+
+    for (disk_index, file_info) in parts_metadata.iter().enumerate() {
+        let Some(&block_index) = distribution.get(disk_index) else {
+            return false;
+        };
+        if block_index == 0 || block_index > erasure.data_shards {
+            continue;
+        }
+        if !disks.get(disk_index).is_some_and(Option::is_some) {
+            return false;
+        }
+
+        let data_slot = block_index - 1;
+        if file_info.name.is_empty() {
+            let scheduled_and_not_failed = fanout_order
+                .get(..scheduled_fanout_len)
+                .is_some_and(|scheduled_disks| scheduled_disks.contains(&disk_index))
+                && errors.get(disk_index).is_some_and(Option::is_none);
+            if scheduled_and_not_failed {
+                data_shards_seen_or_pending[data_slot] = true;
+                missing_pending_data_shards = missing_pending_data_shards.saturating_add(1);
+                continue;
+            }
+            return false;
+        }
+        if file_info.erasure.index != block_index
+            || !file_info.has_valid_erasure_geometry()
+            || !metadata_early_stop_candidate_matches(file_info, candidate)
+            || file_info.data.as_ref().is_none_or(|data| data.is_empty())
+        {
+            return false;
+        }
+        data_shards_seen_or_pending[data_slot] = true;
+    }
+
+    missing_pending_data_shards > 0 && data_shards_seen_or_pending.into_iter().all(|seen_or_pending| seen_or_pending)
+}
+
 pub(in crate::set_disk) fn classify_metadata_response_error(err: &DiskError) -> &'static str {
     match err {
         DiskError::FileNotFound | DiskError::VolumeNotFound => GET_METADATA_RESPONSE_NOT_FOUND,
@@ -2532,6 +2589,7 @@ impl SetDisks {
         }
 
         while let Some(result) = join_set.join_next().await {
+            let mut defer_pending_inline_data_shard = false;
             match result {
                 Ok((index, res, elapsed)) => match res {
                     Ok(file_info) => {
@@ -2574,8 +2632,22 @@ impl SetDisks {
                         {
                             None => true,
                             Some(reason) => {
-                                force_full_wait = true;
                                 final_miss_reason_override = Some(reason);
+                                if bounded_fanout
+                                    && reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD
+                                    && data_read_inline_missing_shards_are_pending(
+                                        candidate,
+                                        &ress,
+                                        &errors,
+                                        disks,
+                                        &fanout_order,
+                                        next_fanout_index,
+                                    )
+                                {
+                                    defer_pending_inline_data_shard = true;
+                                } else {
+                                    force_full_wait = true;
+                                }
                                 false
                             }
                         },
@@ -2621,6 +2693,7 @@ impl SetDisks {
             let pending_responses = join_set.len();
             let should_hedge_single_pending_data_read = read_data
                 && !force_full_wait
+                && !defer_pending_inline_data_shard
                 && pending_responses == 1
                 && accumulator.can_still_reach_early_stop_with_pending(pending_responses);
             if bounded_fanout && force_full_wait {
@@ -2633,6 +2706,7 @@ impl SetDisks {
                     next_fanout_index = next_fanout_index.saturating_add(1);
                 }
             } else if bounded_fanout
+                && !defer_pending_inline_data_shard
                 && next_fanout_index < disks.len()
                 && (!accumulator.can_still_reach_early_stop_with_pending(pending_responses)
                     || should_hedge_single_pending_data_read)
@@ -5790,9 +5864,20 @@ mod tests {
         object: &str,
         payload: &[u8],
         uses_legacy_checksum: bool,
+    ) -> Vec<FileInfo> {
+        inline_metadata_fanout_fileinfos_with_geometry(bucket, object, payload, uses_legacy_checksum, 2, 2).await
+    }
+
+    async fn inline_metadata_fanout_fileinfos_with_geometry(
+        bucket: &str,
+        object: &str,
+        payload: &[u8],
+        uses_legacy_checksum: bool,
+        data_shards: usize,
+        parity_shards: usize,
     ) -> Vec<FileInfo> {
         let distribution_key = metadata_distribution_key(bucket, object);
-        let mut base = FileInfo::new(&distribution_key, 2, 2);
+        let mut base = FileInfo::new(&distribution_key, data_shards, parity_shards);
         base.volume = bucket.to_string();
         base.name = object.to_string();
         base.size = i64::try_from(payload.len()).expect("test payload should fit i64");
@@ -5855,6 +5940,21 @@ mod tests {
         install_inline_metadata_fanout_files(disks, bucket, object, files).await;
     }
 
+    async fn install_inline_metadata_fanout_fileinfo_with_geometry(
+        disks: &[Option<DiskStore>],
+        bucket: &str,
+        object: &str,
+        payload: &[u8],
+        data_shards: usize,
+        parity_shards: usize,
+        mutate: impl FnOnce(&mut [FileInfo]),
+    ) {
+        let mut files =
+            inline_metadata_fanout_fileinfos_with_geometry(bucket, object, payload, false, data_shards, parity_shards).await;
+        mutate(&mut files);
+        install_inline_metadata_fanout_files(disks, bucket, object, files).await;
+    }
+
     async fn install_inline_metadata_fanout_files(disks: &[Option<DiskStore>], bucket: &str, object: &str, files: Vec<FileInfo>) {
         let distribution = files
             .first()
@@ -6075,6 +6175,118 @@ mod tests {
         drop(dirs);
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn bounded_metadata_early_stop_waits_for_pending_inline_data_shard() {
+        const DISKS: usize = 6;
+        const DATA_SHARDS: usize = 4;
+        const PARITY_SHARDS: usize = 2;
+        let bucket = "bounded-inline-data-get-pending-shard-bucket";
+        let object =
+            object_with_initial_data_shards(bucket, "bounded-inline-data-get-pending-shard-object", DATA_SHARDS, DATA_SHARDS);
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        install_inline_metadata_fanout_fileinfo_with_geometry(
+            &disks,
+            bucket,
+            &object,
+            b"verified inline payload",
+            DATA_SHARDS,
+            PARITY_SHARDS,
+            |_| {},
+        )
+        .await;
+
+        temp_env::async_with_vars(
+            [
+                ("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", Some("true")),
+            ],
+            async {
+                let fanout_order = bounded_metadata_fanout_order(bucket, &object, DISKS, PARITY_SHARDS);
+                let distribution_key = metadata_distribution_key(bucket, &object);
+                let distribution = FileInfo::new(&distribution_key, DATA_SHARDS, PARITY_SHARDS)
+                    .erasure
+                    .distribution;
+                let paused_data_disk = *fanout_order
+                    .iter()
+                    .take(DATA_SHARDS)
+                    .find(|disk_index| {
+                        distribution
+                            .get(**disk_index)
+                            .is_some_and(|block_index| (1..=DATA_SHARDS).contains(block_index))
+                    })
+                    .expect("initial fanout should include a data shard to pause");
+                let hedged_parity_disk = fanout_order[DATA_SHARDS];
+                let unscheduled_parity_disk = fanout_order[DATA_SHARDS + 1];
+
+                let barrier = rename_fanout_barrier::arm(&object, paused_data_disk, rename_fanout_barrier::PHASE_READ_VERSION);
+                let tracker = rename_fanout_barrier::observe_tasks(&object);
+                let calls = disk_call_counters::observe(&object);
+                let disks_for_read = disks.clone();
+                let object_for_read = object.clone();
+                let mut read = tokio::spawn(async move {
+                    SetDisks::read_all_fileinfo_observed(
+                        &disks_for_read,
+                        bucket,
+                        bucket,
+                        &object_for_read,
+                        "",
+                        true,
+                        false,
+                        false,
+                        true,
+                        PARITY_SHARDS,
+                    )
+                    .await
+                });
+
+                tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
+                    .await
+                    .expect("initial data shard should pause before returning");
+                tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
+                    while calls.for_disk(disk_call_counters::KIND_READ_VERSION, hedged_parity_disk) == 0 {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("bounded fanout should hedge one parity disk while the data shard is pending");
+
+                assert!(
+                    tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await.is_err(),
+                    "inline data-read early-stop must wait for a scheduled missing data shard instead of forcing full wait"
+                );
+
+                barrier.release();
+                let (parts_metadata, errs, diagnostics) = read
+                    .await
+                    .expect("metadata read task should not panic")
+                    .expect("pending data shard should let the inline verifier finish");
+
+                assert_eq!(
+                    calls.total(disk_call_counters::KIND_READ_VERSION),
+                    5,
+                    "pending data-shard defer should not schedule the final parity disk"
+                );
+                assert_eq!(
+                    calls.for_disk(disk_call_counters::KIND_READ_VERSION, unscheduled_parity_disk),
+                    0,
+                    "the remaining parity disk must stay unissued when pending data verification succeeds"
+                );
+                assert_eq!(
+                    tracker.running(),
+                    0,
+                    "early-stop should drain spawned read_version tasks before returning"
+                );
+                assert_eq!(diagnostics.total_responses(), 5);
+                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), 5);
+                assert!(errs.iter().all(Option::is_none));
+            },
+        )
+        .await;
+
+        drop(dirs);
+    }
+
     #[tokio::test]
     async fn data_read_early_stop_verifies_legacy_inline_checksum_payload() {
         let bucket = "legacy-inline-data-get-fanout-bucket";