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 <heihutu@gmail.com>

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::<usize>();
         self.prefetched_bufs.extend(queued_buffers);
 
         match result {
@@ -912,17 +920,25 @@ mod tests {
         data: &[u8],
         missing_indexes: &[usize],
     ) -> io::Result<Vec<u8>>
+    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<E>(
+        erasure: &Erasure,
+        engine: E,
+        data: &[u8],
+        missing_indexes: &[usize],
+        fill_policy: FillPolicy,
+    ) -> io::Result<Vec<u8>>
     where
         E: ErasureDecodeEngine + Clone + Send + Sync + 'static,
     {
         let source = source_from_data(erasure, data, missing_indexes);
-        let mut reader = ErasureDecodeReader::new_with_fill_policy(
+        let mut reader =
-            source,
+            ErasureDecodeReader::new_with_fill_policy(source, engine, data.len(), GET_OBJECT_PATH_CODEC_STREAMING, fill_policy)?;
-            engine,
-            data.len(),
-            GET_OBJECT_PATH_CODEC_STREAMING,
-            FillPolicy::SingleInFlight,
-        )?;
         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::<Vec<_>>();
+        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::<Vec<_>>();
+        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::<Vec<_>>();
+
+        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);

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<HTTPRangeSpec>,
+    part_number: Option<usize>,
     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,

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 =
+        let codec_streaming_gate = get_codec_streaming_reader_gate(
-            get_codec_streaming_reader_gate(bucket, object, &range, &object_info, &fi, lock_optimization_enabled);
+            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() {

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<HTTPRangeSpec>,
+        part_number: Option<usize>,
+        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(