//! HDF5 Fractal Heap parsing for v2 group link storage.

#[cfg(not(feature = "std"))]
use alloc::{format, vec::Vec};

#[cfg(feature = "checksum")]
use byteorder::{ByteOrder, LittleEndian};

use crate::addr::to_usize;
use crate::btree_v2::{BTreeV2Header, find_btree_v2_records_in};
use crate::error::FormatError;
use crate::filter_pipeline::FilterPipeline;
use crate::storage::{Storage, Window, len_usize, read_exact_at};

/// Parsed fractal heap header (signature "FRHP").
#[derive(Debug, Clone)]
pub struct FractalHeapHeader {
    /// Length of heap IDs in bytes (typically 7).
    pub heap_id_length: u16,
    /// I/O filter encoded length (0 = no filters).
    pub io_filter_encoded_length: u16,
    /// Maximum size of a managed object.
    pub max_managed_object_size: u32,
    /// Width of the doubling table.
    pub table_width: u16,
    /// Starting block size in the doubling table.
    pub starting_block_size: u64,
    /// Maximum direct block size.
    pub max_direct_block_size: u64,
    /// Maximum heap size in bits (determines offset bit width in heap IDs).
    pub max_heap_size: u16,
    /// Starting row of indirect blocks in the doubling table.
    pub starting_row_of_indirect_blocks: u16,
    /// Address of the root block.
    pub root_block_address: u64,
    /// Number of rows in root indirect block (0 = root is direct block).
    pub current_rows_in_root_indirect_block: u16,
    /// Total number of managed objects.
    pub managed_objects_count: u64,
    /// Address of the v2 B-tree indexing "huge" objects (undefined address
    /// when the heap has none). Huge objects are those larger than
    /// `max_managed_object_size`; they live outside the heap's blocks.
    pub huge_btree_address: u64,
    /// The heap's I/O filter pipeline, if it has one. It applies to managed
    /// direct blocks and to huge objects.
    pub filter_pipeline: Option<FilterPipeline>,
    /// Stored (filtered) size of the root direct block; meaningful only when
    /// the heap is filtered and its root is a direct block.
    pub root_direct_block_filtered_size: u64,
    /// Filter mask of the root direct block (bit *i* set = filter *i*
    /// skipped); meaningful only when the heap is filtered.
    pub root_direct_block_filter_mask: u32,
    /// Size of addresses in the file ("Size of Offsets").
    pub offset_size: u8,
    /// Size of lengths in the file ("Size of Lengths").
    pub length_size: u8,
}

fn read_offset(data: &[u8], pos: usize, size: u8) -> Result<u64, FormatError> {
    let s = size as usize;
    if pos.checked_add(s).is_none_or(|end| end > data.len()) {
        return Err(FormatError::UnexpectedEof {
            expected: pos.saturating_add(s),
            available: data.len(),
        });
    }
    Ok(match size {
        2 => u16::from_le_bytes([data[pos], data[pos + 1]]) as u64,
        4 => u32::from_le_bytes([data[pos], data[pos + 1], data[pos + 2], data[pos + 3]]) as u64,
        8 => u64::from_le_bytes([
            data[pos],
            data[pos + 1],
            data[pos + 2],
            data[pos + 3],
            data[pos + 4],
            data[pos + 5],
            data[pos + 6],
            data[pos + 7],
        ]),
        _ => return Err(FormatError::InvalidOffsetSize(size)),
    })
}

fn ensure_len(data: &[u8], pos: usize, needed: usize) -> Result<(), FormatError> {
    match pos.checked_add(needed) {
        Some(end) if end <= data.len() => Ok(()),
        _ => Err(FormatError::UnexpectedEof {
            expected: pos.saturating_add(needed),
            available: data.len(),
        }),
    }
}

fn is_undefined(val: u64, offset_size: u8) -> bool {
    match offset_size {
        2 => val == 0xFFFF,
        4 => val == 0xFFFF_FFFF,
        8 => val == 0xFFFF_FFFF_FFFF_FFFF,
        _ => false,
    }
}

/// Little-endian unsigned integer of up to 8 bytes.
fn le_uint(bytes: &[u8]) -> u64 {
    bytes
        .iter()
        .take(8)
        .enumerate()
        .fold(0u64, |acc, (i, &b)| acc | (u64::from(b) << (i * 8)))
}

fn heap_error(msg: &str) -> FormatError {
    FormatError::ChunkedReadError(format!("fractal heap: {msg}"))
}

/// Heap ID type, from bits 4-5 of an ID's first byte (libhdf5's
/// `H5HF_ID_TYPE_MASK`, 0x30); bits 6-7 are the ID version, which must be 0.
const HEAP_ID_MANAGED: u8 = 0;
const HEAP_ID_HUGE: u8 = 1;
const HEAP_ID_TINY: u8 = 2;

/// The type (0 managed, 1 huge, 2 tiny) of a heap ID from its first byte,
/// refusing an ID version other than 0.
fn heap_id_type(first: u8) -> Result<u8, FormatError> {
    if first >> 6 != 0 {
        return Err(heap_error("unsupported heap ID version"));
    }
    Ok((first >> 4) & 0x03)
}

/// v2 B-tree record types indexing a heap's huge objects.
const BTREE_HUGE_INDIRECT: u8 = 1;
const BTREE_HUGE_INDIRECT_FILTERED: u8 = 2;

impl FractalHeapHeader {
    /// Parse a fractal heap header at the given offset.
    pub fn parse(
        file_data: &[u8],
        offset: usize,
        offset_size: u8,
        length_size: u8,
    ) -> Result<FractalHeapHeader, FormatError> {
        Self::parse_in(file_data, offset as u64, offset_size, length_size)
    }

