fix(format): bound each filter stage by what the stages before it produce

Every decode stage was capped at the chunk's decoded size. That holds
only when every filter ahead of the codec preserves size; Fletcher32
does not (it appends a 4-byte checksum), so a pipeline with Fletcher32
before deflate (NetCDF-4's fletcher32 -> shuffle -> deflate ordering,
h5repack's "all filters") failed with "deflate: output exceeds size
limit" on every chunk.

decompress_chunk_masked now computes each stage's bound by running the
chunk size forward through the filters that precede it in write order
(and that the chunk's mask did not skip): shuffle keeps the size,
Fletcher32 adds 4, any codec adds at most n/8 + 64. The cap is still a
small constant factor of the chunk, so a decompression bomb is rejected
as before (tested).

Shuffle also had to learn libhdf5's handling of a length that is not a
whole number of elements (chunk + checksum): shuffle the whole elements
and leave the trailing bytes in place, in both directions. It used to
refuse such data.

Regression: h5py_fletcher32_before_deflate_reads (fletcher->shuffle->
gzip, fletcher->gzip, shuffle->fletcher->gzip, and a 2-D i32 grid),
fletcher32_ahead_of_deflate_stays_bounded and
shuffle_leaves_a_partial_trailing_element_in_place.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

crates/clawhdf5-format/src/filters.rs

@@ -31,6 +31,26 @@ pub fn decompress_chunk(
     decompress_chunk_masked(compressed, pipeline, chunk_size, element_size, 0)
 }
 
+/// Upper bound on the output of filter `filter_id` applied (in the write
+/// direction) to `input` bytes. 0 means "unknown" and stays unknown.
+///
+/// Shuffle preserves the size and Fletcher32 appends a 4-byte checksum. Any
+/// other filter is a codec whose output can exceed its input on
+/// incompressible data (deflate's stored blocks, LZ4's and zstd's literal
+/// runs, codec headers); `n + n/8 + 64` covers every supported codec's worst
+/// case while still bounding a decompression bomb to a small multiple of the
+/// chunk.
+fn filter_output_bound(filter_id: u16, input: usize) -> usize {
+    if input == 0 {
+        return 0;
+    }
+    match filter_id {
+        FILTER_SHUFFLE => input,
+        FILTER_FLETCHER32 => input.saturating_add(4),
+        _ => input.saturating_add(input / 8).saturating_add(64),
+    }
+}
+
 /// Whether bit `index` of a chunk's filter mask says filter `index` was
 /// skipped when the chunk was written.
 fn filter_skipped(filter_mask: u32, index: usize) -> bool {
@@ -48,8 +68,12 @@ pub fn all_filters_skipped(pipeline: &FilterPipeline, filter_mask: u32) -> bool
 /// optional filter that declined, or a direct chunk write), so only that
 /// filter is skipped here; the others are still undone, in reverse order.
 ///
-/// `chunk_size` is the chunk's decoded size (0 if unknown); it caps each
+/// `chunk_size` is the chunk's decoded size (0 if unknown). Each stage's
-/// stage's output.
+/// output is capped at what the filters before it (in write order) can have
+/// produced from `chunk_size` bytes — e.g. a Fletcher32 checksum placed
+/// before deflate (NetCDF-4's ordering) makes deflate's output 4 bytes
+/// larger than the chunk — so the decompression-bomb limit stays tight
+/// without rejecting valid pipelines.
 pub fn decompress_chunk_masked(
     compressed: &[u8],
     pipeline: &FilterPipeline,
@@ -57,12 +81,23 @@ pub fn decompress_chunk_masked(
     element_size: u32,
     filter_mask: u32,
 ) -> Result<Vec<u8>, FormatError> {
+    // bounds[i]: the most bytes that entered filter i on the write side, and
+    // so the most that undoing filter i may produce.
+    let mut bounds = Vec::with_capacity(pipeline.filters.len());
+    let mut size = chunk_size;
+    for (i, filter) in pipeline.filters.iter().enumerate() {
+        bounds.push(size);
+        if !filter_skipped(filter_mask, i) {
+            size = filter_output_bound(filter.filter_id, size);
+        }
+    }
+
     let mut data = compressed.to_vec();
     for (i, filter) in pipeline.filters.iter().enumerate().rev() {
         if filter_skipped(filter_mask, i) {
             continue;
         }
-        let bound = chunk_size;
+        let bound = bounds[i];
         data = match filter.filter_id {
             FILTER_SHUFFLE => shuffle_decompress(&data, element_size as usize)?,
             // `bound` caps the decoded size so these decoders can't be forced
@@ -944,13 +979,13 @@ fn shuffle_decompress(data: &[u8], element_size: usize) -> Result<Vec<u8>, Forma
     if element_size <= 1 {
         return Ok(data.to_vec());
     }
-    if !data.len().is_multiple_of(element_size) {
+    // Like libhdf5, only whole elements are shuffled; trailing bytes (e.g. a
-        return Err(FormatError::FilterError(
+    // Fletcher32 checksum appended before the shuffle) are stored as-is.
-            "shuffle: data length not a multiple of element size".into(),
+    let whole = data.len() - data.len() % element_size;
-        ));
+    let (data, tail) = data.split_at(whole);
-    }
     let num_elements = data.len() / element_size;
-    let mut result = vec![0u8; data.len()];
+    let mut result = vec![0u8; whole];
+    result.reserve_exact(tail.len());
 
     // The shuffled stream is `element_size` byte planes of `num_elements`
     // bytes each; un-shuffling interleaves them. This is on the read path of
@@ -981,6 +1016,7 @@ fn shuffle_decompress(data: &[u8], element_size: usize) -> Result<Vec<u8>, Forma
             }
         }
     }
+    result.extend_from_slice(tail);
 
     Ok(result)
 }
@@ -996,19 +1032,19 @@ fn shuffle_compress(data: &[u8], element_size: usize) -> Result<Vec<u8>, FormatE
     if element_size <= 1 {
         return Ok(data.to_vec());
     }
-    if !data.len().is_multiple_of(element_size) {
+    // Trailing bytes that don't make a whole element are left in place, as
-        return Err(FormatError::FilterError(
+    // libhdf5 does.
-            "shuffle: data length not a multiple of element size".into(),
+    let whole = data.len() - data.len() % element_size;
-        ));
+    let (data, tail) = data.split_at(whole);
-    }
     let num_elements = data.len() / element_size;
-    let mut result = vec![0u8; data.len()];
+    let mut result = vec![0u8; whole];
 
     match element_size {
         4 => shuffle_compress_4(data, num_elements, &mut result),
         8 => shuffle_compress_general(data, num_elements, element_size, &mut result),
         _ => shuffle_compress_general(data, num_elements, element_size, &mut result),
     }
+    result.extend_from_slice(tail);
 
     Ok(result)
 }
@@ -1503,6 +1539,51 @@ mod tests {
         );
     }
 
+    #[test]
+    #[cfg(feature = "deflate")]
+    fn fletcher32_ahead_of_deflate_stays_bounded() {
+        // NetCDF-4 order: the checksum is appended before shuffle and deflate,
+        // so deflate decodes chunk + 4 bytes.
+        let pipeline = FilterPipeline {
+            version: 2,
+            filters: vec![
+                filter(FILTER_FLETCHER32),
+                filter(FILTER_SHUFFLE),
+                filter(FILTER_DEFLATE),
+            ],
+        };
+        let data: Vec<u8> = (0..400).map(|i| (i * 13 % 251) as u8).collect();
+        let n = data.len();
+        let stored = compress_chunk(&data, &pipeline, 8).unwrap();
+        assert_eq!(decompress_chunk(&stored, &pipeline, n, 8).unwrap(), data);
+
+        // The cap still bites: a stream that inflates past chunk + 4 bytes
+        // is rejected rather than allocated.
+        let only_deflate = FilterPipeline {
+            version: 2,
+            filters: vec![filter(FILTER_DEFLATE)],
+        };
+        let oversized = compress_chunk(&vec![0u8; n + 5], &only_deflate, 8).unwrap();
+        let err = decompress_chunk(&oversized, &pipeline, n, 8).unwrap_err();
+        assert!(
+            matches!(err, FormatError::DecompressionError(_)),
+            "expected a size-limit error, got {err:?}"
+        );
+        // A bomb is still stopped near the chunk size.
+        let bomb = compress_chunk(&vec![0u8; 64 * n], &only_deflate, 8).unwrap();
+        assert!(decompress_chunk(&bomb, &pipeline, n, 8).is_err());
+    }
+
+    #[test]
+    fn shuffle_leaves_a_partial_trailing_element_in_place() {
+        // libhdf5 shuffles whole elements and copies the remainder as-is.
+        let data: Vec<u8> = (0..20).collect();
+        let shuffled = shuffle_compress(&data, 8).unwrap();
+        assert_eq!(&shuffled[16..], &data[16..]);
+        assert_eq!(&shuffled[..4], &[0, 8, 1, 9]);
+        assert_eq!(shuffle_decompress(&shuffled, 8).unwrap(), data);
+    }
+
     #[test]
     #[cfg(feature = "deflate")]
     fn pipeline_compress_decompress_roundtrip() {

crates/clawhdf5/tests/h5py_chunked_read_tests.rs

@@ -173,3 +173,67 @@ with h5py.File("{p}", "w") as f:
     let want: Vec<u8> = (1006..1026i32).flat_map(i32::to_le_bytes).collect();
     assert_eq!(part, want);
 }
+
+// ---------------------------------------------------------------------------
+// Size-changing filters ahead of a codec
+// ---------------------------------------------------------------------------
+
+/// Fletcher32 placed before the compressor (NetCDF-4's ordering) makes the
+/// codec's decoded output 4 bytes larger than the chunk. The decompression
+/// cap was the chunk size for every stage, so these files failed with
+/// "deflate: output exceeds size limit".
+#[test]
+fn h5py_fletcher32_before_deflate_reads() {
+    skip_if_no_python!();
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("fletcher_first.h5");
+    let p = path.display().to_string();
+    run_python(&format!(
+        r#"
+import h5py, numpy as np
+arr = np.sin(np.arange(5000) / 50.0)
+grid = np.arange(37 * 21, dtype="<i4").reshape(37, 21)
+ids = {{"fletcher32": 3, "shuffle": 2, "deflate": 1}}
+with h5py.File("{p}", "w") as f:
+    for name, steps, data, chunk in (
+        ("fl_shuf_gzip", ("fletcher32", "shuffle", "deflate"), arr, (500,)),
+        ("fl_gzip", ("fletcher32", "deflate"), arr, (500,)),
+        ("shuf_fl_gzip", ("shuffle", "fletcher32", "deflate"), arr, (500,)),
+        ("grid_fl_shuf_gzip", ("fletcher32", "shuffle", "deflate"), grid, (8, 5)),
+    ):
+        dcpl = h5py.h5p.create(h5py.h5p.DATASET_CREATE)
+        dcpl.set_chunk(chunk)
+        for s in steps:
+            if s == "deflate":
+                dcpl.set_deflate(4)
+            elif s == "shuffle":
+                dcpl.set_shuffle()
+            else:
+                dcpl.set_fletcher32()
+        tid = h5py.h5t.py_create(data.dtype)
+        space = h5py.h5s.create_simple(data.shape)
+        d = h5py.h5d.create(f.id, name.encode(), tid, space, dcpl=dcpl)
+        d.write(h5py.h5s.ALL, h5py.h5s.ALL, np.ascontiguousarray(data))
+        order = [d.get_create_plist().get_filter(i)[0] for i in range(len(steps))]
+        assert order == [ids[s] for s in steps], order
+"#
+    ));
+
+    let file = File::open(&path).unwrap();
+    let arr: Vec<f64> = (0..5000).map(|i| (f64::from(i) / 50.0).sin()).collect();
+    for name in ["fl_shuf_gzip", "fl_gzip", "shuf_fl_gzip"] {
+        let values = file.dataset(name).unwrap().read_f64().unwrap();
+        assert_eq!(values.len(), arr.len(), "{name}");
+        for (i, (v, w)) in values.iter().zip(&arr).enumerate() {
+            assert!((v - w).abs() < 1e-12, "{name}[{i}]: {v} vs {w}");
+        }
+    }
+    let grid: Vec<i32> = (0..37 * 21).collect();
+    assert_eq!(
+        file.dataset("grid_fl_shuf_gzip")
+            .unwrap()
+            .read_i32()
+            .unwrap(),
+        grid
+    );
+}