clawhdf5/crates/clawhdf5-tools/src/check.rs

//! `h5rs check`: a structural validator.
//!
//! It walks every object reachable from the root group (and the superblock
//! extension), parses every header message, verifies the checksum of every
//! checksummed (version 2+) structure it meets, checks each chunked
//! dataset's chunk index against the dataset's shape, and checks that raw
//! data lies inside the file without overlapping other raw data. Every
//! problem is reported with the address of the structure involved.

use std::collections::{BTreeMap, HashSet};
use std::panic::{self, AssertUnwindSafe};

use clawhdf5_format::attribute_info::AttributeInfoMessage;
use clawhdf5_format::btree_v2::{BTreeV2Header, collect_btree_v2_records};
use clawhdf5_format::data_layout::DataLayout;
use clawhdf5_format::dataspace::{Dataspace, DataspaceType};
use clawhdf5_format::datatype::Datatype;
use clawhdf5_format::error::FormatError;
use clawhdf5_format::group_info::GroupInfoMessage;
use clawhdf5_format::link_info::LinkInfoMessage;
use clawhdf5_format::message_type::MessageType;
use clawhdf5_format::object_header::ObjectHeader;
use clawhdf5_format::symbol_table::SymbolTableMessage;
use clawhdf5_format::vl_data::{VlResolver, check_element_size, parse_vl_references};

use crate::cli::{Args, Out};
use crate::h5::{Error, ErrorKind, H5, Kind};
use crate::info::{self, DsInfo};

pub const USAGE: &str = "\
usage: h5rs check [--data] [-q] [--max-bytes N] FILE

Validate FILE's structure: walk every object from the root group, parse
every header message, verify the checksums of version 2+ structures
(superblock, object headers and continuation chunks, v2 B-tree nodes,
fractal heap headers and blocks, extensible/fixed array chunk indexes),
check each chunked dataset's chunk index against its shape (offsets aligned
to the chunk size and inside the extent, no duplicates, sizes plausible),
and check that all raw data lies inside the file without overlaps. Every
problem is printed with the address of the structure involved.

  --data          also read every dataset, decoding every chunk through its
                  filters (catches corrupt compressed data and Fletcher-32
                  mismatches), and follow every variable-length element of
                  datasets and attributes into its global heap collection
  -q, --quiet     print only the problems, not the summary
  --max-bytes N   largest dataset read by --data (default 1 GiB)

Exit status: 0 no problems, 1 problems found, 2 usage error or file not
found, 3 internal error.";

const MAX_CHUNKS_CHECKED: usize = 10_000_000;

/// A variable-length element's problem, worded as `check` reports heap
/// problems ("global heap ...").
fn heap_problem(e: FormatError) -> String {
    match e {
        FormatError::VlDataError(m) if m.starts_with("global heap") => m,
        e => format!("global heap: {e}"),
    }
}

/// Whether values of `dt` hold variable-length data (in the global heap).
fn has_vl(dt: &Datatype, depth: u32) -> bool {
    if depth > 32 {
        return false;
    }
    match dt {
        Datatype::VariableLength { .. } => true,
        Datatype::Compound { members, .. } => {
            members.iter().any(|m| has_vl(&m.datatype, depth + 1))
        }
        Datatype::Array { base_type, .. } => has_vl(base_type, depth + 1),
        _ => false,
    }
}

#[derive(Default)]
struct Counts {
    objects: u64,
    groups: u64,
    datasets: u64,
    datatypes: u64,
    messages: u64,
    chunks: u64,
    datasets_read: u64,
    /// Global heap collections that variable-length data points into,
    /// parsed without error (with --data).
    global_heaps: u64,
    sb_checksum: u64,
    ohdr_v2: u64,
    btree_v2: u64,
    heaps: u64,
    heap_block_checksums: u64,
    chunk_index_checksummed: u64,
}

struct Problem {
    addr: u64,
    path: String,
    msg: String,
}