    /// [`Self::parse`] over any [`Storage`]: one read of the header (two
    /// when it holds an I/O filter pipeline).
    pub fn parse_in<S: Storage + ?Sized>(
        file: &S,
        offset: u64,
        offset_size: u8,
        length_size: u8,
    ) -> Result<FractalHeapHeader, FormatError> {
        // Every field up to the checksum, without and with the filter
        // information; the window holds all of it (or ends at the end of
        // the file), so its bounds checks are the whole-file ones.
        let (os, ls) = (usize::from(offset_size), usize::from(length_size));
        let unfiltered_len = 26 + 12 * ls + 3 * os;
        let mut w = Window::read(file, offset, unfiltered_len)?;
        if w.bytes.len() == unfiltered_len {
            let filter_len = usize::from(u16::from_le_bytes([w.bytes[7], w.bytes[8]]));
            if filter_len > 0 {
                w = Window::read(file, offset, unfiltered_len + ls + 4 + filter_len)?;
            }
        }
        let ensure_len = |_: &[u8], pos: usize, needed: usize| w.ensure(pos, needed);
        let read_offset = |_: &[u8], pos: usize, size: u8| {
            w.ensure(pos, usize::from(size))?;
            read_offset(&w.bytes, pos, size)
        };
        let file_data: &[u8] = &w.bytes;
        let offset = 0usize;
        ensure_len(file_data, offset, 5)?;
        if &file_data[offset..offset + 4] != b"FRHP" {
            return Err(FormatError::InvalidFractalHeapSignature);
        }

        let version = file_data[offset + 4];
        if version != 0 {
            return Err(FormatError::InvalidFractalHeapVersion(version));
        }

        let mut pos = offset + 5;
        ensure_len(file_data, pos, 2)?;
        let heap_id_length = u16::from_le_bytes([file_data[pos], file_data[pos + 1]]);
        pos += 2;

        ensure_len(file_data, pos, 2)?;
        let io_filter_encoded_length = u16::from_le_bytes([file_data[pos], file_data[pos + 1]]);
        pos += 2;

        ensure_len(file_data, pos, 1)?;
        let _flags = file_data[pos];
        pos += 1;

        ensure_len(file_data, pos, 4)?;
        let max_managed_object_size = u32::from_le_bytes([
            file_data[pos],
            file_data[pos + 1],
            file_data[pos + 2],
            file_data[pos + 3],
        ]);
        pos += 4;

        // next_huge_object_id (length_size)
        ensure_len(file_data, pos, ls)?;
        pos += ls;
        // btree_huge_objects_address (offset_size)
        let huge_btree_address = read_offset(file_data, pos, offset_size)?;
        pos += os;

        // Skip: free_space_managed_blocks(ls), managed_block_free_space_manager_address(os),
        // managed_space_in_heap(ls), allocated_managed_space_in_heap(ls),
        // direct_block_allocation_iterator_offset(ls)
        let skip_size = 4 * ls + os;
        ensure_len(file_data, pos, skip_size)?;
        pos += skip_size;

        // managed_objects_count (length_size)
        let managed_objects_count = read_offset(file_data, pos, length_size)?;
        pos += ls;

        // huge_objects_size, huge_objects_count, tiny_objects_size,
        // tiny_objects_count (length_size each)
        pos += 4 * ls;

        // table_width (2)
        ensure_len(file_data, pos, 2)?;
        let table_width = u16::from_le_bytes([file_data[pos], file_data[pos + 1]]);
        pos += 2;

        // starting_block_size (length_size)
        let starting_block_size = read_offset(file_data, pos, length_size)?;
        pos += ls;

        // max_direct_block_size (length_size)
        let max_direct_block_size = read_offset(file_data, pos, length_size)?;
        pos += ls;

        // max_heap_size (2)
        ensure_len(file_data, pos, 2)?;
        let max_heap_size = u16::from_le_bytes([file_data[pos], file_data[pos + 1]]);
        pos += 2;

        // starting_row_of_indirect_blocks (2)
        ensure_len(file_data, pos, 2)?;
        let starting_row_of_indirect_blocks =
            u16::from_le_bytes([file_data[pos], file_data[pos + 1]]);
        pos += 2;

        // root_block_address (offset_size)
        let root_block_address = read_offset(file_data, pos, offset_size)?;
        pos += os;

        // current_rows_in_root_indirect_block (2)
        ensure_len(file_data, pos, 2)?;
        let current_rows_in_root_indirect_block =
            u16::from_le_bytes([file_data[pos], file_data[pos + 1]]);
        pos += 2;

        // With I/O filters: root direct block's filtered size (length_size),
        // its filter mask (4), then the encoded filter pipeline message.
        let mut filter_pipeline = None;
        let mut root_direct_block_filtered_size = 0;
        let mut root_direct_block_filter_mask = 0;
        if io_filter_encoded_length > 0 {
            root_direct_block_filtered_size = read_offset(file_data, pos, length_size)?;
            pos += ls;
            ensure_len(file_data, pos, 4)?;
            root_direct_block_filter_mask = u32::from_le_bytes([
                file_data[pos],
                file_data[pos + 1],
                file_data[pos + 2],
                file_data[pos + 3],
            ]);
            pos += 4;
            let n = io_filter_encoded_length as usize;
            ensure_len(file_data, pos, n)?;
            filter_pipeline = Some(FilterPipeline::parse(&file_data[pos..pos + n])?);
            pos += n;
        }

        // Validate header checksum
        #[cfg(feature = "checksum")]
        {
            ensure_len(file_data, pos, 4)?;
            let stored = LittleEndian::read_u32(&file_data[pos..pos + 4]);
            let computed = crate::checksum::jenkins_lookup3(&file_data[offset..pos]);
            if computed != stored {
                return Err(FormatError::ChecksumMismatch {
                    expected: stored,
                    computed,
                });
            }
        }
        #[cfg(not(feature = "checksum"))]
        let _ = pos;

        Ok(FractalHeapHeader {
            heap_id_length,
            io_filter_encoded_length,
            max_managed_object_size,
            table_width,
            starting_block_size,
            max_direct_block_size,
            max_heap_size,
            starting_row_of_indirect_blocks,
            root_block_address,
            current_rows_in_root_indirect_block,
            managed_objects_count,
            huge_btree_address,
            filter_pipeline,
            root_direct_block_filtered_size,
            root_direct_block_filter_mask,
            offset_size,
            length_size,
        })
    }

    /// Decode a managed heap ID into (offset_in_heap, object_length).
    ///
    /// The heap ID layout for managed objects (type 0):
    /// - Byte 0: bits 6-7 = version (0), bits 4-5 = type (0), bits 0-3 = reserved
    /// - Bytes 1+: offset (max_heap_size bits, LE) then length (remaining bits, LE)
    pub fn decode_managed_id(&self, id_bytes: &[u8]) -> Result<(u64, u64), FormatError> {
        if id_bytes.is_empty() {
            return Err(FormatError::UnexpectedEof {
                expected: 1,
                available: 0,
            });
        }

        let id_type = heap_id_type(id_bytes[0])?;
        if id_type != HEAP_ID_MANAGED {
            return Err(FormatError::InvalidHeapIdType(id_type));
        }

        // Bytes 1+ contain offset and length packed in little-endian order.
        // offset uses max_heap_size bits, length uses the remaining bits.
        let payload = &id_bytes[1..];
        let mut combined: u64 = 0;
        for (i, &b) in payload.iter().enumerate() {
            if i >= 8 {
                break;
            }
            combined |= (b as u64) << (i * 8);
        }

        let offset_bits = self.max_heap_size as u32;
        let offset_mask = if offset_bits >= 64 {
            u64::MAX
        } else {
            (1u64 << offset_bits) - 1
        };
        let heap_offset = combined & offset_mask;

        let total_payload_bits = (payload.len() as u32) * 8;
        let length_bits = total_payload_bits.saturating_sub(offset_bits);
        let length_val = if length_bits == 0 {
            0
        } else {
            let length_mask = if length_bits >= 64 {
                u64::MAX
            } else {
                (1u64 << length_bits) - 1
            };
            (combined >> offset_bits) & length_mask
        };

        Ok((heap_offset, length_val))
    }

