From 15808254d32daaaaf27a03d8c249b731fbf951b3 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 9 Jul 2026 01:35:39 +0800 Subject: [PATCH] fix(ecstore): correct codec-streaming byte accounting and partNumber routing (#4535) Two correctness defects on the opt-in codec-streaming GET path. ECA-02 (#943): ErasureDecodeReader only decremented `remaining` for the main fill buffer. Under the default DualInFlight policy each fill also produces a queued stripe that is delivered to the client via `prefetched_bufs.pop_front()` without touching `remaining`, so any object larger than one erasure block finished with `remaining > 0` and the GET terminated with LessData despite delivering all bytes. The inflated `remaining` was also fed back into the fill worker, which used it to trim the final stripe and to decide whether to read past EOF. Account for the queued-stripe bytes when they enter the prefetch queue; queued buffers come only from `Ok(true)` decodes so they are non-empty and bounded by `remaining - main_buf.len()`, ruling out underflow. ECA-04 (#945): the codec-streaming gate did not inspect `opts.part_number`. A partNumber GET carries `range == None`, so it was not classified as a Range request and reached the full-object codec-streaming reader, which drops the storage offset/length returned by GetObjectReader::new. A partNumber >= 2 request would then stream the whole object. Mirror the direct-memory part_number fallback and route any partNumber request back to the legacy duplex path, which applies the offset/length correctly. Regression tests: DualInFlight read_to_end on a multi-block object and on a non-block-aligned object; SingleInFlight vs DualInFlight byte-identical output; gate fallback on partNumber requests. Co-authored-by: heihutu --- .../src/erasure/coding/decode_reader.rs | 94 +++++++++++++++++-- crates/ecstore/src/set_disk/mod.rs | 17 ++++ crates/ecstore/src/set_disk/ops/object.rs | 11 ++- crates/ecstore/src/set_disk/read.rs | 61 ++++++++++++ 4 files changed, 174 insertions(+), 9 deletions(-) diff --git a/crates/ecstore/src/erasure/coding/decode_reader.rs b/crates/ecstore/src/erasure/coding/decode_reader.rs index d9f58c9bf..68f06ac0d 100644 --- a/crates/ecstore/src/erasure/coding/decode_reader.rs +++ b/crates/ecstore/src/erasure/coding/decode_reader.rs @@ -298,6 +298,14 @@ where if let Some(deferred_error) = deferred_error { self.prefetch_error = Some(deferred_error); } + // Queued stripes are delivered to the client via `prefetched_bufs.pop_front()` + // in `poll_read` without touching `self.remaining`, so their bytes must be + // accounted for here. Otherwise, under multi-in-flight fill policies (e.g. the + // default `DualInFlight`), `remaining` never reaches 0 and a fully delivered + // multi-block object still terminates the GET with `LessData`. Buffers in + // `queued_buffers` come only from `Ok(true)` decodes, so they are non-empty and + // their total is bounded by `remaining - main_buf.len()`, ruling out underflow. + self.remaining -= queued_buffers.iter().map(Vec::len).sum::(); self.prefetched_bufs.extend(queued_buffers); match result { @@ -912,17 +920,25 @@ mod tests { data: &[u8], missing_indexes: &[usize], ) -> io::Result> + where + E: ErasureDecodeEngine + Clone + Send + Sync + 'static, + { + decode_all_with_engine_and_policy(erasure, engine, data, missing_indexes, FillPolicy::SingleInFlight).await + } + + async fn decode_all_with_engine_and_policy( + erasure: &Erasure, + engine: E, + data: &[u8], + missing_indexes: &[usize], + fill_policy: FillPolicy, + ) -> io::Result> where E: ErasureDecodeEngine + Clone + Send + Sync + 'static, { let source = source_from_data(erasure, data, missing_indexes); - let mut reader = ErasureDecodeReader::new_with_fill_policy( - source, - engine, - data.len(), - GET_OBJECT_PATH_CODEC_STREAMING, - FillPolicy::SingleInFlight, - )?; + let mut reader = + ErasureDecodeReader::new_with_fill_policy(source, engine, data.len(), GET_OBJECT_PATH_CODEC_STREAMING, fill_policy)?; let mut decoded = Vec::new(); reader.read_to_end(&mut decoded).await?; Ok(decoded) @@ -1109,6 +1125,70 @@ mod tests { .expect("dual inflight policy should prefetch two future stripes before current output drains"); } + #[tokio::test] + async fn erasure_decode_reader_dual_inflight_reads_multi_block_object_to_end() { + // Regression for the byte-accounting bug: under DualInFlight each fill + // delivers a main stripe plus a queued stripe, but only the main stripe + // used to decrement `remaining`. A fully delivered multi-block object then + // terminated with LessData. With three 32-byte stripes (96 bytes total) + // the queued-stripe bytes must be accounted for so read_to_end succeeds. + let erasure = Erasure::new(4, 2, 32); + let data = (0..96u8).collect::>(); + let engine = LegacyEcDecodeEngine::new(erasure.clone()); + let decoded = decode_all_with_engine_and_policy(&erasure, engine, &data, &[], FillPolicy::DualInFlight) + .await + .expect("dual inflight read_to_end should succeed for a multi-block object"); + assert_eq!(decoded, data); + } + + #[tokio::test] + async fn erasure_decode_reader_dual_inflight_trims_non_block_aligned_tail() { + // Regression companion: a non-block-aligned object (100 bytes, block_size + // 32) must decode to exactly its real length under DualInFlight with no + // trailing erasure padding. An inflated `remaining` would both fail with + // LessData and let the final partial stripe emit padded output. + let erasure = Erasure::new(4, 2, 32); + let data = (0..100u8).collect::>(); + let engine = LegacyEcDecodeEngine::new(erasure.clone()); + let decoded = decode_all_with_engine_and_policy(&erasure, engine, &data, &[], FillPolicy::DualInFlight) + .await + .expect("dual inflight read_to_end should succeed for a non-aligned object"); + assert_eq!(decoded.len(), 100); + assert_eq!(decoded, data); + } + + #[tokio::test] + async fn erasure_decode_reader_fill_policies_produce_identical_output() { + // Policy-parameterized comparison: SingleInFlight and DualInFlight must + // decode the same multi-stripe object to byte-identical output. This locks + // the byte-accounting fix against future policy changes. + let erasure = Erasure::new(4, 2, 32); + let data = (0..96u8).collect::>(); + + let single = decode_all_with_engine_and_policy( + &erasure, + LegacyEcDecodeEngine::new(erasure.clone()), + &data, + &[], + FillPolicy::SingleInFlight, + ) + .await + .expect("single inflight decode should succeed"); + + let dual = decode_all_with_engine_and_policy( + &erasure, + LegacyEcDecodeEngine::new(erasure.clone()), + &data, + &[], + FillPolicy::DualInFlight, + ) + .await + .expect("dual inflight decode should succeed"); + + assert_eq!(single, data); + assert_eq!(single, dual); + } + #[tokio::test] async fn erasure_decode_reader_defers_short_read_error_until_buffer_drains() { let erasure = Erasure::new(4, 2, 32); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 506307977..1624950ef 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -974,6 +974,7 @@ enum GetCodecStreamingFallbackReason { HeaderCompatibilityUnconfirmed, LockOptimizationDisabled, Range, + PartNumber, BelowMinSize, Encrypted, Compressed, @@ -994,6 +995,7 @@ impl GetCodecStreamingFallbackReason { Self::HeaderCompatibilityUnconfirmed => "header_compatibility_unconfirmed", Self::LockOptimizationDisabled => "lock_optimization_disabled", Self::Range => "range", + Self::PartNumber => "part_number", Self::BelowMinSize => "below_min_size", Self::Encrypted => "encrypted", Self::Compressed => "compressed", @@ -1298,6 +1300,7 @@ fn get_codec_streaming_reader_gate( bucket: &str, object: &str, range: &Option, + part_number: Option, object_info: &ObjectInfo, fi: &FileInfo, lock_optimization_enabled: bool, @@ -1346,6 +1349,20 @@ fn get_codec_streaming_reader_gate( prefer_data_blocks_first_reader_setup: false, }; } + // A partNumber GET arrives with `range == None`, so it is not caught by the + // Range class above, yet it still requires a non-zero storage offset/length + // (synthesized from the part size). The codec-streaming path builds a + // full-object reader and drops the offset/length returned by + // `GetObjectReader::new`, so partNumber >= 2 would stream the whole object. + // Mirror the direct-memory part_number fallback and route these requests back + // to the legacy duplex path, which applies the offset/length correctly. + if part_number.is_some() { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::PartNumber), + prefer_data_blocks_first_reader_setup: false, + }; + } if !lock_optimization_enabled { return GetCodecStreamingGate { object_class, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 40018a115..e0e7cbcb7 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -299,8 +299,15 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } let path_decision_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); - let codec_streaming_gate = - get_codec_streaming_reader_gate(bucket, object, &range, &object_info, &fi, lock_optimization_enabled); + let codec_streaming_gate = get_codec_streaming_reader_gate( + bucket, + object, + &range, + opts.part_number, + &object_info, + &fi, + lock_optimization_enabled, + ); record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_PATH_DECISION, path_decision_stage_start); if object_info.is_remote() { diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index bb2d89d1b..66a6e2bf7 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -2719,6 +2719,25 @@ mod tests { CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, range, + None, + object_info, + fi, + lock_optimization_enabled, + ) + } + + fn codec_streaming_reader_gate_for_test_with_part_number( + range: &Option, + part_number: Option, + object_info: &ObjectInfo, + fi: &FileInfo, + lock_optimization_enabled: bool, + ) -> GetCodecStreamingGate { + get_codec_streaming_reader_gate( + CODEC_STREAMING_TEST_BUCKET, + CODEC_STREAMING_TEST_OBJECT, + range, + part_number, object_info, fi, lock_optimization_enabled, @@ -4080,6 +4099,48 @@ mod tests { ); } + #[test] + fn codec_streaming_reader_gate_falls_back_on_part_number_request() { + // A partNumber GET has `range == None`, so it is classified as a plain + // object and would otherwise reach the codec-streaming path. That path + // builds a full-object reader and drops the storage offset/length, so + // partNumber >= 2 would stream the whole object. The gate must route any + // partNumber request back to the legacy duplex path via Fallback. + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("100")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + ], + || { + let fi = codec_streaming_test_fileinfo(1024, 1); + let object_info = codec_streaming_test_object_info(&fi); + + // Baseline: without a partNumber the gate uses codec streaming. + assert_eq!( + codec_streaming_reader_gate_for_test_with_part_number(&None, None, &object_info, &fi, true).decision, + GetCodecStreamingDecision::Use + ); + + // partNumber >= 2 must fall back to legacy duplex. + assert_eq!( + codec_streaming_reader_gate_for_test_with_part_number(&None, Some(2), &object_info, &fi, true).decision, + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::PartNumber) + ); + + // partNumber == 1 also falls back even though its offset is 0, so the + // legacy path stays the single owner of part offset application. + assert_eq!( + codec_streaming_reader_gate_for_test_with_part_number(&None, Some(1), &object_info, &fi, true).decision, + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::PartNumber) + ); + }, + ); + } + #[tokio::test] async fn codec_streaming_reader_build_falls_back_when_read_quorum_is_not_safe() { let setup = setup_inline_bitrot_readers(