struct Checker<'a> {
    h5: &'a H5,
    read_data: bool,
    eof: u64,
    problems: Vec<Problem>,
    /// Things not checked, which are not problems with the file.
    notes: Vec<Problem>,
    counts: Counts,
    /// Raw data extents: (start, end, owner path).
    extents: Vec<(u64, u64, String)>,
    heaps_seen: HashSet<u64>,
    btrees_seen: HashSet<u64>,
    /// Global heap collections already read (with --data).
    gcols_seen: HashSet<u64>,
    /// Resolves variable-length elements (with --data), for the whole file.
    vl: VlResolver<'a>,
    panicked: bool,
}

pub fn run(args: &mut Args, out: &mut Out) -> std::io::Result<i32> {
    let mut read_data = false;
    let mut quiet = false;
    let mut max_bytes = None;
    let mut file = None;
    while let Some(a) = args.next() {
        match a.as_str() {
            "--data" => read_data = true,
            "-q" | "--quiet" => quiet = true,
            "--max-bytes" => match args.number() {
                Some(n) => max_bytes = Some(n),
                None => return args.usage_error(out, "--max-bytes needs a number", USAGE),
            },
            "-h" | "--help" => {
                writeln!(out.o, "{USAGE}")?;
                return Ok(0);
            }
            s if s.starts_with('-') && s.len() > 1 => {
                return args.usage_error(out, &format!("unknown option {a}"), USAGE);
            }
            _ if file.is_none() => file = Some(a),
            _ => return args.usage_error(out, &format!("unexpected argument {a}"), USAGE),
        }
    }
    let Some(file) = file else {
        return args.usage_error(out, "missing FILE", USAGE);
    };
    let path = std::path::Path::new(&file);
    let mut h5 = if crate::h5::is_url(&file) {
        // check validates every byte, so a remote file is downloaded whole.
        match H5::open_arg_whole(&file) {
            Ok(h) => h,
            Err(e) => {
                writeln!(out.e, "h5rs check: {e}")?;
                return Ok(2);
            }
        }
    } else {
        if !path.is_file() {
            writeln!(out.e, "h5rs check: {file}: no such file")?;
            return Ok(2);
        }
        match H5::open(path) {
            Ok(h) => h,
            Err(_) => return unopenable(path, out),
        }
    };
    if let Some(m) = max_bytes {
        h5.max_bytes = m;
    }
    let mut c = Checker {
        h5: &h5,
        read_data,
        eof: h5.data().len() as u64,
        problems: Vec::new(),
        notes: Vec::new(),
        counts: Counts::default(),
        extents: Vec::new(),
        heaps_seen: HashSet::new(),
        btrees_seen: HashSet::new(),
        gcols_seen: HashSet::new(),
        vl: VlResolver::new(h5.data(), h5.os(), h5.ls()),
        panicked: false,
    };
    c.superblock();
    c.objects();
    c.overlaps();
    for p in &c.problems {
        writeln!(out.o, "problem: {:#x} {}: {}", p.addr, p.path, p.msg)?;
    }
    if !quiet {
        for n in &c.notes {
            writeln!(out.o, "note: {:#x} {}: {}", n.addr, n.path, n.msg)?;
        }
    }
    if !quiet {
        c.summary(&file, out)?;
    }
    Ok(if c.panicked {
        3
    } else if c.problems.is_empty() {
        0
    } else {
        1
    })
}

/// The file could not be opened at all: say why, as precisely as possible.
fn unopenable(path: &std::path::Path, out: &mut Out) -> std::io::Result<i32> {
    let data = match std::fs::read(path) {
        Ok(d) => d,
        Err(e) => {
            writeln!(out.e, "h5rs check: {}: {e}", path.display())?;
            return Ok(2);
        }
    };
    let msg = match clawhdf5_format::signature::find_signature(&data) {
        Err(_) => (
            0,
            "no HDF5 signature at offset 0 or any power of two from 512".to_string(),
        ),
        Ok(off) => match clawhdf5_format::superblock::Superblock::parse(&data[off..], 0) {
            Ok(sb) => match sb.data_end(off as u64, data.len() as u64) {
                Err(clawhdf5_format::error::FormatError::TruncatedFile {
                    stored_eof,
                    actual_len,
                }) => (
                    0,
                    format!(
                        "file is truncated: the superblock's end-of-file address is \
                         {stored_eof:#x} but the file is {actual_len:#x} bytes long"
                    ),
                ),
                // Say what the library refused, not just that it did.
                _ => match clawhdf5::File::from_bytes(data.clone()) {
                    Err(e) => (off as u64, format!("file cannot be opened: {e}")),
                    Ok(_) => (off as u64, "file cannot be opened".to_string()),
                },
            },
            Err(e) => (off as u64, format!("superblock: {e}")),
        },
    };
    writeln!(out.o, "problem: {:#x} /: {}", msg.0, msg.1)?;
    writeln!(
        out.o,
        "checked {}: 1 problem found (file cannot be opened)",
        path.display()
    )?;
    Ok(1)
}