    /// Read any object from the heap given its raw heap ID bytes: managed
    /// (stored in the heap's blocks), huge (stored outside them, found
    /// directly from the ID or through the huge-object v2 B-tree, optionally
    /// filtered) or tiny (stored in the ID itself).
    ///
    /// Despite its name this accepts every ID type; `offset_size` must match
    /// the one the header was parsed with.
    pub fn read_managed_object(
        &self,
        file_data: &[u8],
        id_bytes: &[u8],
        offset_size: u8,
    ) -> Result<Vec<u8>, FormatError> {
        self.read_managed_object_in(file_data, id_bytes, offset_size)
    }

    /// [`Self::read_managed_object`] over any [`Storage`].
    pub fn read_managed_object_in<S: Storage + ?Sized>(
        &self,
        file_data: &S,
        id_bytes: &[u8],
        offset_size: u8,
    ) -> Result<Vec<u8>, FormatError> {
        crate::lookup_stats::heap_object_read();
        let Some(&first) = id_bytes.first() else {
            return Err(FormatError::UnexpectedEof {
                expected: 1,
                available: 0,
            });
        };
        match heap_id_type(first)? {
            HEAP_ID_MANAGED => self.read_heap_managed(file_data, id_bytes, offset_size),
            HEAP_ID_HUGE => self.read_huge_object(file_data, id_bytes),
            HEAP_ID_TINY => self.read_tiny_object(id_bytes),
            other => Err(FormatError::InvalidHeapIdType(other)),
        }
    }

    /// Whether a huge object's ID holds its address and length directly
    /// (libhdf5 does this when they fit in the ID), rather than a key into
    /// the huge-object B-tree.
    fn huge_ids_direct(&self) -> bool {
        let room = usize::from(self.heap_id_length).saturating_sub(1);
        let os = usize::from(self.offset_size);
        let ls = usize::from(self.length_size);
        if self.filter_pipeline.is_some() {
            room >= os + ls + 4 + ls
        } else {
            room >= os + ls
        }
    }

    /// Read a huge object (heap ID type 1).
    fn read_huge_object<S: Storage + ?Sized>(
        &self,
        file: &S,
        id: &[u8],
    ) -> Result<Vec<u8>, FormatError> {
        let os = usize::from(self.offset_size);
        let ls = usize::from(self.length_size);
        // (address, stored length, filter mask, decoded length); the last two
        // only matter for a filtered heap.
        let (addr, stored_len, mask, mem_len) = if self.huge_ids_direct() {
            let body = &id[1..];
            let need = if self.filter_pipeline.is_some() {
                os + ls + 4 + ls
            } else {
                os + ls
            };
            ensure_len(body, 0, need)?;
            let addr = le_uint(&body[..os]);
            let len = le_uint(&body[os..os + ls]);
            if self.filter_pipeline.is_some() {
                let mask = u32::from_le_bytes([
                    body[os + ls],
                    body[os + ls + 1],
                    body[os + ls + 2],
                    body[os + ls + 3],
                ]);
                let mem = le_uint(&body[os + ls + 4..os + ls + 4 + ls]);
                (addr, len, mask, mem)
            } else {
                (addr, len, 0, len)
            }
        } else {
            let key_len = (usize::from(self.heap_id_length).saturating_sub(1)).min(8);
            ensure_len(id, 1, key_len)?;
            let key = le_uint(&id[1..1 + key_len]);
            self.find_huge_record(file, key)?
        };

        let start = usize::try_from(addr).map_err(|_| heap_error("huge object address"))?;
        let len = usize::try_from(stored_len).map_err(|_| heap_error("huge object length"))?;
        let stored = read_exact_at(file, start as u64, len)?;
        match &self.filter_pipeline {
            None => Ok(stored.into_owned()),
            Some(pipeline) => {
                let mem = usize::try_from(mem_len).map_err(|_| heap_error("huge object size"))?;
                let out = crate::filters::decompress_chunk_masked(&stored, pipeline, mem, 1, mask)?;
                if out.len() != mem {
                    return Err(heap_error("filtered huge object decoded to the wrong size"));
                }
                Ok(out)
            }
        }
    }

    /// Look up huge object `key` in the huge-object v2 B-tree, returning
    /// (address, stored length, filter mask, decoded length).
    fn find_huge_record<S: Storage + ?Sized>(
        &self,
        file: &S,
        key: u64,
    ) -> Result<(u64, u64, u32, u64), FormatError> {
        if is_undefined(self.huge_btree_address, self.offset_size) {
            return Err(heap_error(
                "huge object ID but the heap has no huge-object index",
            ));
        }
        let hdr = BTreeV2Header::parse_in(
            file,
            self.huge_btree_address,
            self.offset_size,
            self.length_size,
        )?;
        let os = usize::from(self.offset_size);
        let ls = usize::from(self.length_size);
        let filtered = self.filter_pipeline.is_some();
        let (expected_type, rec_len) = if filtered {
            (BTREE_HUGE_INDIRECT_FILTERED, os + ls + 4 + ls + ls)
        } else {
            (BTREE_HUGE_INDIRECT, os + ls + ls)
        };
        if hdr.tree_type != expected_type || usize::from(hdr.record_size) < rec_len {
            return Err(heap_error("unexpected huge-object B-tree record type"));
        }
        // Records are ordered by ID (the last field): descend to the ones
        // equal to `key` instead of reading the whole index.
        let id_at = rec_len - ls;
        let records = find_btree_v2_records_in(file, &hdr, self.offset_size, &mut |r| {
            le_uint(&r[id_at..id_at + ls]).cmp(&key)
        })?;
        for rec in &records {
            let d = &rec.data;
            if d.len() < rec_len {
                continue;
            }
            let addr = le_uint(&d[..os]);
            let len = le_uint(&d[os..os + ls]);
            if filtered {
                let mask = u32::from_le_bytes([
                    d[os + ls],
                    d[os + ls + 1],
                    d[os + ls + 2],
                    d[os + ls + 3],
                ]);
                let mem = le_uint(&d[os + ls + 4..os + 2 * ls + 4]);
                let id = le_uint(&d[os + 2 * ls + 4..os + 3 * ls + 4]);
                if id == key {
                    return Ok((addr, len, mask, mem));
                }
            } else {
                let id = le_uint(&d[os + ls..os + 2 * ls]);
                if id == key {
                    return Ok((addr, len, 0, len));
                }
            }
        }
        Err(heap_error("huge object not found in its B-tree"))
    }

