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(
-            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::<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);