fn msg_name(t: MessageType) -> String {
    match t {
        MessageType::Unknown(n) => format!("message type {n:#x}"),
        t => format!("{t:?} message"),
    }
}

impl Checker<'_> {
    fn problem(&mut self, addr: u64, path: &str, msg: impl Into<String>) {
        self.problems.push(Problem {
            addr,
            path: path.to_string(),
            msg: msg.into(),
        });
    }

    fn err(&mut self, default_addr: u64, path: &str, e: &Error) {
        // Reading attributes or links fails on a damaged heap or B-tree that
        // the structure checks have already reported at the same address.
        if let Some(a) = e.addr
            && self.problems.iter().any(|p| p.addr == a && p.path == path)
        {
            return;
        }
        self.problem(e.addr.unwrap_or(default_addr), path, e.msg.clone());
    }

    /// Run `f`; a panic inside it becomes a problem instead of aborting the
    /// whole check.
    fn guarded(&mut self, addr: u64, path: &str, f: impl FnOnce(&mut Self)) {
        let r = panic::catch_unwind(AssertUnwindSafe(|| f(self)));
        if r.is_err() {
            self.panicked = true;
            self.problem(
                addr,
                path,
                "internal error while checking this object (see stderr)",
            );
        }
    }

    fn superblock(&mut self) {
        let sb = self.h5.sb().clone();
        if sb.version >= 2 {
            // Superblock::parse verified it, or the file would not be open.
            self.counts.sb_checksum += 1;
        }
        // libhdf5 stores the end-of-file address counting the user block
        // (h5py's `userblock_size` files show it), unlike every other
        // address, which is relative to the superblock.
        let file_len = self.eof.saturating_add(self.h5.file.user_block_size());
        if sb.eof_address > file_len {
            self.problem(
                0,
                "/",
                format!(
                    "file is truncated: the superblock's end-of-file address is {:#x} but the file \
                     is {file_len:#x} bytes long",
                    sb.eof_address
                ),
            );
        }
        if sb.root_group_address >= self.eof {
            self.problem(
                0,
                "/",
                format!(
                    "root group address {:#x} is past the end of the file",
                    sb.root_group_address
                ),
            );
        }
        let undef = if sb.offset_size >= 8 {
            u64::MAX
        } else {
            (1u64 << (8 * u32::from(sb.offset_size))) - 1
        };
        if let Some(ext) = sb.superblock_extension_address.filter(|&a| a != undef) {
            self.guarded(ext, "<superblock extension>", |c| match c.h5.header(ext) {
                Ok(h) => {
                    if h.version >= 2 {
                        c.counts.ohdr_v2 += 1;
                    }
                    c.messages(ext, "<superblock extension>", &h);
                }
                Err(e) => c.err(ext, "<superblock extension>", &e),
            });
        }
    }

    fn objects(&mut self) {
        let h5 = self.h5;
        // Collect first, then check: the walk borrows the file, the checks
        // borrow `self` mutably.
        let (items, walk) = h5.walk_collect();
        if let Err(e) = walk {
            self.err(h5.root(), "/", &e);
        }
        for it in items {
            let path = it.path;
            // Soft, external and user-defined links have no header here.
            let (Some(addr), None, Some(header)) = (it.addr, it.first_path, it.header) else {
                continue;
            };
            if addr >= self.eof {
                self.problem(
                    addr,
                    &path,
                    "object header address is past the end of the file",
                );
                continue;
            }
            let h = match header {
                Ok(h) => h,
                Err(e) => {
                    self.err(addr, &path, &e);
                    continue;
                }
            };
            self.counts.objects += 1;
            self.guarded(addr, &path, |c| c.object(addr, &path, &h));
        }
    }

    fn object(&mut self, addr: u64, path: &str, h: &ObjectHeader) {
        if h.version >= 2 {
            self.counts.ohdr_v2 += 1;
        }
        self.messages(addr, path, h);
        match self.h5.attributes(h) {
            Ok((attrs, errs)) => {
                for e in errs {
                    self.problem(addr, path, format!("attribute: {e}"));
                }
                if self.read_data {
                    for a in &attrs {
                        let what = format!("attribute \"{}\": ", a.name);
                        self.vl_data(addr, path, &what, &a.datatype, &a.dataspace, &a.raw_data);
                    }
                }
            }
            Err(e) => self.err(addr, path, &e),
        }
        let is_root = addr == self.h5.root();
        match Kind::of(h) {
            Kind::Group => {
                self.counts.groups += 1;
                self.group(addr, path, h);
            }
            Kind::Dataset => {
                self.counts.datasets += 1;
                self.dataset(addr, path, h);
            }
            Kind::Datatype => self.counts.datatypes += 1,
            Kind::Unknown if is_root => self.counts.groups += 1,
            Kind::Unknown => {
                self.problem(
                    addr,
                    path,
                    "object header describes no group, dataset or named datatype",
                );
            }
        }
    }

    /// Parse every message of a header by type.
    fn messages(&mut self, addr: u64, path: &str, h: &ObjectHeader) {
        let (os, ls) = (self.h5.os(), self.h5.ls());
        for m in &h.messages {
            self.counts.messages += 1;
            let data =
                match clawhdf5_format::shared_message::message_data(self.h5.data(), m, os, ls) {
                    Ok(d) => d.into_owned(),
                    Err(e) => {
                        self.problem(addr, path, format!("shared {}: {e}", msg_name(m.msg_type)));
                        continue;
                    }
                };
            let r: Result<(), String> = match m.msg_type {
                MessageType::Dataspace => Dataspace::parse(&data, ls)
                    .map(drop)
                    .map_err(|e| e.to_string()),
                MessageType::Datatype => {
                    Datatype::parse(&data).map(drop).map_err(|e| e.to_string())
                }
                MessageType::FillValue | MessageType::FillValueOld => {
                    let mut mm = m.clone();
                    mm.data = data.clone();
                    clawhdf5_format::fill_value::parse_fill_value(&mm)
                        .map(drop)
                        .map_err(|e| e.to_string())
                }
                MessageType::DataLayout => DataLayout::parse(&data, os, ls)
                    .map(drop)
                    .map_err(|e| e.to_string()),
                MessageType::FilterPipeline => {
                    clawhdf5_format::filter_pipeline::FilterPipeline::parse(&data)
                        .map(drop)
                        .map_err(|e| e.to_string())
                }
                MessageType::Attribute => {
                    clawhdf5_format::attribute::AttributeMessage::parse_in_file(
                        &data,
                        self.h5.data(),
                        os,
                        ls,
                    )
                    .map(drop)
                    .map_err(|e| e.to_string())
                }
                MessageType::AttributeInfo => match AttributeInfoMessage::parse(&data, os) {
                    Ok(ai) => {
                        self.dense_storage(
                            addr,
                            path,
                            "attribute",
                            ai.fractal_heap_address,
                            [ai.btree_name_index_address, ai.btree_creation_order_address],
                        );
                        Ok(())
                    }
                    Err(e) => Err(e.to_string()),
                },
                MessageType::LinkInfo => LinkInfoMessage::parse(&data, os)
                    .map(drop)
                    .map_err(|e| e.to_string()),
                MessageType::Link => clawhdf5_format::link_message::LinkMessage::parse(&data, os)
                    .map(drop)
                    .or_else(|e| match e {
                        clawhdf5_format::error::FormatError::InvalidLinkType(t) if t >= 65 => {
                            Ok(())
                        }
                        e => Err(e.to_string()),
                    }),
                MessageType::GroupInfo => GroupInfoMessage::parse(&data)
                    .map(drop)
                    .map_err(|e| e.to_string()),
                MessageType::SymbolTable => SymbolTableMessage::parse(&data, os)
                    .map(drop)
                    .map_err(|e| e.to_string()),
                _ => Ok(()),
            };
            if let Err(e) = r {
                self.problem(addr, path, format!("{}: {e}", msg_name(m.msg_type)));
            }
        }
    }

    /// Dense (fractal heap + v2 B-tree) link or attribute storage.
    fn dense_storage(
        &mut self,
        addr: u64,
        path: &str,
        what: &str,
        heap: Option<u64>,
        btrees: [Option<u64>; 2],
    ) {
        if let Some(fh) = heap
            && self.heaps_seen.insert(fh)
        {
            self.counts.heaps += 1;
            let r = crate::heap_blocks::verify(self.h5, fh);
            self.counts.heap_block_checksums += r.checksums as u64;
            for e in r.problems {
                self.err(fh, path, &e.context(&format!("dense {what} storage")));
            }
        }
        for bt in btrees.into_iter().flatten() {
            if !self.btrees_seen.insert(bt) {
                continue;
            }
            let (os, ls) = (self.h5.os(), self.h5.ls());
            let Ok(off) = usize::try_from(bt) else {
                self.problem(bt, path, format!("{what} index address out of range"));
                continue;
            };
            match BTreeV2Header::parse(self.h5.data(), off, os, ls) {
                Ok(hdr) => {
                    self.counts.btree_v2 += 1;
                    match collect_btree_v2_records(self.h5.data(), &hdr, os, ls) {
                        Ok(recs) => {
                            if recs.len() as u64 != hdr.total_records {
                                self.problem(
                                    bt,
                                    path,
                                    format!(
                                        "{what} index: v2 B-tree header counts {} records, the tree holds {}",
                                        hdr.total_records,
                                        recs.len()
                                    ),
                                );
                            }
                        }
                        Err(e) => self.problem(bt, path, format!("{what} index (v2 B-tree): {e}")),
                    }
                }
                Err(e) => self.problem(bt, path, format!("{what} index (v2 B-tree header): {e}")),
            }
        }
        let _ = addr;
    }

    fn group(&mut self, addr: u64, path: &str, h: &ObjectHeader) {
        if let Some(m) = h
            .messages
            .iter()
            .find(|m| m.msg_type == MessageType::LinkInfo)
            && let Ok(li) = LinkInfoMessage::parse(&m.data, self.h5.os())
        {
            self.dense_storage(
                addr,
                path,
                "link",
                li.fractal_heap_address,
                [li.btree_name_index_address, li.btree_creation_order_address],
            );
        }
        if let Err(e) = self.h5.links(h) {
            self.err(addr, path, &e.context("cannot list the group"));
        }
        if let Some(m) = h
            .messages
            .iter()
            .find(|m| m.msg_type == MessageType::SymbolTable)
            && let Ok(st) = SymbolTableMessage::parse(&m.data, self.h5.os())
        {
            for (a, what) in [
                (st.btree_address, "B-tree"),
                (st.local_heap_address, "local heap"),
            ] {
                if a >= self.eof {
                    self.problem(
                        addr,
                        path,
                        format!("symbol table {what} address {a:#x} is past the end of the file"),
                    );
                }
            }
        }
    }

    fn dataset(&mut self, addr: u64, path: &str, h: &ObjectHeader) {
        let info = DsInfo::read(self.h5, path, h);
        for e in [
            info.dt.as_ref().err(),
            info.ds.as_ref().err(),
            info.layout.as_ref().err(),
            info.filters.as_ref().err(),
        ]
        .into_iter()
        .flatten()
        {
            let e = e.clone();
            self.err(addr, path, &e);
        }
        let (Ok(dt), Ok(ds), Ok(layout)) = (&info.dt, &info.ds, &info.layout) else {
            return;
        };
        let esize = u64::from(dt.type_size());
        if esize == 0 {
            self.problem(addr, path, "datatype has size 0");
            return;
        }
        let need = match crate::h5::byte_len(ds, dt) {
            Ok(n) => n,
            Err(e) => {
                self.err(addr, path, &e);
                return;
            }
        };
        if let Some(max) = &ds.max_dimensions {
            for (i, (&d, &m)) in ds.dimensions.iter().zip(max).enumerate() {
                if m != u64::MAX && d > m {
                    self.problem(
                        addr,
                        path,
                        format!("dimension {i} is {d}, above its maximum {m}"),
                    );
                }
            }
        }
        match layout {
            DataLayout::Compact { data } => {
                if (data.len() as u64) < need {
                    self.problem(
                        addr,
                        path,
                        format!(
                            "compact data holds {} bytes; the dataset needs {need}",
                            data.len()
                        ),
                    );
                }
            }
            DataLayout::Contiguous { address, size } => {
                if let Some(a) = address {
                    if !info.external && *size < need {
                        self.problem(
                            *a,
                            path,
                            format!("contiguous storage is {size} bytes; the dataset needs {need}"),
                        );
                    }
                    self.extent(*a, *size, path);
                }
            }
            DataLayout::Chunked { .. } => self.chunked(addr, path, &info, dt, ds, layout),
            DataLayout::Virtual { .. } => {
                let mut l = layout.clone();
                if let Err(e) = l.resolve_vds_mappings(self.h5.data(), self.h5.ls()) {
                    self.problem(addr, path, format!("virtual dataset mappings: {e}"));
                }
            }
        }
        if self.read_data {
            match self.h5.read_dataset(path, dt, ds) {
                Ok(raw) => {
                    self.counts.datasets_read += 1;
                    self.vl_data(addr, path, "", dt, ds, &raw);
                }
                // Valid data this tool cannot decode is not a problem with
                // the file.
                Err(e) if e.kind != ErrorKind::Corrupt => self.notes.push(Problem {
                    addr,
                    path: path.to_string(),
                    msg: format!("data not read: {}", e.msg),
                }),
                Err(e) => self.problem(addr, path, format!("reading the data: {}", e.msg)),
            }
        }
    }

    /// With --data: follow every variable-length element of `raw` (the
    /// values of a dataset or attribute) into the global heap, so a damaged
    /// collection, a missing heap object or a sequence longer than its heap
    /// object is reported at the collection's address. Each bad collection
    /// is reported once per object.
    fn vl_data(
        &mut self,
        addr: u64,
        path: &str,
        what: &str,
        dt: &Datatype,
        ds: &Dataspace,
        raw: &[u8],
    ) {
        if !has_vl(dt, 0) {
            return;
        }
        let n = crate::h5::num_elements(ds).unwrap_or(0);
        let size = dt.type_size() as usize;
        let mut bad: BTreeMap<u64, String> = BTreeMap::new();
        for i in 0..n {
            let Some(b) = usize::try_from(i)
                .ok()
                .and_then(|i| i.checked_mul(size))
                .and_then(|s| raw.get(s..s.checked_add(size)?))
            else {
                self.problem(
                    addr,
                    path,
                    format!("{what}element {i} is past the data read"),
                );
                break;
            };
            self.vl_element(dt, b, 0, &mut bad);
            if bad.len() >= 100 {
                break;
            }
        }
        for (a, msg) in bad {
            self.problem(a, path, format!("{what}variable-length data: {msg}"));
        }
    }

    fn vl_element(&mut self, dt: &Datatype, b: &[u8], depth: u32, bad: &mut BTreeMap<u64, String>) {
        if depth > 32 {
            return;
        }
        match dt {
            Datatype::VariableLength {
                size,
                is_string,
                base_type,
                ..
            } => {
                // Resolved by the library's VlResolver, as every other
                // reader resolves them (and as libhdf5 does): a heap object
                // whose size is not the element's length × base size, a
                // collection that overlaps another, or a missing object is
                // a problem at the collection's address.
                let Ok(vl) = parse_vl_references(b, 1, self.h5.os()) else {
                    return;
                };
                let gcol = vl[0].collection_address;
                if gcol == 0 || bad.contains_key(&gcol) {
                    return;
                }
                if let Err(e) = check_element_size(*size, self.h5.os()) {
                    bad.insert(gcol, e.to_string());
                    return;
                }
                let bs = if *is_string {
                    1
                } else {
                    base_type.type_size() as usize
                };
                if bs == 0 {
                    return;
                }
                let obj = match self.vl.element(b, bs) {
                    Ok(o) => o.unwrap_or(&[]),
                    Err(e) => {
                        bad.insert(gcol, heap_problem(e));
                        return;
                    }
                };
                if self.gcols_seen.insert(gcol) {
                    self.counts.global_heaps += 1;
                }
                if !*is_string && has_vl(base_type, depth + 1) {
                    for eb in obj.chunks_exact(bs) {
                        self.vl_element(base_type, eb, depth + 1, bad);
                    }
                }
            }
            Datatype::Compound { members, .. } => {
                for m in members {
                    if let Some(mb) = usize::try_from(m.byte_offset).ok().and_then(|o| b.get(o..)) {
                        self.vl_element(&m.datatype, mb, depth + 1, bad);
                    }
                }
            }
            Datatype::Array {
                base_type,
                dimensions,
            } => {
                let bs = base_type.type_size() as usize;
                let n = dimensions
                    .iter()
                    .try_fold(1usize, |a, &d| a.checked_mul(d as usize))
                    .unwrap_or(usize::MAX);
                for k in 0..n {
                    match b.get(k * bs..(k + 1) * bs) {
                        Some(eb) => self.vl_element(base_type, eb, depth + 1, bad),
                        None => break,
                    }
                }
            }
            _ => {}
        }
    }

    fn chunked(
        &mut self,
        addr: u64,
        path: &str,
        info: &DsInfo,
        dt: &Datatype,
        ds: &Dataspace,
        layout: &DataLayout,
    ) {
        let DataLayout::Chunked {
            chunk_dimensions,
            btree_address,
            chunk_index_type,
            ..
        } = layout
        else {
            return;
        };
        let rank = match ds.space_type {
            DataspaceType::Simple => ds.dimensions.len(),
            _ => 0,
        };
        if chunk_dimensions.len() != rank + 1 {
            self.problem(
                addr,
                path,
                format!(
                    "chunked layout has {} chunk dimensions for a rank-{rank} dataset (expected {})",
                    chunk_dimensions.len(),
                    rank + 1
                ),
            );
            return;
        }
        let cdims = &chunk_dimensions[..rank];
        if cdims.contains(&0) {
            self.problem(addr, path, "a chunk dimension is 0");
            return;
        }
        let Some(index_addr) = *btree_address else {
            return; // no chunk allocated yet
        };
        if index_addr >= self.eof {
            self.problem(
                addr,
                path,
                format!("chunk index address {index_addr:#x} is past the end of the file"),
            );
            return;
        }
        let chunks = match info::chunks(self.h5, layout, ds, dt) {
            Ok(c) => c,
            Err(e) => {
                self.err(index_addr, path, &e);
                return;
            }
        };
        if matches!(chunk_index_type, Some(3..=5)) {
            self.counts.chunk_index_checksummed += 1;
        }
        let filtered = matches!(&info.filters, Ok(Some(p)) if !p.filters.is_empty());
        let chunk_bytes = cdims.iter().try_fold(u64::from(dt.type_size()), |a, &d| {
            a.checked_mul(u64::from(d))
        });
        let max = ds.max_dimensions.clone();
        let mut seen: HashSet<Vec<u64>> = HashSet::with_capacity(chunks.len().min(1 << 20));
        let mut reported = 0usize;
        for (n, c) in chunks.iter().enumerate() {
            if n >= MAX_CHUNKS_CHECKED {
                self.problem(index_addr, path, "too many chunks; stopped checking them");
                break;
            }
            self.counts.chunks += 1;
            let mut bad = Vec::new();
            if c.offsets.len() < rank {
                bad.push(format!(
                    "has {} coordinates for a rank-{rank} dataset",
                    c.offsets.len()
                ));
            } else {
                for (i, (&o, &cd)) in c.offsets.iter().zip(cdims).enumerate() {
                    if o % u64::from(cd) != 0 {
                        bad.push(format!(
                            "offset {o} in dimension {i} is not a multiple of the chunk size {cd}"
                        ));
                    }
                    let limit = match max.as_ref().and_then(|m| m.get(i)) {
                        Some(&u64::MAX) | None => ds.dimensions[i],
                        Some(&m) => m.max(ds.dimensions[i]),
                    };
                    if o >= limit && !(limit == 0 && o == 0) {
                        bad.push(format!(
                            "offset {o} in dimension {i} is outside the extent {limit}"
                        ));
                    }
                }
                if !seen.insert(c.offsets[..rank].to_vec()) {
                    bad.push("appears twice in the chunk index".into());
                }
            }
            if c.chunk_size == 0 {
                bad.push("has size 0".into());
            } else if !filtered
                && let Some(cb) = chunk_bytes
                && u64::from(c.chunk_size) != cb
            {
                bad.push(format!(
                    "is {} bytes; an unfiltered chunk is {cb}",
                    c.chunk_size
                ));
            }
            if !bad.is_empty() {
                reported += 1;
                if reported <= 50 {
                    self.problem(
                        c.address,
                        path,
                        format!("chunk at {:?} {}", c.offsets, bad.join("; ")),
                    );
                }
            }
            self.extent(c.address, u64::from(c.chunk_size), path);
        }
        if reported > 50 {
            self.problem(
                index_addr,
                path,
                format!("{} more bad chunks not listed", reported - 50),
            );
        }
    }

    /// Record raw data at `[start, start + len)`, checking it is in the file.
    fn extent(&mut self, start: u64, len: u64, path: &str) {
        if len == 0 {
            return;
        }
        match start.checked_add(len) {
            Some(end) if end <= self.eof => self.extents.push((start, end, path.to_string())),
            _ => self.problem(
                start,
                path,
                format!("raw data ({len} bytes) extends past the end of the file"),
            ),
        }
    }

    fn overlaps(&mut self) {
        let mut ext = std::mem::take(&mut self.extents);
        ext.sort_unstable_by_key(|e| (e.0, e.1));
        let mut reported = 0usize;
        let mut far: Option<(u64, String)> = None;
        let mut msgs = Vec::new();
        for (s, e, p) in &ext {
            if let Some((end, owner)) = &far
                && s < end
            {
                reported += 1;
                if reported <= 50 {
                    msgs.push((
                        *s,
                        p.clone(),
                        format!("raw data at {s:#x} overlaps raw data of {owner}"),
                    ));
                }
            }
            if far.as_ref().is_none_or(|(end, _)| e > end) {
                far = Some((*e, p.clone()));
            }
        }
        for (a, p, m) in msgs {
            self.problem(a, &p, m);
        }
        if reported > 50 {
            self.problem(
                0,
                "/",
                format!("{} more raw data overlaps not listed", reported - 50),
            );
        }
    }

    fn summary(&self, file: &str, out: &mut Out) -> std::io::Result<()> {
        let c = &self.counts;
        let sb = self.h5.sb();
        writeln!(
            out.o,
            "checked {file}: superblock v{}, {} objects ({} groups, {} datasets, {} named datatypes), \
             {} header messages, {} chunks",
            sb.version, c.objects, c.groups, c.datasets, c.datatypes, c.messages, c.chunks
        )?;
        writeln!(
            out.o,
            "checksums verified: superblock {}, v2 object headers {}, v2 B-trees {}, \
             fractal heaps {} (+{} blocks), chunk indexes {}",
            c.sb_checksum,
            c.ohdr_v2,
            c.btree_v2,
            c.heaps,
            c.heap_block_checksums,
            c.chunk_index_checksummed
        )?;
        if self.read_data {
            writeln!(
                out.o,
                "datasets read: {}, global heap collections read: {}",
                c.datasets_read, c.global_heaps
            )?;
        }
        match self.problems.len() {
            0 => writeln!(out.o, "no problems found"),
            1 => writeln!(out.o, "1 problem found"),
            n => writeln!(out.o, "{n} problems found"),
        }
    }
}