    /// Read a tiny object (heap ID type 2), stored in the ID itself.
    fn read_tiny_object(&self, id: &[u8]) -> Result<Vec<u8>, FormatError> {
        // libhdf5 uses a one-byte length (low 4 bits of byte 0) unless the ID
        // is long enough to need 12 bits, which then borrow byte 1.
        let extended = usize::from(self.heap_id_length).saturating_sub(1) > 17;
        let (len, start) = if extended {
            ensure_len(id, 0, 2)?;
            (
                ((usize::from(id[0] & 0x0F)) << 8 | usize::from(id[1])) + 1,
                2,
            )
        } else {
            (usize::from(id[0] & 0x0F) + 1, 1)
        };
        ensure_len(id, start, len)?;
        Ok(id[start..start + len].to_vec())
    }

    /// Read a managed object (heap ID type 0).
    fn read_heap_managed<S: Storage + ?Sized>(
        &self,
        file_data: &S,
        id_bytes: &[u8],
        offset_size: u8,
    ) -> Result<Vec<u8>, FormatError> {
        let (heap_offset, obj_len) = self.decode_managed_id(id_bytes)?;

        if is_undefined(self.root_block_address, offset_size) {
            return Err(FormatError::UnexpectedEof {
                expected: 1,
                available: 0,
            });
        }

        if self.current_rows_in_root_indirect_block == 0 {
            // Root is a direct block
            self.read_from_direct_block(
                file_data,
                DirectBlock {
                    addr: to_usize(self.root_block_address)?,
                    size: self.starting_block_size,
                    heap_offset: 0,
                    filtered_size: self.root_direct_block_filtered_size,
                    filter_mask: self.root_direct_block_filter_mask,
                },
                heap_offset,
                to_usize(obj_len)?,
            )
        } else {
            // Root is an indirect block — limit recursion to 64 levels
            self.read_from_indirect_block(
                file_data,
                to_usize(self.root_block_address)?,
                self.current_rows_in_root_indirect_block,
                0, // block offset
                heap_offset,
                to_usize(obj_len)?,
                offset_size,
                64, // max recursion depth
            )
        }
    }

    /// Read an object from a direct block.
    ///
    /// The heap offset is relative to the start of the block (including its
    /// header), so we just add it to the block address minus the block's heap
    /// offset. A filtered heap stores each direct block (header included)
    /// through its filter pipeline, so the block is decoded first.
    fn read_from_direct_block<S: Storage + ?Sized>(
        &self,
        file: &S,
        block: DirectBlock,
        target_offset: u64,
        length: usize,
    ) -> Result<Vec<u8>, FormatError> {
        if target_offset < block.heap_offset {
            return Err(FormatError::UnexpectedEof {
                expected: to_usize(block.heap_offset)?,
                available: to_usize(target_offset)?,
            });
        }
        let local_offset = to_usize(target_offset - block.heap_offset)?;
        if let Some(pipeline) = &self.filter_pipeline {
            let stored_len = usize::try_from(block.filtered_size)
                .map_err(|_| heap_error("direct block size"))?;
            let size = usize::try_from(block.size).map_err(|_| heap_error("direct block size"))?;
            let stored = read_exact_at(file, block.addr as u64, stored_len)?;
            let decoded = crate::filters::decompress_chunk_masked(
                &stored,
                pipeline,
                size,
                1,
                block.filter_mask,
            )?;
            ensure_len(&decoded, local_offset, length)?;
            return Ok(decoded[local_offset..local_offset + length].to_vec());
        }
        let pos = block
            .addr
            .checked_add(local_offset)
            .ok_or(FormatError::UnexpectedEof {
                expected: usize::MAX,
                available: len_usize(file),
            })?;
        Ok(read_exact_at(file, pos as u64, length)?.into_owned())
    }

    /// Read an object by traversing an indirect block to find the right direct block.
    #[allow(clippy::too_many_arguments)]
    fn read_from_indirect_block<S: Storage + ?Sized>(
        &self,
        file: &S,
        iblock_addr: usize,
        nrows: u16,
        iblock_heap_offset: u64,
        target_offset: u64,
        length: usize,
        offset_size: u8,
        depth_remaining: u16,
    ) -> Result<Vec<u8>, FormatError> {
        if depth_remaining == 0 {
            return Err(FormatError::ChunkedReadError(
                "fractal heap: maximum recursion depth exceeded".into(),
            ));
        }
        let block_offset_bytes = (self.max_heap_size as usize).div_ceil(8);
        let iblock_header = 5 + offset_size as usize + block_offset_bytes;
        let nrows_usize = nrows as usize;

        // Rows below max_direct_rows hold direct blocks; rows at/above hold
        // child indirect blocks. (NOT the FRHP "starting rows" field.)
        let start_indirect = self.max_direct_rows();
        let max_direct_rows = nrows_usize.min(start_indirect);

        // The block up to its last child entry. The walk below reads
        // entries in order and stops at the one covering the target, which
        // the geometry alone locates, so the first window ends there: a
        // header claiming a huge table costs a read of the entries in front
        // of the target, not of the rest of the file. Only when that entry
        // is unallocated (or none covers the target) does the walk go on,
        // over the whole block. Either window holds what it was asked for or
        // ends at the end of the file, so its bounds checks are the
        // whole-file ones.
        let direct_entry = usize::from(offset_size)
            + if self.filter_pipeline.is_some() {
                usize::from(self.length_size) + 4
            } else {
                0
            };
        let direct_entries = max_direct_rows.saturating_mul(usize::from(self.table_width));
        let entries_len = |n: usize| {
            n.min(direct_entries)
                .saturating_mul(direct_entry)
                .saturating_add(
                    n.saturating_sub(direct_entries)
                        .saturating_mul(usize::from(offset_size)),
                )
        };
        let all_entries = direct_entries.saturating_add(
            nrows_usize
                .saturating_sub(start_indirect)
                .saturating_mul(usize::from(self.table_width)),
        );
        let block_len = iblock_header.saturating_add(entries_len(all_entries));
        let target_entry = self.indirect_entry_for(nrows_usize, iblock_heap_offset, target_offset);
        let first_len = target_entry.map_or(block_len, |i| {
            iblock_header
                .saturating_add(entries_len(i.saturating_add(1)))
                .min(block_len)
        });
        let mut next = self.walk_indirect_block(
            &Window::read(file, iblock_addr as u64, first_len)?,
            nrows_usize,
            iblock_heap_offset,
            target_offset,
            offset_size,
            target_entry.map_or(usize::MAX, |i| i.saturating_add(1)),
        )?;
        if next.is_none() && first_len < block_len {
            next = self.walk_indirect_block(
                &Window::read(file, iblock_addr as u64, block_len)?,
                nrows_usize,
                iblock_heap_offset,
                target_offset,
                offset_size,
                usize::MAX,
            )?;
        }
        match next {
            Some(IndirectChild::Direct(block)) => {
                self.read_from_direct_block(file, block, target_offset, length)
            }
            Some(IndirectChild::Indirect {
                addr,
                nrows,
                heap_offset,
            }) => self.read_from_indirect_block(
                file,
                addr,
                nrows,
                heap_offset,
                target_offset,
                length,
                offset_size,
                depth_remaining - 1,
            ),
            None => Err(FormatError::UnexpectedEof {
                expected: to_usize(target_offset)?.saturating_add(length),
                available: len_usize(file),
            }),
        }
    }

