diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 658f2e4bd..e54f8bb75 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -1721,6 +1721,23 @@ pub(in crate::set_disk) fn should_allow_metadata_early_stop( || (is_version_early_stop_enabled() && !version_id.is_empty() && !healing) } +/// Final gate for the metadata early-stop fast path. +/// +/// `caller_allows_early_stop=false` unconditionally forces the full quorum +/// fanout so read-before-write callers (object tagging) get the complete +/// online-disk set as their write target; the early-stop subset would only +/// carry read quorum and fail write quorum (backlog#872 regression). +pub(in crate::set_disk) fn metadata_early_stop_permitted( + caller_allows_early_stop: bool, + observe: bool, + read_data: bool, + version_id: &str, + healing: bool, + incl_free_versions: bool, +) -> bool { + caller_allows_early_stop && observe && should_allow_metadata_early_stop(read_data, version_id, healing, incl_free_versions) +} + impl SetDisks { pub(in crate::set_disk) async fn read_parts( disks: &[Option], @@ -1797,6 +1814,7 @@ impl SetDisks { healing, incl_free_versions, false, + true, 0, ) .await?; @@ -1813,6 +1831,7 @@ impl SetDisks { read_data: bool, healing: bool, incl_free_versions: bool, + allow_early_stop: bool, default_parity_count: usize, ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { Self::read_all_fileinfo_inner( @@ -1825,6 +1844,7 @@ impl SetDisks { healing, incl_free_versions, true, + allow_early_stop, default_parity_count, ) .await @@ -1841,10 +1861,18 @@ impl SetDisks { healing: bool, incl_free_versions: bool, observe: bool, + // When false, the caller opts out of the early-stop fast path even for + // otherwise-eligible reads. Read-before-write callers (e.g. object + // tagging) must set this so the returned online-disk set reflects the + // full quorum fanout rather than the early-stop subset — writing to the + // subset would fail write quorum (backlog#872 regression). + caller_allows_early_stop: bool, default_parity_count: usize, ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { - let early_stop_enabled = observe && (is_get_metadata_early_stop_enabled() || is_version_early_stop_enabled()); - let allow_early_stop = observe && should_allow_metadata_early_stop(read_data, version_id, healing, incl_free_versions); + let early_stop_enabled = + caller_allows_early_stop && observe && (is_get_metadata_early_stop_enabled() || is_version_early_stop_enabled()); + let allow_early_stop = + metadata_early_stop_permitted(caller_allows_early_stop, observe, read_data, version_id, healing, incl_free_versions); if allow_early_stop { return Self::read_all_fileinfo_early_stop( disks, diff --git a/crates/ecstore/src/set_disk/metadata.rs b/crates/ecstore/src/set_disk/metadata.rs index 3568171a7..4feab05ca 100644 --- a/crates/ecstore/src/set_disk/metadata.rs +++ b/crates/ecstore/src/set_disk/metadata.rs @@ -839,6 +839,55 @@ impl SetDisks { Err(DiskError::ErasureReadQuorum) } + /// Ownership-taking variant of `shuffle_disks_and_parts_metadata_by_index` + /// (backlog#873): callers that already own the vectors avoid one deep + /// `FileInfo` clone per disk by moving entries into their shuffled slots. + /// + /// Semantics match the borrowing variant, including the fallback to the + /// mod-time based placement when `parity_blocks` or more sources are + /// inconsistent; the consistency check runs as a read-only first pass so + /// the fallback still sees the untouched inputs. + pub(super) fn shuffle_disks_and_parts_metadata_by_index_owned( + mut disks: Vec>, + mut parts_metadata: Vec, + fi: &FileInfo, + ) -> (Vec>, Vec) { + let distribution = &fi.erasure.distribution; + + let mut inconsistent = 0; + for (k, v) in parts_metadata.iter().enumerate() { + if disks[k].is_none() || !v.is_valid() || distribution[k] != v.erasure.index { + inconsistent += 1; + } + } + + let use_by_index = inconsistent < fi.erasure.parity_blocks; + let init = fi.mod_time.is_none(); + + let mut shuffled_disks = vec![None; disks.len()]; + let mut shuffled_parts_metadata = vec![FileInfo::default(); parts_metadata.len()]; + + for k in 0..parts_metadata.len() { + if disks[k].is_none() { + continue; + } + let eligible = if use_by_index { + parts_metadata[k].is_valid() && distribution[k] == parts_metadata[k].erasure.index + } else { + init || parts_metadata[k].is_valid() + }; + if !eligible { + continue; + } + + let block_idx = distribution[k]; + shuffled_parts_metadata[block_idx - 1] = std::mem::take(&mut parts_metadata[k]); + shuffled_disks[block_idx - 1] = disks[k].take(); + } + + (shuffled_disks, shuffled_parts_metadata) + } + pub(super) fn shuffle_disks_and_parts_metadata_by_index( disks: &[Option], parts_metadata: &[FileInfo], @@ -949,3 +998,78 @@ impl SetDisks { shuffled_parts_errs } } + +#[cfg(test)] +mod tests { + use super::*; + + async fn shuffle_test_disks(tempdir: &tempfile::TempDir, count: usize) -> Vec> { + let endpoint = + Endpoint::try_from(tempdir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("disk should be created"); + // The shuffle only inspects Some/None and clones the Arc handle, so + // one shared disk handle per slot is sufficient. + (0..count).map(|_| Some(disk.clone())).collect() + } + + fn shuffle_fixture(consistent: bool) -> (FileInfo, Vec) { + let mut fi = FileInfo::new("bucket/object", 2, 1); + fi.mod_time = Some(time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")); + fi.size = 1; + fi.add_object_part(1, String::new(), 1, None, 1, None, None); + + let slots = fi.erasure.distribution.len(); + let parts = (0..slots) + .map(|k| { + let mut part_fi = fi.clone(); + part_fi.erasure.index = if consistent { + fi.erasure.distribution[k] + } else { + // Misplace every source so the by-index pass is rejected + // and the mod-time fallback placement runs instead. + fi.erasure.distribution[(k + 1) % slots] + }; + part_fi + }) + .collect(); + (fi, parts) + } + + #[tokio::test] + async fn owned_shuffle_matches_borrowing_variant_when_consistent() { + let tempdir = tempfile::tempdir().expect("tempdir should be created"); + let (fi, parts) = shuffle_fixture(true); + let disks = shuffle_test_disks(&tempdir, parts.len()).await; + + let (expected_disks, expected_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index(&disks, &parts, &fi); + let (owned_disks, owned_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index_owned(disks, parts, &fi); + + assert_eq!(owned_parts, expected_parts, "owned shuffle must place identical metadata"); + let expected_slots: Vec = expected_disks.iter().map(Option::is_some).collect(); + let owned_slots: Vec = owned_disks.iter().map(Option::is_some).collect(); + assert_eq!(owned_slots, expected_slots, "owned shuffle must fill identical disk slots"); + } + + #[tokio::test] + async fn owned_shuffle_matches_borrowing_variant_on_fallback() { + let tempdir = tempfile::tempdir().expect("tempdir should be created"); + let (fi, parts) = shuffle_fixture(false); + let disks = shuffle_test_disks(&tempdir, parts.len()).await; + + let (expected_disks, expected_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index(&disks, &parts, &fi); + let (owned_disks, owned_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index_owned(disks, parts, &fi); + + assert_eq!(owned_parts, expected_parts, "fallback placement must match the borrowing variant"); + let expected_slots: Vec = expected_disks.iter().map(Option::is_some).collect(); + let owned_slots: Vec = owned_disks.iter().map(Option::is_some).collect(); + assert_eq!(owned_slots, expected_slots, "fallback disk slots must match the borrowing variant"); + } +} diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 69135fd9f..caa55136f 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -459,7 +459,12 @@ const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: usize = 128 * 102 // --- Metadata Early-Stop Configuration --- const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE"; -const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = false; +// Enabled by default (backlog#872): the early-stop path only engages for +// requests `should_allow_metadata_early_stop` classifies as safe (metadata-only +// reads without version_id / healing / free-version needs) and still requires +// a full read-quorum agreement before stopping. Set the env var to `false` to +// fall back to full-wait metadata fanout. +const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true; const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT"; const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100; @@ -467,6 +472,11 @@ const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100; const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE"; const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false; +// --- Multipart Reader-Setup Prefetch Configuration (backlog#870) --- + +const ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH: &str = "RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH"; +const DEFAULT_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH: bool = true; + static OBJECT_LOCK_DIAG_ENABLED: OnceLock = OnceLock::new(); mod core; @@ -682,6 +692,31 @@ fn is_version_early_stop_enabled() -> bool { } } +/// Check if multipart reads prefetch the next part's bitrot reader setup +/// while the current part decodes (backlog#870). +/// +/// **Note**: Cached via `OnceLock` in production. In test builds the env var +/// is read directly so that `temp_env` overrides take effect. +fn is_multipart_reader_setup_prefetch_enabled() -> bool { + #[cfg(test)] + { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH, + DEFAULT_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH, + ) + } + #[cfg(not(test))] + { + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH, + DEFAULT_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH, + ) + }) + } +} + // --- Rollout Percentage Functions --- fn get_codec_streaming_rollout_pct() -> u32 { @@ -6899,6 +6934,119 @@ mod tests { assert_eq!(out, payload[range_offset..range_offset + range_length]); } + #[tokio::test] + async fn multipart_reads_stream_all_parts_with_setup_prefetch() { + use tokio::io::AsyncReadExt; + use uuid::Uuid; + + let tempdir = tempfile::tempdir().expect("tempdir should be created"); + let endpoint = + Endpoint::try_from(tempdir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("disk should be created"); + + let bucket = "bucket"; + let object = "object"; + // Three parts with distinct fill bytes so cross-part ordering bugs and + // prefetch boundary mistakes surface as content mismatches + // (backlog#870 exercises the prefetch hit path for parts 2 and 3). + let parts: Vec> = vec![ + vec![b'a'; 2 * 1024 * 1024 + 111], + vec![b'b'; 1024 * 1024 + 17], + vec![b'c'; 3 * 1024 * 1024 + 923], + ]; + let total_size: usize = parts.iter().map(|part| part.len()).sum(); + + disk.make_volume(bucket).await.expect("bucket should be created"); + + let mut fi = FileInfo::new(&format!("{bucket}/{object}"), 1, 0); + let data_dir = Uuid::new_v4(); + fi.data_dir = Some(data_dir); + fi.size = total_size as i64; + for (index, part) in parts.iter().enumerate() { + fi.add_object_part(index + 1, String::new(), part.len(), None, part.len() as i64, None, None); + } + + let erasure = coding::Erasure::new_with_options( + fi.erasure.data_blocks, + fi.erasure.parity_blocks, + fi.erasure.block_size, + fi.uses_legacy_checksum, + ); + + for (index, payload) in parts.iter().enumerate() { + let part_number = index + 1; + let shard_path = format!("{object}/{data_dir}/part.{part_number}"); + let checksum_info = fi.erasure.get_checksum_info(part_number); + + let mut bitrot_writer = create_bitrot_writer( + true, + None, + bucket, + &shard_path, + payload.len() as i64, + erasure.shard_size(), + checksum_info.algorithm.clone(), + ) + .await + .expect("bitrot writer should be created"); + + for chunk in payload.chunks(erasure.shard_size()) { + bitrot_writer.write(chunk).await.expect("payload chunk should be written"); + } + + let encoded = bitrot_writer.into_inline_data().expect("bitrot encoded data should exist"); + disk.write_all(bucket, &shard_path, Bytes::from(encoded)) + .await + .expect("encoded shard should be stored"); + } + + let files = vec![fi.clone()]; + let disks = vec![Some(disk.clone())]; + let (mut reader, mut writer) = tokio::io::duplex(64 * 1024); + let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(fi.size); + + let read_task = tokio::spawn(async move { + SetDisks::get_object_with_fileinfo( + bucket, + object, + 0, + total_size as i64, + &mut writer, + fi, + files, + &disks, + 0, + 0, + true, + false, + GET_OBJECT_PATH_LEGACY_DUPLEX, + GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, + metrics_size_bucket, + ) + .await + }); + + let mut out = Vec::new(); + reader.read_to_end(&mut out).await.expect("all part bytes should be readable"); + + read_task + .await + .expect("read task should complete") + .expect("multipart read should succeed"); + + let expected: Vec = parts.concat(); + assert_eq!(out.len(), expected.len(), "all parts should be streamed"); + assert_eq!(out, expected, "part contents and ordering must survive setup prefetch"); + } + #[test] fn parts_after_marker_uses_marker_position() { let part_numbers = (1..=1002).collect::>(); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index e1059d730..a8931f114 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -2171,7 +2171,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { // } // _lock_guard = guard_opt; // } - let (mut fi, _, disks) = self.get_object_fileinfo(bucket, object, opts, false).await?; + // Force the full quorum fanout (allow_early_stop=false): `disks` is the + // write target below, and an early-stop subset would only carry read + // quorum, failing write quorum on update_object_meta (backlog#872). + let (mut fi, _, disks) = self.get_object_fileinfo_gated(bucket, object, opts, false, false).await?; fi.metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_owned()); diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 60640ae05..af0a58538 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -167,6 +167,24 @@ impl SetDisks { object: &str, opts: &ObjectOptions, read_data: bool, + ) -> Result<(FileInfo, Vec, Vec>)> { + // Read-only callers (GET/HEAD/tag read) may use the metadata early-stop + // fast path. + self.get_object_fileinfo_gated(bucket, object, opts, read_data, true).await + } + + /// Like `get_object_fileinfo`, but `allow_early_stop=false` forces the full + /// quorum fanout. Read-before-write callers (object tagging) must use this: + /// the returned online-disk set is the write target, and the early-stop + /// subset would fail write quorum (backlog#872 regression fix). + #[tracing::instrument(level = "debug", skip(self))] + pub(in crate::set_disk) async fn get_object_fileinfo_gated( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + read_data: bool, + allow_early_stop: bool, ) -> Result<(FileInfo, Vec, Vec>)> { let vid = opts.version_id.clone().unwrap_or_default(); let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); @@ -221,7 +239,10 @@ impl SetDisks { let disks = disks.clone(); - // TODO: optimize concurrency and break once enough slots are available + // Early-stop for safe metadata reads is handled inside + // read_all_fileinfo_observed (see read_all_fileinfo_early_stop in + // core/io_primitives.rs); unsafe requests and callers that opt out + // (allow_early_stop=false) fall back to full-wait. let (parts_metadata, errs, metadata_fanout_diagnostics) = Self::read_all_fileinfo_observed( &disks, "", @@ -231,6 +252,7 @@ impl SetDisks { read_data, false, opts.incl_free_versions, + allow_early_stop, self.default_parity_count, ) .await?; @@ -516,7 +538,10 @@ impl SetDisks { { let pipeline_started = Instant::now(); debug!(bucket, object, requested_length = length, offset, "get_object_with_fileinfo start"); - let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi); + // Owned shuffle (backlog#873): `files` is consumed and its FileInfo + // entries move into their shuffled slots, avoiding one deep clone per + // disk; the disk handles are Arc clones and stay cheap. + let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index_owned(disks.to_vec(), files, &fi); let total_size = fi.size as usize; @@ -607,6 +632,17 @@ impl SetDisks { let part_indices: Vec = (part_index..=last_part_index).collect(); debug!(bucket, object, ?part_indices, "Multipart part indices to stream"); + // Pipeline prefetch (backlog#870): while the current part decodes and + // streams out, the next part's bitrot reader setup (file opens + + // read-quorum wait across all disks) runs concurrently, so multipart + // reads no longer serialize setup latency between parts. Depth is one + // part; shared inputs move behind Arc so the prefetch task is 'static. + let use_mmap_read = object_mmap_read_enabled(); + let files = Arc::new(files); + let disks = Arc::new(disks); + let prefetch_enabled = is_multipart_reader_setup_prefetch_enabled(); + let mut prefetched: Option<(usize, PrefetchedReaderSetup)> = None; + let mut total_read = 0; for current_part in part_indices { if total_read == length { @@ -647,43 +683,98 @@ impl SetDisks { "Streaming multipart part" ); - let checksum_info = fi.erasure.get_checksum_info(part_number); - let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { - HashAlgorithm::HighwayHash256SLegacy - } else { - checksum_info.algorithm - }; + let checksum_algo = multipart_part_checksum_algo(&fi, part_number); let read_length = till_offset.saturating_sub(read_offset); - let use_mmap_read = object_mmap_read_enabled(); - - let reader_setup_stage_start = Instant::now(); let read_costs = coding::decode::should_collect_shard_read_costs().then(|| shard_read_costs_for_disks(&disks)); - let reader_setup = create_bitrot_readers_until_quorum_with_preference( - &files, - &disks, - bucket, - object, + let sync_spec = PartReaderSetupSpec { part_number, read_offset, read_length, - erasure.shard_size(), checksum_algo, - skip_verify_bitrot, - use_mmap_read, - erasure.data_shards, - erasure.parity_shards, - BitrotReaderSetupMode::ReadQuorum, - prefer_data_blocks_first_reader_setup, - None, - Some(BitrotReaderSetupAttribution { - path: metrics_path, - object_class: metrics_object_class, - size_bucket: metrics_size_bucket, - }), - ) - .await; - let reader_setup_elapsed = reader_setup_stage_start.elapsed(); + }; + let mut setup_result = None; + // A stale prefetch (part mismatch) is dropped by the failed let + // chain, which aborts its task via the guard. + if let Some((prefetched_part, handle)) = prefetched.take() + && prefetched_part == current_part + { + setup_result = handle.join().await; + if setup_result.is_none() { + warn!( + bucket, + object, + part_index = current_part, + "Multipart reader-setup prefetch task did not complete; retrying synchronously" + ); + } + } + let (reader_setup, reader_setup_elapsed) = match setup_result { + Some(result) => result, + None => { + setup_multipart_part_readers( + &files, + &disks, + bucket, + object, + sync_spec, + erasure.shard_size(), + erasure.data_shards, + erasure.parity_shards, + skip_verify_bitrot, + use_mmap_read, + prefer_data_blocks_first_reader_setup, + metrics_path, + metrics_object_class, + metrics_size_bucket, + ) + .await + } + }; + + // Kick off the next part's reader setup before decoding this one + // so the disk opens overlap with decode + client writeback + // (backlog#870). + let remaining_after_current = length - total_read - part_length; + if prefetch_enabled && remaining_after_current > 0 && current_part < last_part_index { + let next_part = current_part + 1; + let next_number = fi.parts[next_part].number; + let next_size = fi.parts[next_part].size; + let next_length = next_size.min(remaining_after_current); + let spec = PartReaderSetupSpec { + part_number: next_number, + read_offset: 0, + read_length: erasure.shard_file_offset(0, next_length, next_size), + checksum_algo: multipart_part_checksum_algo(&fi, next_number), + }; + let files = Arc::clone(&files); + let disks = Arc::clone(&disks); + let bucket = bucket.to_owned(); + let object = object.to_owned(); + let shard_size = erasure.shard_size(); + let data_shards = erasure.data_shards; + let parity_shards = erasure.parity_shards; + let handle = tokio::task::spawn(async move { + setup_multipart_part_readers( + &files, + &disks, + &bucket, + &object, + spec, + shard_size, + data_shards, + parity_shards, + skip_verify_bitrot, + use_mmap_read, + prefer_data_blocks_first_reader_setup, + metrics_path, + metrics_object_class, + metrics_size_bucket, + ) + .await + }); + prefetched = Some((next_part, PrefetchedReaderSetup::new(handle))); + } rustfs_io_metrics::record_get_object_shard_reader_setup_duration(reader_setup_elapsed.as_secs_f64()); rustfs_io_metrics::record_get_object_stage_duration_by_size( metrics_path, @@ -1022,36 +1113,78 @@ impl SetDisks { return Err(Error::other("codec streaming multipart part sizes do not match object size")); } - let mut readers = Vec::with_capacity(fi.parts.len()); - for part in &fi.parts { - match Self::build_codec_streaming_part_reader( - bucket, - object, - fi, - &files, - &disks, - &erasure, - part.number, - 0, - part.size, - part.size, - skip_verify_bitrot, - metrics_object_class, - metrics_size_bucket, - false, - ) - .await? - { - GetCodecStreamingReaderBuildOutcome::Reader(reader) => readers.push(reader), - GetCodecStreamingReaderBuildOutcome::Fallback(reason) => { - return Ok(GetCodecStreamingReaderBuildOutcome::Fallback(reason)); - } + // Lazy multipart construction (backlog#871): only the first part's + // shard readers are opened before streaming starts, so TTFB no longer + // pays for `parts x disks` file opens and an early client disconnect + // never touches the remaining parts. The first part stays eager so the + // dominant fallback conditions (missing shards, read quorum) are still + // detected before any byte is streamed and the whole request can fall + // back to the legacy duplex path. + let first_part = &fi.parts[0]; + let first_reader = match Self::build_codec_streaming_part_reader( + bucket, + object, + fi, + &files, + &disks, + &erasure, + first_part.number, + 0, + first_part.size, + first_part.size, + skip_verify_bitrot, + metrics_object_class, + metrics_size_bucket, + false, + ) + .await? + { + GetCodecStreamingReaderBuildOutcome::Reader(reader) => reader, + GetCodecStreamingReaderBuildOutcome::Fallback(reason) => { + return Ok(GetCodecStreamingReaderBuildOutcome::Fallback(reason)); } - } + }; - Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(MultipartCodecStreamingReader::new( - readers, - )))) + let remaining_parts: Vec<(usize, usize)> = fi.parts[1..].iter().map(|part| (part.number, part.size)).collect(); + let total_parts = fi.parts.len(); + let ctx = Arc::new(LazyCodecPartContext { + bucket: bucket.to_owned(), + object: object.to_owned(), + fi: fi.clone(), + files, + disks, + erasure, + skip_verify_bitrot, + metrics_object_class, + metrics_size_bucket, + }); + let builder: LazyPartBuilder = Box::new(move |remaining_index| { + let ctx = Arc::clone(&ctx); + let (part_number, part_size) = remaining_parts[remaining_index]; + tokio::task::spawn(async move { + SetDisks::build_codec_streaming_part_reader( + &ctx.bucket, + &ctx.object, + &ctx.fi, + &ctx.files, + &ctx.disks, + &ctx.erasure, + part_number, + 0, + part_size, + part_size, + ctx.skip_verify_bitrot, + ctx.metrics_object_class, + ctx.metrics_size_bucket, + false, + ) + .await + }) + }); + + Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new( + LazyMultipartCodecStreamingReader::new(first_reader, total_parts, builder, get_codec_streaming_metrics_path()), + ))) } #[allow(clippy::too_many_arguments)] @@ -1165,6 +1298,243 @@ impl SetDisks { } } +/// Per-part parameters for a multipart bitrot reader setup. +struct PartReaderSetupSpec { + part_number: usize, + read_offset: usize, + read_length: usize, + checksum_algo: HashAlgorithm, +} + +/// Resolve the bitrot checksum algorithm for one part, honoring the legacy +/// HighwayHash flag. +fn multipart_part_checksum_algo(fi: &FileInfo, part_number: usize) -> HashAlgorithm { + let checksum_info = fi.erasure.get_checksum_info(part_number); + if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { + HashAlgorithm::HighwayHash256SLegacy + } else { + checksum_info.algorithm + } +} + +/// Run one part's bitrot reader setup and measure its wall-clock duration. +/// +/// Shared by the synchronous path and the prefetch task in +/// `get_object_with_fileinfo` (backlog#870) so both report the same +/// stage-duration semantics. +#[allow(clippy::too_many_arguments)] +async fn setup_multipart_part_readers( + files: &[FileInfo], + disks: &[Option], + bucket: &str, + object: &str, + spec: PartReaderSetupSpec, + shard_size: usize, + data_shards: usize, + parity_shards: usize, + skip_verify_bitrot: bool, + use_mmap_read: bool, + prefer_data_blocks_first: bool, + metrics_path: &'static str, + metrics_object_class: &'static str, + metrics_size_bucket: &'static str, +) -> (BitrotReaderSetup, Duration) { + let started = Instant::now(); + let setup = create_bitrot_readers_until_quorum_with_preference( + files, + disks, + bucket, + object, + spec.part_number, + spec.read_offset, + spec.read_length, + shard_size, + spec.checksum_algo, + skip_verify_bitrot, + use_mmap_read, + data_shards, + parity_shards, + BitrotReaderSetupMode::ReadQuorum, + prefer_data_blocks_first, + None, + Some(BitrotReaderSetupAttribution { + path: metrics_path, + object_class: metrics_object_class, + size_bucket: metrics_size_bucket, + }), + ) + .await; + (setup, started.elapsed()) +} + +/// Guard around an in-flight prefetch of the next part's reader setup +/// (backlog#870). Dropping the guard without consuming it aborts the task so +/// error returns and early breaks stop the background disk IO. +struct PrefetchedReaderSetup(Option>); + +impl PrefetchedReaderSetup { + fn new(handle: tokio::task::JoinHandle<(BitrotReaderSetup, Duration)>) -> Self { + Self(Some(handle)) + } + + /// Wait for the prefetch to finish; `None` means the task was cancelled + /// or panicked and the caller must set up synchronously. + async fn join(mut self) -> Option<(BitrotReaderSetup, Duration)> { + let handle = self.0.take()?; + handle.await.ok() + } +} + +impl Drop for PrefetchedReaderSetup { + fn drop(&mut self) { + if let Some(handle) = self.0.take() { + handle.abort(); + } + } +} + +/// Owned context for lazily constructing codec streaming part readers after +/// the first part has started streaming (backlog#871). +struct LazyCodecPartContext { + bucket: String, + object: String, + fi: FileInfo, + files: Vec, + disks: Vec>, + erasure: coding::Erasure, + skip_verify_bitrot: bool, + metrics_object_class: &'static str, + metrics_size_bucket: &'static str, +} + +type LazyPartBuildHandle = tokio::task::JoinHandle>; +type LazyPartBuilder = Box LazyPartBuildHandle + Send + Sync>; + +/// Multipart codec streaming reader that constructs part readers on demand. +/// +/// The first part reader is built eagerly by the caller so the dominant +/// fallback conditions are detected before any byte is streamed; every +/// subsequent part is only built once the previous part reaches EOF. If a +/// later part hits a fallback condition mid-stream, the reader surfaces a read +/// error instead: data has already been streamed, so switching the whole +/// request to the legacy path is no longer possible. The next request detects +/// the condition on its eager first-part setup and falls back cleanly. +struct LazyMultipartCodecStreamingReader { + current: Option>, + pending: Option, + dispatched_remaining: usize, + total_parts: usize, + builder: LazyPartBuilder, + metrics_path: &'static str, +} + +impl LazyMultipartCodecStreamingReader { + fn new( + first_reader: Box, + total_parts: usize, + builder: LazyPartBuilder, + metrics_path: &'static str, + ) -> Self { + Self { + current: Some(first_reader), + pending: None, + dispatched_remaining: 0, + total_parts, + builder, + metrics_path, + } + } +} + +impl AsyncRead for LazyMultipartCodecStreamingReader { + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + if buf.remaining() == 0 { + return Poll::Ready(Ok(())); + } + + let this = self.get_mut(); + loop { + if let Some(reader) = this.current.as_mut() { + let filled_before = buf.filled().len(); + match Pin::new(reader).poll_read(cx, buf) { + Poll::Ready(Ok(())) if buf.filled().len() == filled_before => { + // Part EOF: drop its shard readers before building the + // next part. + this.current = None; + } + result => return result, + } + continue; + } + + if let Some(handle) = this.pending.as_mut() { + match Pin::new(handle).poll(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(join_result) => { + this.pending = None; + match join_result { + Ok(Ok(GetCodecStreamingReaderBuildOutcome::Reader(reader))) => { + this.current = Some(reader); + } + Ok(Ok(GetCodecStreamingReaderBuildOutcome::Fallback(reason))) => { + // KNOWN LIMITATION (backlog#871 follow-up): once + // earlier parts have streamed we can no longer + // hand the whole request back to the legacy + // duplex path, so a degraded later part surfaces + // as a read error even though legacy per-part + // decode could still reconstruct it. This only + // affects the OPT-IN multipart codec streaming + // path (RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE, + // default off): the first part stays eager so the + // common case where part 1 is already degraded + // still falls back cleanly before any byte ships. + // A proper fix (in-place per-part legacy + // degradation) is tracked as a follow-up. + record_get_object_pipeline_failure_for_path( + this.metrics_path, + GET_STAGE_READER_SETUP, + GetObjectFailureReason::ReadQuorum, + ); + warn!( + metrics_path = this.metrics_path, + fallback_reason = ?reason, + state = "codec_streaming_mid_stream_fallback", + "Lazy multipart part construction hit a fallback condition mid-stream; surfacing read error" + ); + return Poll::Ready(Err(std::io::Error::other(format!( + "codec streaming multipart reader cannot fall back mid-stream: {reason:?}" + )))); + } + Ok(Err(err)) => return Poll::Ready(Err(std::io::Error::other(err))), + Err(join_err) => return Poll::Ready(Err(std::io::Error::other(join_err))), + } + } + } + continue; + } + + if this.dispatched_remaining + 1 < this.total_parts { + let handle = (this.builder)(this.dispatched_remaining); + this.dispatched_remaining += 1; + this.pending = Some(handle); + continue; + } + + return Poll::Ready(Ok(())); + } + } +} + +impl Drop for LazyMultipartCodecStreamingReader { + fn drop(&mut self) { + // Abort an in-flight part construction so an early client disconnect + // does not keep opening shard readers in the background. + if let Some(handle) = self.pending.take() { + handle.abort(); + } + } +} + fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOptions, read_data: bool) -> Option<&'static str> { if !read_data { return Some(GET_METADATA_CACHE_REASON_NOT_READ_DATA); @@ -2145,12 +2515,47 @@ mod tests { } #[test] - fn metadata_early_stop_gate_defaults_to_disabled() { + fn metadata_early_stop_gate_defaults_to_enabled() { temp_env::with_var(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, None::<&str>, || { - assert!(!is_get_metadata_early_stop_enabled()); + assert!(is_get_metadata_early_stop_enabled()); }); } + #[test] + fn metadata_early_stop_gate_honors_explicit_opt_out() { + temp_env::with_var(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("false"), || { + assert!(!is_get_metadata_early_stop_enabled()); + // With the gate off, even safe metadata-only requests must fall + // back to the full-wait fanout. + assert!(!should_allow_metadata_early_stop(false, "", false, false)); + }); + } + + #[test] + fn metadata_early_stop_permitted_respects_caller_opt_out() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE, Some("true")), + ], + || { + // Read-before-write callers (object tagging) pass + // caller_allows_early_stop=false and must never early-stop, even + // for an otherwise-eligible safe metadata-only read. The + // early-stop subset would fail write quorum (backlog#872). + assert!(!metadata_early_stop_permitted(false, true, false, "", false, false)); + assert!(!metadata_early_stop_permitted(false, true, false, "version-id", false, false)); + // With the caller allowing it and every other condition safe, + // the fast path is permitted. + assert!(metadata_early_stop_permitted(true, true, false, "", false, false)); + // observe=false (non-observed fanout) also disables early-stop. + assert!(!metadata_early_stop_permitted(true, false, false, "", false, false)); + // Data reads are never eligible regardless of caller opt-in. + assert!(!metadata_early_stop_permitted(true, true, true, "", false, false)); + }, + ); + } + #[test] fn metadata_early_stop_rejects_data_reads() { temp_env::with_vars( @@ -2458,6 +2863,86 @@ mod tests { assert_eq!(drops.load(Ordering::SeqCst), 2); } + fn lazy_test_builder( + parts: Vec<&'static [u8]>, + builds: Arc, + ) -> Box LazyPartBuildHandle + Send + Sync> { + Box::new(move |remaining_index| { + builds.fetch_add(1, Ordering::SeqCst); + let data = parts[remaining_index]; + tokio::task::spawn( + async move { Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(Cursor::new(data.to_vec())))) }, + ) + }) + } + + #[tokio::test] + async fn lazy_multipart_codec_streaming_reader_reads_parts_in_order() { + let builds = Arc::new(AtomicUsize::new(0)); + let builder = lazy_test_builder(vec![b"multi", b"part"], Arc::clone(&builds)); + let mut reader = + LazyMultipartCodecStreamingReader::new(Box::new(Cursor::new(b"hello ".to_vec())), 3, builder, "codec_streaming"); + + let mut output = Vec::new(); + reader + .read_to_end(&mut output) + .await + .expect("lazy multipart reader should stream all parts"); + + assert_eq!(output, b"hello multipart"); + assert_eq!(builds.load(Ordering::SeqCst), 2, "both remaining parts should be built exactly once"); + } + + #[tokio::test] + async fn lazy_multipart_codec_streaming_reader_defers_construction_until_needed() { + let builds = Arc::new(AtomicUsize::new(0)); + let builder = lazy_test_builder(vec![b"never"], Arc::clone(&builds)); + { + let mut reader = + LazyMultipartCodecStreamingReader::new(Box::new(Cursor::new(b"abcdef".to_vec())), 2, builder, "codec_streaming"); + + let mut first = [0u8; 6]; + reader + .read_exact(&mut first) + .await + .expect("first part should satisfy the read"); + assert_eq!(&first, b"abcdef"); + } + + assert_eq!( + builds.load(Ordering::SeqCst), + 0, + "dropping the reader before crossing the part boundary must not build later parts" + ); + } + + #[tokio::test] + async fn lazy_multipart_codec_streaming_reader_surfaces_mid_stream_fallback_as_error() { + let builds = Arc::new(AtomicUsize::new(0)); + let builds_clone = Arc::clone(&builds); + let builder: Box LazyPartBuildHandle + Send + Sync> = Box::new(move |_| { + builds_clone.fetch_add(1, Ordering::SeqCst); + tokio::task::spawn(async move { + Ok(GetCodecStreamingReaderBuildOutcome::Fallback(GetCodecStreamingFallbackReason::Multipart)) + }) + }); + let mut reader = + LazyMultipartCodecStreamingReader::new(Box::new(Cursor::new(b"first".to_vec())), 2, builder, "codec_streaming"); + + let mut output = Vec::new(); + let err = reader + .read_to_end(&mut output) + .await + .expect_err("mid-stream fallback must surface as a read error"); + + assert_eq!(output, b"first", "bytes streamed before the fallback stay intact"); + assert!( + err.to_string().contains("cannot fall back mid-stream"), + "error should explain the mid-stream fallback: {err}" + ); + assert_eq!(builds.load(Ordering::SeqCst), 1); + } + fn inline_reader_setup_fileinfo(data: Option<&'static [u8]>) -> FileInfo { let mut fi = FileInfo::new("object", 2, 2); fi.volume = "bucket".to_string(); diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 49ce6cac1..f97db9eef 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -39,14 +39,15 @@ use futures::future::join_all; use rand::seq::SliceRandom; use rustfs_filemeta::{ FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntriesSortedResult, MetaCacheEntry, - MetacacheReader, MetadataResolutionParams, is_io_eof, merge_file_meta_versions, + MetacacheReader, MetadataResolutionParams, is_io_eof, }; use rustfs_io_metrics::{ LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED, LIST_OBJECTS_GATHER_OUTCOME_LIMIT_REACHED, LIST_OBJECTS_SOURCE_WALKER, ListObjectsGatherObservation, ListObjectsIndexPageObservation, }; use rustfs_utils::path::{self, SLASH_SEPARATOR, base_dir_from_prefix}; -use std::collections::{HashMap, HashSet}; +use std::cmp::Reverse; +use std::collections::{BinaryHeap, HashMap, HashSet}; use std::future::Future; use std::path::{Path, PathBuf}; use std::sync::{ @@ -4376,7 +4377,7 @@ impl ECStore { continue; } - let fvs = match if opts.include_free_versions { + let mut fvs = match if opts.include_free_versions { entry.file_info_versions_with_free_versions(&bucket_clone) } else { entry.file_info_versions(&bucket_clone) @@ -4395,8 +4396,13 @@ impl ECStore { } }; + // `FileMeta` keeps versions newest-first (`sort_by_mod_time` + // sorts descending), so ascending emission is the exact + // reverse. Callers such as replication resync walk with the + // default (ascending) order so that re-applied versions + // preserve the original version-stack order. if opts.versions_sort == WalkVersionsSortOrder::Ascending { - //TODO: SORT + fvs.versions.reverse(); } for fi in fvs.versions.iter() { @@ -4639,28 +4645,64 @@ async fn gather_results( Ok(GatherResultsState::InputClosed) } -async fn select_from( - rx: &CancellationToken, - in_channels: &mut [Receiver], +/// Head entry of one input channel inside the k-way merge. +/// +/// `cleaned` caches `path::clean(name)` only when it differs from the raw +/// name, so heap comparisons never allocate (walker-produced names are +/// already clean in the common case). +struct MergeHead { + entry: MetaCacheEntry, + cleaned: Option, idx: usize, - top: &mut [Option], - n_done: &mut usize, -) -> Result { - let entry = tokio::select! { - entry = in_channels[idx].recv() => entry, - _ = rx.cancelled() => return Ok(false), - }; +} - match entry { - Some(entry) => { - top[idx] = Some(entry); - } - None => { - top[idx] = None; - *n_done += 1; - } +impl MergeHead { + fn new(entry: MetaCacheEntry, idx: usize) -> Self { + let cleaned = path::clean(&entry.name); + let cleaned = if cleaned == entry.name { None } else { Some(cleaned) }; + Self { entry, cleaned, idx } + } + + fn sort_key(&self) -> &str { + self.cleaned.as_deref().unwrap_or(&self.entry.name) + } +} + +impl PartialEq for MergeHead { + fn eq(&self, other: &Self) -> bool { + self.cmp(other) == std::cmp::Ordering::Equal + } +} + +impl Eq for MergeHead {} + +impl PartialOrd for MergeHead { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for MergeHead { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + // Ascending channel index preserves the legacy scan order for equal keys. + self.sort_key().cmp(other.sort_key()).then_with(|| self.idx.cmp(&other.idx)) + } +} + +enum MergePoll { + Entry(Box), + Closed, + Cancelled, +} + +async fn poll_merge_head(rx: &CancellationToken, in_channels: &mut [Receiver], idx: usize) -> MergePoll { + tokio::select! { + entry = in_channels[idx].recv() => match entry { + Some(entry) => MergePoll::Entry(Box::new(MergeHead::new(entry, idx))), + None => MergePoll::Closed, + }, + _ = rx.cancelled() => MergePoll::Cancelled, } - Ok(true) } async fn send_or_cancel(rx: &CancellationToken, out_channel: &Sender, entry: MetaCacheEntry) -> Result { @@ -4707,147 +4749,100 @@ async fn merge_entry_channels( } } - let mut top: Vec> = vec![None; in_channels.len()]; - let mut n_done = 0; - - let in_channels_len = in_channels.len(); - - for idx in 0..in_channels_len { - if !select_from(&rx, &mut in_channels, idx, &mut top, &mut n_done).await? { - return Ok(()); + // K-way merge across per-channel sorted streams via a min-heap keyed by + // the path-cleaned entry name: advancing the merge costs O(log channels) + // per entry and comparisons are allocation-free (see `MergeHead`). + // + // Note on same-name resolution: prefix-dir groups collapse into a single + // emission and objects shadow same-named prefix dirs. The legacy + // `merge_file_meta_versions` block that used to live here was dead code: + // it only ran for prefix-dir groups, whose entries have empty metadata, + // so `xl_meta()` always failed. Cross-drive version merging happens in + // the metadata resolve path (`MetaCacheEntries::resolve`), not here. + let mut heap: BinaryHeap>> = BinaryHeap::with_capacity(in_channels.len()); + for idx in 0..in_channels.len() { + match poll_merge_head(&rx, &mut in_channels, idx).await { + MergePoll::Entry(head) => heap.push(Reverse(head)), + MergePoll::Closed => {} + MergePoll::Cancelled => { + info!("merge_entry_channels rx.recv() cancel"); + return Ok(()); + } } } let mut last = String::new(); - let mut to_merge: Vec = Vec::new(); - loop { - if n_done == in_channels.len() { + let mut group: Vec> = Vec::new(); + let mut refill: Vec = Vec::with_capacity(in_channels.len()); + + while let Some(Reverse(first)) = heap.pop() { + // Collect every channel head that resolves to the same cleaned key so + // the duplicate rules below see the whole same-name group at once. + group.clear(); + refill.clear(); + refill.push(first.idx); + while heap.peek().is_some_and(|Reverse(next)| next.sort_key() == first.sort_key()) { + let Some(Reverse(next)) = heap.pop() else { break }; + refill.push(next.idx); + group.push(next); + } + + // Legacy same-name resolution rules (heads arrive in ascending + // channel order): + // - prefix dir vs prefix dir with the same suffix shape: the first + // head wins and the rest collapse; + // - prefix dir vs object: the object wins; + // - object vs object: the later channel wins; + // - both dirs with different suffix shape: the smaller raw name wins. + let mut winner = first; + for other in group.drain(..) { + let dir_matches = winner.entry.is_dir() && other.entry.is_dir(); + let suffix_matches = winner.entry.name.ends_with(SLASH_SEPARATOR) == other.entry.name.ends_with(SLASH_SEPARATOR); + + if dir_matches && suffix_matches { + continue; + } + + if !dir_matches { + if other.entry.is_dir() { + continue; + } + winner = other; + continue; + } + + if winner.entry.name > other.entry.name { + winner = other; + } + } + + // Gate emission on the same cleaned key the heap orders by, not the raw + // name. The heap pops in non-decreasing cleaned order, so comparing the + // raw name here could drop a legitimate entry whose cleaned order and + // raw order disagree (e.g. redundant slashes or `./` segments). + let emit = winner.sort_key() > last.as_str(); + if emit { + last.clear(); + last.push_str(winner.sort_key()); + } + let MergeHead { entry, .. } = *winner; + if emit && !send_or_cancel(&rx, &out_channel, entry).await? { return Ok(()); } - let mut best = top[0].clone(); - let mut best_idx = 0; - to_merge.clear(); - - // Note: `select_from` mutates `top[idx]` during the inner loop, but this is safe - // because each borrow from `top[other_idx]` is only used before any later - // `select_from` call that can mutate that slot. - - for other_idx in 1..top.len() { - if let Some(other_entry) = &top[other_idx] { - if let Some(best_entry) = &best { - if path::clean(&best_entry.name) == path::clean(&other_entry.name) { - let dir_matches = best_entry.is_dir() && other_entry.is_dir(); - let suffix_matches = - best_entry.name.ends_with(SLASH_SEPARATOR) == other_entry.name.ends_with(SLASH_SEPARATOR); - - if dir_matches && suffix_matches { - to_merge.push(other_idx); - continue; - } - - if !dir_matches { - // dir and object has the save name - if other_entry.is_dir() { - if !select_from(&rx, &mut in_channels, other_idx, &mut top, &mut n_done).await? { - return Ok(()); - } - continue; - } - - to_merge.clear(); - - best = Some(other_entry.clone()); - best_idx = other_idx; - continue; - } - } - - if best_entry.name > other_entry.name { - to_merge.clear(); - best = Some(other_entry.clone()); - best_idx = other_idx; - } - } else { - best = Some(other_entry.clone()); - best_idx = other_idx; + for &idx in &refill { + match poll_merge_head(&rx, &mut in_channels, idx).await { + MergePoll::Entry(head) => heap.push(Reverse(head)), + MergePoll::Closed => {} + MergePoll::Cancelled => { + info!("merge_entry_channels rx.recv() cancel"); + return Ok(()); } } } - - if !to_merge.is_empty() { - if let Some(entry) = &best { - let mut versions = Vec::with_capacity(to_merge.len() + 1); - - let mut has_xl = { entry.clone().xl_meta().ok() }; - - if let Some(x) = &has_xl { - versions.push(x.versions.clone()); - } - - for &idx in to_merge.iter() { - let has_entry = top[idx].clone(); - - if let Some(entry) = has_entry { - let xl2 = match entry.clone().xl_meta() { - Ok(res) => res, - Err(_) => { - if !select_from(&rx, &mut in_channels, idx, &mut top, &mut n_done).await? { - return Ok(()); - } - - continue; - } - }; - - versions.push(xl2.versions.clone()); - - if has_xl.is_none() { - if !select_from(&rx, &mut in_channels, best_idx, &mut top, &mut n_done).await? { - return Ok(()); - } - - best_idx = idx; - best = Some(entry.clone()); - has_xl = Some(xl2); - } else { - if !select_from(&rx, &mut in_channels, best_idx, &mut top, &mut n_done).await? { - return Ok(()); - } - } - } - } - - if let Some(xl) = has_xl.as_mut() - && !versions.is_empty() - { - xl.versions = merge_file_meta_versions(read_quorum, true, 0, &versions); - - if let Ok(meta) = xl.marshal_msg() - && let Some(b) = best.as_mut() - { - b.metadata = meta; - b.cached = Some(xl.clone()); - } - } - } - - to_merge.clear(); - } - - if let Some(best_entry) = &best - && best_entry.name > last - { - if !send_or_cancel(&rx, &out_channel, best_entry.clone()).await? { - return Ok(()); - } - last = best_entry.name.clone(); - } - - if !select_from(&rx, &mut in_channels, best_idx, &mut top, &mut n_done).await? { - return Ok(()); - } } + + Ok(()) } impl Sets { @@ -9230,6 +9225,178 @@ mod test { assert_eq!(results, vec!["obj-a", "obj-b"]); } + #[test] + fn walk_ascending_versions_contract_reverses_newest_first_metadata() { + // Documents the invariant the walk `versions_sort` handling relies on: + // `FileMeta` keeps versions newest-first, so ascending emission is the + // exact reverse of `file_info_versions` output. + let older = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let newer = time::OffsetDateTime::from_unix_timestamp(1_705_312_500).expect("valid timestamp"); + let entry = + test_object_meta_entry_with_erasure_versions("obj-a", &[(older, "etag-old", 4, 2), (newer, "etag-new", 4, 2)]); + + let fvs = entry.file_info_versions("bucket").expect("versions should parse"); + assert_eq!(fvs.versions.len(), 2, "both versions should be visible"); + assert!( + fvs.versions[0].mod_time >= fvs.versions[1].mod_time, + "file_info_versions must yield newest-first" + ); + assert!(fvs.versions[0].is_latest, "the first (newest) version carries is_latest"); + + let mut ascending = fvs.versions; + ascending.reverse(); + assert!( + ascending[0].mod_time <= ascending[1].mod_time, + "reversing newest-first output must produce ascending mod_time order" + ); + assert!( + ascending.last().expect("versions present").is_latest, + "ascending emission ends with the latest version" + ); + } + + #[tokio::test] + async fn merge_entry_channels_merges_interleaved_channels_in_global_order() { + let mut txs = Vec::new(); + let mut rxs = Vec::new(); + for _ in 0..4 { + let (tx, rx) = mpsc::channel(16); + txs.push(tx); + rxs.push(rx); + } + + // Round-robin keys so every channel stays sorted while the global + // stream interleaves across all four channels. + let names: Vec = (0..24).map(|i| format!("obj-{i:02}")).collect(); + for (i, name) in names.iter().enumerate() { + txs[i % 4].send(test_meta_entry(name)).await.unwrap(); + } + // The same trailing key on three channels must be emitted once. + for tx in txs.iter().take(3) { + tx.send(test_meta_entry("obj-99")).await.unwrap(); + } + drop(txs); + + let (out_tx, mut out_rx) = mpsc::channel(64); + let cancel = CancellationToken::new(); + let handle = tokio::spawn(merge_entry_channels(cancel, rxs, out_tx, 1)); + + let mut results = Vec::new(); + while let Some(entry) = out_rx.recv().await { + results.push(entry.name.clone()); + } + handle.await.unwrap().unwrap(); + + let mut expected = names; + expected.push("obj-99".to_owned()); + assert_eq!(results, expected); + } + + #[tokio::test] + async fn merge_entry_channels_emits_when_cleaned_order_differs_from_raw_order() { + // Regression: `a//c` cleans to `a/c` and `a/b` is already clean. In + // cleaned order `a/b` < `a/c`, but raw byte order is the opposite + // (`a//c` < `a/b` because '/' < 'b'). The heap pops in cleaned order, + // so gating emission on the raw name would drop `a//c` after `a/b` is + // emitted. Both distinct keys must survive. + let (tx_a, rx_a) = mpsc::channel(4); + let (tx_b, rx_b) = mpsc::channel(4); + let (out_tx, mut out_rx) = mpsc::channel(8); + + tx_a.send(test_meta_entry("a/b")).await.unwrap(); + drop(tx_a); + tx_b.send(test_meta_entry("a//c")).await.unwrap(); + drop(tx_b); + + let cancel = CancellationToken::new(); + let handle = tokio::spawn(merge_entry_channels(cancel, vec![rx_a, rx_b], out_tx, 1)); + + let mut results = Vec::new(); + while let Some(entry) = out_rx.recv().await { + results.push(entry.name.clone()); + } + handle.await.unwrap().unwrap(); + + assert_eq!( + results, + vec!["a/b", "a//c"], + "distinct cleaned keys must both be emitted in cleaned order" + ); + } + + #[tokio::test] + async fn merge_entry_channels_object_wins_over_same_named_prefix_dir() { + let (tx_a, rx_a) = mpsc::channel(4); + let (tx_b, rx_b) = mpsc::channel(4); + let (out_tx, mut out_rx) = mpsc::channel(8); + + // `path::clean("a/") == "a"`, so the prefix dir and the object group + // under the same key and the object must win. + tx_a.send(test_dir_meta_entry("a/")).await.unwrap(); + drop(tx_a); + tx_b.send(test_object_meta_entry("a")).await.unwrap(); + drop(tx_b); + + let cancel = CancellationToken::new(); + let handle = tokio::spawn(merge_entry_channels(cancel, vec![rx_a, rx_b], out_tx, 1)); + + let mut results = Vec::new(); + while let Some(entry) = out_rx.recv().await { + assert!(entry.is_object(), "the object entry must shadow the same-named prefix dir"); + results.push(entry.name.clone()); + } + handle.await.unwrap().unwrap(); + + assert_eq!(results, vec!["a"]); + } + + #[tokio::test] + async fn merge_entry_channels_collapses_equivalent_prefix_dirs() { + let (tx_a, rx_a) = mpsc::channel(4); + let (tx_b, rx_b) = mpsc::channel(4); + let (out_tx, mut out_rx) = mpsc::channel(8); + + tx_a.send(test_dir_meta_entry("x/")).await.unwrap(); + drop(tx_a); + tx_b.send(test_dir_meta_entry("x/")).await.unwrap(); + drop(tx_b); + + let cancel = CancellationToken::new(); + let handle = tokio::spawn(merge_entry_channels(cancel, vec![rx_a, rx_b], out_tx, 1)); + + let mut results = Vec::new(); + while let Some(entry) = out_rx.recv().await { + results.push(entry.name.clone()); + } + handle.await.unwrap().unwrap(); + + assert_eq!(results, vec!["x/"]); + } + + #[tokio::test] + async fn merge_entry_channels_normalizes_uncleaned_names_before_grouping() { + let (tx_a, rx_a) = mpsc::channel(4); + let (tx_b, rx_b) = mpsc::channel(4); + let (out_tx, mut out_rx) = mpsc::channel(8); + + // Both names clean to "a/b"; they must merge into a single emission. + tx_a.send(test_meta_entry("a//b")).await.unwrap(); + drop(tx_a); + tx_b.send(test_meta_entry("a/b")).await.unwrap(); + drop(tx_b); + + let cancel = CancellationToken::new(); + let handle = tokio::spawn(merge_entry_channels(cancel, vec![rx_a, rx_b], out_tx, 1)); + + let mut results = Vec::new(); + while let Some(entry) = out_rx.recv().await { + results.push(entry.name.clone()); + } + handle.await.unwrap().unwrap(); + + assert_eq!(results, vec!["a/b"]); + } + #[tokio::test] async fn merge_entry_channels_respects_cancellation() { let (tx, rx) = mpsc::channel::(4);