From 45a2ccb734af886d983e2bd8d51b99eafe4f9aeb Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 31 Aug 2026 11:26:20 +0800 Subject: [PATCH] fix(ecstore): recover late parity after exact quorum (#6927) Co-authored-by: heihutu --- crates/ecstore/src/erasure/coding/decode.rs | 75 +++- .../src/set_disk/core/io_primitives.rs | 52 ++- crates/ecstore/src/set_disk/mod.rs | 333 +++++++++++++++++- crates/ecstore/src/set_disk/ops/object.rs | 159 +++++++++ crates/ecstore/src/set_disk/read.rs | 166 ++++++++- 5 files changed, 740 insertions(+), 45 deletions(-) diff --git a/crates/ecstore/src/erasure/coding/decode.rs b/crates/ecstore/src/erasure/coding/decode.rs index 403b5df33..b0c67c62e 100644 --- a/crates/ecstore/src/erasure/coding/decode.rs +++ b/crates/ecstore/src/erasure/coding/decode.rs @@ -46,6 +46,7 @@ type ShardReadFuture<'a> = Pin = Pin, Error>, Option>, bool)> + Send + 'a>>; pub(crate) type DeferredReaderReopener = Arc Option> + Send + Sync>; +pub(crate) type DecodeOutcome = (usize, Option, 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()) - .await + let (written, error, _) = self + .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()) - .await + let (written, error, _) = self + .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>, deferred_reopeners: Vec>>, ) -> (usize, Option) + 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( + &self, + writer: &mut W, + readers: Vec>>, + offset: usize, + length: usize, + total_length: usize, + read_costs: Option>, + deferred_handles: Vec>, + deferred_reopeners: Vec>>, + ) -> DecodeOutcome where W: AsyncWrite + Send + Sync + Unpin, R: crate::erasure::coding::ShardSource, @@ -2411,36 +2447,48 @@ impl Erasure { read_costs: Option>, deferred_handles: Vec>, deferred_reopeners: Vec>>, - ) -> (usize, Option) + ) -> 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) } } diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 499a47352..282bc926f 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/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, - 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_get_metadata_non_inline_data_read_early_stop_enabled, is_object_dangling, is_version_early_stop_enabled, - issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure, merge_file_meta_versions, path_join_buf, - record_global_dirty_scope, 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, + capacity_scope_from_disks, codec_streaming_rollout_applies, 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_get_metadata_non_inline_data_read_early_stop_enabled, is_object_dangling, + is_version_early_stop_enabled, issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure, + merge_file_meta_versions, object_fits_single_block, path_join_buf, record_global_dirty_scope, 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; @@ -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(())); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index deb8230e3..4fac5362a 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -894,8 +894,11 @@ struct OwnedGetObjectFileInfo { fi: FileInfo, parts_metadata: Vec, online_disks: Vec>, + late_metadata_fanout_disks: Option>>, } +type OwnedGetObjectFileInfoParts = (FileInfo, Vec, Vec>, Option>>); + impl GetObjectFileInfo { fn owned(fi: FileInfo, parts_metadata: Vec, online_disks: Vec>) -> 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, + online_disks: Vec>, + late_metadata_fanout_disks: Vec>, + ) -> 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, Vec>) { + 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) => { - let OwnedGetObjectFileInfo { - fi, - parts_metadata, - online_disks, - } = snapshot; - (fi, parts_metadata, online_disks) - } + (Some(snapshot), None) => ( + snapshot.fi, + snapshot.parts_metadata, + snapshot.online_disks, + snapshot.late_metadata_fanout_disks, + ), (None, Some(entry)) => match Arc::try_unwrap(entry) { - Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks), - Err(entry) => (entry.fi.clone(), entry.parts_metadata.clone(), entry.online_disks.clone()), + Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks, None), + 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, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index fa9aebe71..0cf6b1b3a 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -1379,6 +1379,14 @@ fn data_read_metadata_early_stop_request_shape_allowed(range: &Option, output: &mut Vec, 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; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 32ca821f3..2e8f58ead 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -609,7 +609,20 @@ impl SetDisks { // let online_disks: Vec> = 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 - .decode_with_stripe_handles_and_reopeners( - writer, - readers, - part_offset, - part_length, - part_size, - read_costs, - deferred_stripe_handles, - deferred_reopeners, - ) - .await; + let (written, err, exact_quorum) = if require_reconstruction_surplus { + erasure + .decode_with_stripe_handles_and_reopeners_with_diagnostics( + writer, + readers, + part_offset, + part_length, + part_size, + read_costs, + deferred_stripe_handles, + deferred_reopeners, + ) + .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, + 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], + bucket: &str, + object: &str, + expected: &LateMetadataIdentity, + metrics_path: &'static str, + ) -> Result<(FileInfo, Vec, Vec>)> { + 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",