    /// Which child entry of an indirect block (numbered in walk order:
    /// direct rows, then indirect rows) covers `target_offset`, from the
    /// doubling-table geometry alone — the entry
    /// [`Self::walk_indirect_block`] stops at if it is allocated. `None`
    /// when no entry does.
    fn indirect_entry_for(&self, nrows: usize, heap_offset: u64, target: u64) -> Option<usize> {
        // The walk adds block sizes with saturation; in u128 the same test
        // is `cur <= target < cur + size` without it (a target of u64::MAX
        // is never inside a saturated range).
        if target == u64::MAX {
            return None;
        }
        let (tw, target) = (u128::from(self.table_width), u128::from(target));
        let mut cur = u128::from(heap_offset);
        let mut before = 0usize;
        for row in 0..nrows {
            if target < cur {
                return None;
            }
            // Direct and indirect rows alike span this row's block size per
            // entry.
            let size = u128::from(self.block_size_for_row(row));
            let span = size * tw;
            if size > 0 && target < cur + span {
                let col = usize::try_from((target - cur) / size).ok()?;
                return before.checked_add(col);
            }
            cur += span;
            before = before.saturating_add(self.table_width as usize);
        }
        None
    }

    /// Walk an indirect block's child entries in order, in the window `w`
    /// (the block from its signature on), and return the allocated child
    /// covering `target_offset`, or `None` when no entry among the first
    /// `limit` does.
    fn walk_indirect_block(
        &self,
        w: &Window<'_>,
        nrows: usize,
        iblock_heap_offset: u64,
        target_offset: u64,
        offset_size: u8,
        limit: usize,
    ) -> Result<Option<IndirectChild>, FormatError> {
        let ensure_len = |_: &[u8], pos: usize, needed: usize| w.ensure(pos, needed);
        let read_offset = |_: &[u8], pos: usize, size: u8| {
            w.ensure(pos, usize::from(size))?;
            read_offset(&w.bytes, pos, size)
        };
        let file_data: &[u8] = &w.bytes;
        let block_offset_bytes = (self.max_heap_size as usize).div_ceil(8);
        let iblock_header = 5 + offset_size as usize + block_offset_bytes;
        let tw = self.table_width as u64;
        let mut current_heap_offset = iblock_heap_offset;
        let start_indirect = self.max_direct_rows();
        let max_direct_rows = nrows.min(start_indirect);
        let mut walked = 0usize;

        // Parse indirect block header
        ensure_len(file_data, 0, 4)?;
        if &file_data[..4] != b"FHIB" {
            return Err(FormatError::InvalidFractalHeapSignature);
        }
        let mut pos = iblock_header;

        for row in 0..max_direct_rows {
            let block_size = self.block_size_for_row(row);

            for _col in 0..tw {
                if walked == limit {
                    return Ok(None);
                }
                walked += 1;
                let child_addr = read_offset(file_data, pos, offset_size)?;
                pos += offset_size as usize;

                // A filtered heap stores each direct block's filtered size
                // (length_size) and filter mask (4) after its address.
                let (filtered_size, filter_mask) = if self.filter_pipeline.is_some() {
                    let size = read_offset(file_data, pos, self.length_size)?;
                    pos += usize::from(self.length_size);
                    ensure_len(file_data, pos, 4)?;
                    let mask = u32::from_le_bytes([
                        file_data[pos],
                        file_data[pos + 1],
                        file_data[pos + 2],
                        file_data[pos + 3],
                    ]);
                    pos += 4;
                    (size, mask)
                } else {
                    (0, 0)
                };

                let block_end = current_heap_offset.saturating_add(block_size);
                if !is_undefined(child_addr, offset_size)
                    && target_offset >= current_heap_offset
                    && target_offset < block_end
                {
                    return Ok(Some(IndirectChild::Direct(DirectBlock {
                        addr: to_usize(child_addr)?,
                        size: block_size,
                        heap_offset: current_heap_offset,
                        filtered_size,
                        filter_mask,
                    })));
                }
                current_heap_offset = block_end;
            }
        }

        // Rows at and above `start_indirect` hold child indirect blocks. A
        // child in row r spans exactly that row's block size of heap space,
        // so it has as many rows as a table of that total size needs.
        for row in start_indirect..nrows {
            let child_space = self.block_size_for_row(row);
            let child_nrows = self.rows_for_size(child_space);

            for _col in 0..tw {
                if walked == limit {
                    return Ok(None);
                }
                walked += 1;
                let child_addr = read_offset(file_data, pos, offset_size)?;
                pos += offset_size as usize;

                let block_end = current_heap_offset.saturating_add(child_space);
                if !is_undefined(child_addr, offset_size)
                    && target_offset >= current_heap_offset
                    && target_offset < block_end
                {
                    return Ok(Some(IndirectChild::Indirect {
                        addr: to_usize(child_addr)?,
                        nrows: child_nrows,
                        heap_offset: current_heap_offset,
                    }));
                }
                current_heap_offset = block_end;
            }
        }
        Ok(None)
    }

