fix(ecstore): reject zero erasure block_size in codec streaming read (#4340)

The codec streaming GET reader divides by erasure.block_size in
build_codec_streaming_part_reader without validating the erasure
dimensions, unlike the legacy multipart path which already rejects
block_size==0 / data_shards==0. FileInfo::is_valid() does not check
block_size, so corrupted on-disk metadata (block_size==0, data_blocks>0)
passes validation and panics the read task with a divide-by-zero.

Add Erasure::has_valid_dimensions() and reject invalid dimensions at the
codec streaming entry before any disk access, mirroring the legacy guard
(which now reuses the same predicate).

Refs backlog#868 (868-1).

crates/ecstore/src/erasure/coding/erasure.rs

@@ -787,6 +787,17 @@ impl Erasure {
     pub fn total_shard_count(&self) -> usize {
         self.data_shards + self.parity_shards
     }
+
+    /// Whether the erasure dimensions are safe for the shard/offset arithmetic.
+    ///
+    /// `block_size` and `data_shards` come straight from on-disk metadata; a
+    /// corrupted or crafted `xl.meta` can carry zero values that would panic the
+    /// `block_size`/`data_shards` divisions in [`Self::shard_size`],
+    /// [`Self::shard_file_size`] and [`Self::shard_file_offset`]. Read paths must
+    /// reject such metadata before performing those divisions.
+    pub fn has_valid_dimensions(&self) -> bool {
+        self.block_size > 0 && self.data_shards > 0
+    }
     // /// Calculate the shard size and total size for a given data size.
     // // Returns (shard_size, total_size) for the given data size
     // fn need_size(&self, data_size: usize) -> (usize, usize) {
@@ -962,6 +973,21 @@ mod tests {
         assert_eq!(owned, borrowed);
     }
 
+    #[test]
+    fn has_valid_dimensions_rejects_zero_block_size_or_data_shards() {
+        // Well-formed erasure metadata is accepted.
+        assert!(Erasure::new(4, 2, 64).has_valid_dimensions());
+        assert!(Erasure::new_with_options(6, 4, 1, true).has_valid_dimensions());
+
+        // Corrupted on-disk metadata with a zero block_size or zero data_shards
+        // must be rejected before it reaches the shard/offset divisions.
+        // (parity_shards is kept 0 for the zero-data_shards cases so the
+        // Reed-Solomon encoder is not constructed with an invalid shard count.)
+        assert!(!Erasure::new(4, 2, 0).has_valid_dimensions());
+        assert!(!Erasure::new(0, 0, 64).has_valid_dimensions());
+        assert!(!Erasure::new(0, 0, 0).has_valid_dimensions());
+    }
+
     #[test]
     fn encode_data_owned_matches_borrowed_path() {
         for uses_legacy in [false, true] {

crates/ecstore/src/set_disk/read.rs

@@ -597,7 +597,7 @@ impl SetDisks {
 
         // Erasure params come from on-disk metadata; zero values must fail the read
         // instead of panicking on the block/shard divisions below.
-        if erasure.block_size == 0 || erasure.data_shards == 0 {
+        if !erasure.has_valid_dimensions() {
             return Err(Error::other(format!(
                 "invalid erasure metadata for {bucket}/{object}: block_size={}, data_blocks={}",
                 erasure.block_size, erasure.data_shards
@@ -959,8 +959,6 @@ impl SetDisks {
         metrics_size_bucket: &'static str,
         prefer_data_blocks_first_reader_setup: bool,
     ) -> Result<GetCodecStreamingReaderBuildOutcome> {
-        let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
-
         let erasure = coding::Erasure::new_with_options(
             fi.erasure.data_blocks,
             fi.erasure.parity_blocks,
@@ -968,6 +966,18 @@ impl SetDisks {
             fi.uses_legacy_checksum,
         );
 
+        // Erasure params come from on-disk metadata; zero values must fail the read
+        // instead of panicking on the block/shard divisions inside the codec
+        // streaming reader below. Mirrors the legacy multipart guard.
+        if !erasure.has_valid_dimensions() {
+            return Err(Error::other(format!(
+                "invalid erasure metadata for {bucket}/{object}: block_size={}, data_blocks={}",
+                erasure.block_size, erasure.data_shards
+            )));
+        }
+
+        let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
+
         if fi.parts.len() == 1 {
             let part = &fi.parts[0];
             let part_length =
@@ -2296,6 +2306,34 @@ mod tests {
         )
     }
 
+    #[tokio::test]
+    async fn codec_streaming_reader_rejects_zero_block_size_metadata() {
+        // Corrupted on-disk metadata: valid data/parity blocks but block_size == 0.
+        // The codec streaming entry must reject this and return an error instead of
+        // panicking on the block_size division inside the reader (mirrors the legacy
+        // multipart guard). The guard fires before any disk access, so empty disk /
+        // parts-metadata slices are sufficient.
+        let mut fi = codec_streaming_test_fileinfo(1024, 1);
+        fi.erasure.block_size = 0;
+
+        let result = SetDisks::get_object_decode_reader_with_fileinfo(
+            CODEC_STREAMING_TEST_BUCKET,
+            CODEC_STREAMING_TEST_OBJECT,
+            &fi,
+            &[],
+            &[],
+            0,
+            0,
+            false,
+            "test-object-class",
+            "test-size-bucket",
+            false,
+        )
+        .await;
+
+        assert!(result.is_err(), "zero block_size metadata must be rejected, not panic");
+    }
+
     #[tokio::test]
     async fn multipart_codec_streaming_reader_reads_parts_in_order() {
         let readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>> = vec![