    /// Number of rows in the doubling table whose block size is at most the
    /// maximum *direct* block size. Rows below this hold direct blocks; rows at
    /// or above it hold child indirect blocks.
    ///
    /// This is derived from the heap geometry, NOT the FRHP
    /// "Starting # of Rows in Root Indirect Block" field (a constant, often 1)
    /// — confusing the two makes a multi-direct-block heap unreadable.
    fn max_direct_rows(&self) -> usize {
        if self.starting_block_size == 0 {
            return usize::MAX;
        }
        // Rows 0 and 1 share the starting block size; row r (r >= 1) is
        // starting_block_size * 2^(r-1). The largest direct row reaches
        // max_direct_block_size, giving log2(max/start) + 2 direct rows.
        let ratio = (self.max_direct_block_size / self.starting_block_size).max(1);
        let log2 = 63 - ratio.leading_zeros() as usize;
        log2 + 2
    }

    /// Rows an indirect block needs to span `size` bytes of heap space:
    /// `log2(size) - log2(starting_block_size * table_width) + 1`, as
    /// libhdf5's `H5HF__dtable_size_to_rows`.
    fn rows_for_size(&self, size: u64) -> u16 {
        let log2 = |v: u64| 63u32.saturating_sub(v.max(1).leading_zeros());
        let first_row_bits = log2(self.starting_block_size) + log2(u64::from(self.table_width));
        (log2(size).saturating_sub(first_row_bits) + 1) as u16
    }

    /// Get block size for a given row in the doubling table.
    fn block_size_for_row(&self, row: usize) -> u64 {
        let sbs = self.starting_block_size;
        if row <= 1 {
            sbs
        } else {
            sbs.saturating_mul(1u64.checked_shl((row - 1) as u32).unwrap_or(u64::MAX))
        }
    }
}

/// A managed direct block's location, extent and (for a filtered heap) its
/// stored size and filter mask.
/// The child of an indirect block that covers a heap offset.
enum IndirectChild {
    Direct(DirectBlock),
    Indirect {
        addr: usize,
        nrows: u16,
        heap_offset: u64,
    },
}

struct DirectBlock {
    addr: usize,
    size: u64,
    heap_offset: u64,
    filtered_size: u64,
    filter_mask: u32,
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Build a minimal fractal heap with a single direct block at the root.
    /// Returns (file_data, FractalHeapHeader) where file_data contains
    /// the heap header at offset 0 and a direct block with known data.
    fn build_simple_heap(offset_size: u8, length_size: u8) -> (Vec<u8>, usize) {
        let os = offset_size as usize;
        let ls = length_size as usize;
        let max_heap_size: u16 = 16; // bits
        let block_offset_bytes = (max_heap_size as usize).div_ceil(8); // 2

        // Direct block at a known offset
        let dblock_offset = 256usize;
        let block_size: u64 = 128;

        // Build fractal heap header at offset 0
        let mut buf = vec![0u8; 1024];
        let mut pos = 0;
        buf[pos..pos + 4].copy_from_slice(b"FRHP");
        pos += 4;
        buf[pos] = 0; // version
        pos += 1;
        // heap_id_length = 7
        buf[pos..pos + 2].copy_from_slice(&7u16.to_le_bytes());
        pos += 2;
        // io_filter_encoded_length = 0
        buf[pos..pos + 2].copy_from_slice(&0u16.to_le_bytes());
        pos += 2;
        // flags = 0
        buf[pos] = 0;
        pos += 1;
        // max_managed_object_size
        buf[pos..pos + 4].copy_from_slice(&64u32.to_le_bytes());
        pos += 4;
        // next_huge_object_id (length_size)
        pos += ls;
        // btree_huge_objects_address (offset_size) - undefined
        for i in 0..os {
            buf[pos + i] = 0xFF;
        }
        pos += os;
        // free_space_managed_blocks (length_size)
        pos += ls;
        // managed_block_free_space_manager_address (offset_size) - undefined
        for i in 0..os {
            buf[pos + i] = 0xFF;
        }
        pos += os;
        // managed_space_in_heap (length_size)
        pos += ls;
        // allocated_managed_space_in_heap (length_size)
        pos += ls;
        // direct_block_allocation_iterator_offset (length_size)
        pos += ls;
        // managed_objects_count (length_size) = 1
        buf[pos] = 1;
        pos += ls;
        // huge_objects_size (length_size)
        pos += ls;
        // huge_objects_count (length_size)
        pos += ls;
        // tiny_objects_size (length_size)
        pos += ls;
        // tiny_objects_count (length_size)
        pos += ls;
        // table_width = 4
        buf[pos..pos + 2].copy_from_slice(&4u16.to_le_bytes());
        pos += 2;
        // starting_block_size (length_size)
        match length_size {
            4 => buf[pos..pos + 4].copy_from_slice(&(block_size as u32).to_le_bytes()),
            8 => buf[pos..pos + 8].copy_from_slice(&block_size.to_le_bytes()),
            _ => {}
        }
        pos += ls;
        // max_direct_block_size (length_size) = 1024
        match length_size {
            4 => buf[pos..pos + 4].copy_from_slice(&1024u32.to_le_bytes()),
            8 => buf[pos..pos + 8].copy_from_slice(&1024u64.to_le_bytes()),
            _ => {}
        }
        pos += ls;
        // max_heap_size (2) = 16
        buf[pos..pos + 2].copy_from_slice(&max_heap_size.to_le_bytes());
        pos += 2;
        // starting_row_of_indirect_blocks (2) = 2
        buf[pos..pos + 2].copy_from_slice(&2u16.to_le_bytes());
        pos += 2;
        // root_block_address (offset_size) = dblock_offset
        match offset_size {
            4 => buf[pos..pos + 4].copy_from_slice(&(dblock_offset as u32).to_le_bytes()),
            8 => buf[pos..pos + 8].copy_from_slice(&(dblock_offset as u64).to_le_bytes()),
            _ => {}
        }
        pos += os;
        // current_rows_in_root_indirect_block (2) = 0 (root is direct)
        buf[pos..pos + 2].copy_from_slice(&0u16.to_le_bytes());
        pos += 2;
        // checksum
        let checksum = crate::checksum::jenkins_lookup3(&buf[0..pos]);
        buf[pos..pos + 4].copy_from_slice(&checksum.to_le_bytes());
        pos += 4;
        let header_end = pos;

        // Build direct block at dblock_offset
        pos = dblock_offset;
        buf[pos..pos + 4].copy_from_slice(b"FHDB");
        pos += 4;
        buf[pos] = 0; // version
        pos += 1;
        // heap_header_address (offset_size) = 0
        pos += os;
        // block_offset (block_offset_bytes) = 0
        pos += block_offset_bytes;
        // Data starts here - write known pattern
        let data_start = pos;
        // Write "Hello, World!" at offset 0 in the data area
        let test_data = b"Hello, World!";
        buf[data_start..data_start + test_data.len()].copy_from_slice(test_data);

        (buf, header_end)
    }

    #[test]
    fn parse_header() {
        let (file_data, _) = build_simple_heap(8, 8);
        let hdr = FractalHeapHeader::parse(&file_data, 0, 8, 8).unwrap();
        assert_eq!(hdr.heap_id_length, 7);
        assert_eq!(hdr.io_filter_encoded_length, 0);
        assert_eq!(hdr.max_managed_object_size, 64);
        assert_eq!(hdr.table_width, 4);
        assert_eq!(hdr.starting_block_size, 128);
        assert_eq!(hdr.max_heap_size, 16);
        assert_eq!(hdr.current_rows_in_root_indirect_block, 0);
        assert_eq!(hdr.managed_objects_count, 1);
    }

    #[test]
    fn decode_managed_id() {
        let (file_data, _) = build_simple_heap(8, 8);
        let hdr = FractalHeapHeader::parse(&file_data, 0, 8, 8).unwrap();

        // Build a managed heap ID:
        // byte 0: version=0 (bits 6-7), type=0 (bits 4-5), reserved (bits 0-3)
        // bytes 1-6: offset (max_heap_size=16 bits) then length (remaining bits)
        // For offset=0, length=13:
        // payload = offset | (length << 16) = 0 | (13 << 16) = 0x000D0000
        let offset: u64 = 0;
        let length: u64 = 13;
        let payload = offset | (length << hdr.max_heap_size);
        let mut id = vec![0u8; 7];
        id[0] = 0x00; // type=0
        for i in 0..6 {
            id[1 + i] = ((payload >> (i * 8)) & 0xFF) as u8;
        }

        let (off, len) = hdr.decode_managed_id(&id).unwrap();
        assert_eq!(off, 0);
        assert_eq!(len, 13);
    }

    #[test]
    fn read_managed_object_from_direct_block() {
        let (file_data, _) = build_simple_heap(8, 8);
        let hdr = FractalHeapHeader::parse(&file_data, 0, 8, 8).unwrap();

        // Build heap ID for the test data written in build_simple_heap.
        // The test data "Hello, World!" is at the data area of the direct block.
        // The direct block header is 5 + 8 + 2 = 15 bytes (for max_heap_size=16, ceil(16/8)=2).
        // Wait, max_heap_size=16, ceil(16/8)=2. Header = sig(4)+ver(1)+addr(8)+bo(2) = 15.
        // The data was placed at data_start = block_addr + 15.
        // Since offset is from block start, the object is at offset 15 within the block.
        let dblock_header_size = 5 + 8 + (hdr.max_heap_size as usize).div_ceil(8); // 15
        let offset: u64 = dblock_header_size as u64;
        let length: u64 = 13;
        let payload = offset | (length << hdr.max_heap_size);
        let mut id = vec![0u8; 7];
        id[0] = 0x00;
        for i in 0..6 {
            id[1 + i] = ((payload >> (i * 8)) & 0xFF) as u8;
        }

        let obj = hdr.read_managed_object(&file_data, &id, 8).unwrap();
        assert_eq!(&obj, b"Hello, World!");
    }

    #[test]
    fn invalid_signature() {
        let mut data = vec![0u8; 128];
        data[0..4].copy_from_slice(b"XXXX");
        let err = FractalHeapHeader::parse(&data, 0, 8, 8).unwrap_err();
        assert_eq!(err, FormatError::InvalidFractalHeapSignature);
    }

    #[test]
    fn invalid_version() {
        let mut data = vec![0u8; 128];
        data[0..4].copy_from_slice(b"FRHP");
        data[4] = 1; // bad version
        let err = FractalHeapHeader::parse(&data, 0, 8, 8).unwrap_err();
        assert_eq!(err, FormatError::InvalidFractalHeapVersion(1));
    }

    #[test]
    fn invalid_heap_id_type() {
        let (file_data, _) = build_simple_heap(8, 8);
        let hdr = FractalHeapHeader::parse(&file_data, 0, 8, 8).unwrap();
        // Type = 1 (huge) in bits 4-5 is not a managed ID
        let id = vec![0x10u8, 0, 0, 0, 0, 0, 0];
        let err = hdr.decode_managed_id(&id).unwrap_err();
        assert_eq!(err, FormatError::InvalidHeapIdType(1));
    }

    #[test]
    fn tiny_object_is_read_from_the_id() {
        let (file_data, _) = build_simple_heap(8, 8);
        let hdr = FractalHeapHeader::parse(&file_data, 0, 8, 8).unwrap();
        // Type 2 (0x20), length - 1 in the low 4 bits, data after.
        let id = [0x20 | 2, b'a', b'b', b'c', 0, 0, 0];
        assert_eq!(hdr.read_managed_object(&file_data, &id, 8).unwrap(), b"abc");
        // A length running past the ID is an error, not a short read.
        let id = [0x20 | 9, b'a', b'b', b'c', 0, 0, 0];
        assert!(hdr.read_managed_object(&file_data, &id, 8).is_err());
    }

    #[test]
    fn huge_object_with_a_direct_id() {
        // With IDs long enough for an address and a length, libhdf5 stores
        // huge objects' location in the ID instead of the huge-object B-tree.
        let (mut file_data, _) = build_simple_heap(8, 8);
        let mut hdr = FractalHeapHeader::parse(&file_data, 0, 8, 8).unwrap();
        hdr.heap_id_length = 17;
        file_data[900..905].copy_from_slice(b"huge!");
        let mut id = vec![0x10u8];
        id.extend_from_slice(&900u64.to_le_bytes());
        id.extend_from_slice(&5u64.to_le_bytes());
        assert_eq!(
            hdr.read_managed_object(&file_data, &id, 8).unwrap(),
            b"huge!"
        );
    }

    #[test]
    fn unknown_heap_id_version_is_refused() {
        let (file_data, _) = build_simple_heap(8, 8);
        let hdr = FractalHeapHeader::parse(&file_data, 0, 8, 8).unwrap();
        let id = [0x40u8, 0, 0, 0, 0, 0, 0];
        assert!(hdr.read_managed_object(&file_data, &id, 8).is_err());
    }

    /// Headers, and managed (in a direct root and through an indirect
    /// root), huge and tiny objects read identically through a
    /// `read_at`-only storage, for every truncation of the file.
    #[test]
    fn storage_reads_match_slice_reads() {
        use crate::storage::CountingStorage;
        let (mut file, header_end) = build_simple_heap(8, 8);
        // An indirect root block at 600: row 0 holds the direct block at
        // 256, then three undefined blocks.
        file[600..604].copy_from_slice(b"FHIB");
        let mut at = 600 + 5 + 8 + 2;
        for addr in [256u64, u64::MAX, u64::MAX, u64::MAX] {
            file[at..at + 8].copy_from_slice(&addr.to_le_bytes());
            at += 8;
        }
        file[900..905].copy_from_slice(b"huge!");
        let managed_id = |offset: u64, len: u64| {
            let payload = offset | (len << 16);
            let mut id = vec![0u8];
            id.extend_from_slice(&payload.to_le_bytes()[..6]);
            id
        };
        let mut huge = vec![0x10u8];
        huge.extend_from_slice(&900u64.to_le_bytes());
        huge.extend_from_slice(&5u64.to_le_bytes());
        let ids = [
            managed_id(15, 13),
            managed_id(15, 200),
            managed_id(130, 4),
            huge,
            vec![0x22, b'a', b'b', b'c', 0, 0, 0],
        ];
        let mut cuts: Vec<usize> = (0..=header_end + 1).collect();
        cuts.extend([256, 260, 271, 280, 600, 610, 620, 640, 900, 903, file.len()]);
        for cut in cuts {
            let f = &file[..cut];
            let storage = CountingStorage::new(f.to_vec());
            let want = FractalHeapHeader::parse(f, 0, 8, 8);
            let got = FractalHeapHeader::parse_in(&storage, 0, 8, 8);
            assert_eq!(format!("{got:?}"), format!("{want:?}"), "cut {cut}");
            let Ok(direct) = want else { continue };
            let mut indirect = direct.clone();
            indirect.root_block_address = 600;
            indirect.current_rows_in_root_indirect_block = 1;
            let mut huge_ids = direct.clone();
            huge_ids.heap_id_length = 17;
            for hdr in [&direct, &indirect, &huge_ids] {
                for id in &ids {
                    assert_eq!(
                        hdr.read_managed_object_in(&storage, id, 8),
                        hdr.read_managed_object(f, id, 8),
                        "cut {cut}"
                    );
                }
            }
        }
    }

    /// A header claiming a huge doubling table (width 0xFFFF, 0xFFFF rows in
    /// the root indirect block) in a 16 MiB file: reading an object from the
    /// table's first block reads the entries up to it, not the rest of the
    /// file, and gives what the slice read gives. When the covering entry is
    /// unallocated the walk goes on over the whole block, still identically.
    #[test]
    fn huge_table_claims_read_only_what_the_walk_needs() {
        use crate::storage::CountingStorage;
        let (mut file, _) = build_simple_heap(8, 8);
        file.resize(16 << 20, 0);
        file[600..604].copy_from_slice(b"FHIB");
        let first_entry = 600 + 5 + 8 + 2;
        file[first_entry..first_entry + 8].copy_from_slice(&256u64.to_le_bytes());
        let mut hdr = FractalHeapHeader::parse(&file, 0, 8, 8).unwrap();
        hdr.table_width = 0xFFFF;
        hdr.root_block_address = 600;
        hdr.current_rows_in_root_indirect_block = 0xFFFF;
        let managed_id = |offset: u64, len: u64| {
            let payload = offset | (len << 16);
            let mut id = vec![0u8];
            id.extend_from_slice(&payload.to_le_bytes()[..6]);
            id
        };
        let storage = CountingStorage::new(file.clone());
        let id = managed_id(15, 13);
        let want = hdr.read_managed_object(&file, &id, 8);
        assert!(want.is_ok(), "{want:?}");
        storage.reset();
        assert_eq!(hdr.read_managed_object_in(&storage, &id, 8), want);
        assert!(
            storage.bytes_read() < 1024,
            "{} bytes in {} reads",
            storage.bytes_read(),
            storage.reads()
        );
        // The second entry (heap offsets 128..256) is unallocated (zero is
        // not the undefined address, so make it all ones).
        file[first_entry + 8..first_entry + 16].fill(0xFF);
        let storage = CountingStorage::new(file.clone());
        let id = managed_id(130, 4);
        let want = hdr.read_managed_object(&file, &id, 8);
        assert_eq!(hdr.read_managed_object_in(&storage, &id, 8), want);
    }

    /// A huge object found through the huge-object B-tree reads the
    /// B-tree through Storage: the same result (here an error, there is no
    /// B-tree at that address) as from the slice.
    #[test]
    fn huge_object_btree_reads_through_storage() {
        use crate::storage::CountingStorage;
        let (file, _) = build_simple_heap(8, 8);
        let mut hdr = FractalHeapHeader::parse(&file, 0, 8, 8).unwrap();
        hdr.huge_btree_address = 700;
        let id = [0x10, 1, 0, 0, 0, 0, 0];
        let want = hdr.read_managed_object(&file, &id, 8);
        assert!(want.is_err());
        let storage = CountingStorage::new(file);
        assert_eq!(hdr.read_managed_object_in(&storage, &id, 8), want);
    }

    /// A header with an I/O filter pipeline (read in a second, longer
    /// window) parses identically through a `read_at`-only storage, for
    /// every truncation.
    #[test]
    fn filtered_header_parses_identically_through_storage() {
        use crate::storage::CountingStorage;
        let (simple, header_end) = build_simple_heap(8, 8);
        let pipeline = [2u8, 1, 1, 0, 0, 0, 1, 0, 6, 0, 0, 0]; // deflate, level 6
        let mut header = simple[..header_end - 4].to_vec();
        header[7..9].copy_from_slice(&(pipeline.len() as u16).to_le_bytes());
        header.extend_from_slice(&100u64.to_le_bytes()); // root block's stored size
        header.extend_from_slice(&0u32.to_le_bytes()); // its filter mask
        header.extend_from_slice(&pipeline);
        let sum = crate::checksum::jenkins_lookup3(&header);
        header.extend_from_slice(&sum.to_le_bytes());
        let mut file = header.clone();
        file.resize(256, 0);
        let hdr = FractalHeapHeader::parse(&file, 0, 8, 8).unwrap();
        assert!(hdr.filter_pipeline.is_some());
        for cut in 0..=file.len() {
            let f = &file[..cut];
            let storage = CountingStorage::new(f.to_vec());
            assert_eq!(
                format!("{:?}", FractalHeapHeader::parse_in(&storage, 0, 8, 8)),
                format!("{:?}", FractalHeapHeader::parse(f, 0, 8, 8)),
                "cut {cut}"
            );
        }
    }
}