format: read group B-tree (v1) nodes over Storage

BTreeV1Node::parse_in reads a node's header and then its keys and
children, two bounded reads; collect_symbol_table_nodes_in walks the tree
over any Storage. The &[u8] functions are wrappers. New test: nodes with
siblings and 4- and 8-byte offsets cut at every length, and a two-level
tree with truncated leaves, give identical results through a read_at-only
CountingStorage (six reads for the three nodes).

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

crates/clawhdf5-format/src/btree_v1.rs

@@ -4,6 +4,7 @@
 use alloc::vec::Vec;
 
 use crate::error::FormatError;
+use crate::storage::{Storage, read_exact_at};
 
 /// A parsed B-tree v1 node.
 #[derive(Debug, Clone)]
@@ -74,13 +75,28 @@ impl BTreeV1Node {
         file_data: &[u8],
         offset: usize,
         offset_size: u8,
+        length_size: u8,
+    ) -> Result<BTreeV1Node, FormatError> {
+        Self::parse_in(&file_data, offset as u64, offset_size, length_size)
+    }
+
+    /// [`Self::parse`] over any [`Storage`]: one read of the node's header,
+    /// one of its keys and children.
+    pub fn parse_in(
+        file: &dyn Storage,
+        offset: u64,
+        offset_size: u8,
         _length_size: u8,
     ) -> Result<BTreeV1Node, FormatError> {
         // signature(4) + node_type(1) + node_level(1) + entries_used(2) = 8
         // + left_sibling(offset_size) + right_sibling(offset_size)
         let os = offset_size as usize;
         let header_size = 8 + os * 2;
-        ensure_len(file_data, offset, header_size)?;
+        let header = read_exact_at(file, offset, header_size)?;
+        let file_data: &[u8] = &header;
+        // The header's read checked that `offset + header_size` fits.
+        let body_start = offset + header_size as u64;
+        let offset = 0usize;
 
         if &file_data[offset..offset + 4] != b"TREE" {
             return Err(FormatError::InvalidBTreeSignature);
@@ -102,14 +118,15 @@ impl BTreeV1Node {
         } else {
             Some(read_offset(file_data, pos, offset_size)?)
         };
-        pos += os;
 
         // For type 0: keys are offset_size bytes, children are offset_size bytes
         // Layout: key[0], child[0], key[1], child[1], ..., key[N-1], child[N-1], key[N]
         let eu = entries_used as usize;
         let key_size = os; // For type 0, key = offset_size
         let needed = eu * (key_size + os) + key_size; // eu children + (eu+1) keys
-        ensure_len(file_data, pos, needed)?;
+        let body = read_exact_at(file, body_start, needed)?;
+        let file_data: &[u8] = &body;
+        let mut pos = 0usize;
 
         let mut keys = Vec::with_capacity(eu + 1);
         let mut children = Vec::with_capacity(eu);
@@ -150,11 +167,21 @@ pub fn collect_symbol_table_nodes(
     offset_size: u8,
     length_size: u8,
 ) -> Result<Vec<u64>, FormatError> {
-    collect_symbol_table_nodes_inner(file_data, btree_address, offset_size, length_size, 0)
+    collect_symbol_table_nodes_in(&file_data, btree_address, offset_size, length_size)
+}
+
+/// [`collect_symbol_table_nodes`] over any [`Storage`]: two reads per node.
+pub fn collect_symbol_table_nodes_in(
+    file: &dyn Storage,
+    btree_address: u64,
+    offset_size: u8,
+    length_size: u8,
+) -> Result<Vec<u64>, FormatError> {
+    collect_symbol_table_nodes_inner(file, btree_address, offset_size, length_size, 0)
 }
 
 fn collect_symbol_table_nodes_inner(
-    file_data: &[u8],
+    file: &dyn Storage,
     btree_address: u64,
     offset_size: u8,
     length_size: u8,
@@ -164,7 +191,12 @@ fn collect_symbol_table_nodes_inner(
         return Err(FormatError::NestingDepthExceeded);
     }
 
-    let node = BTreeV1Node::parse(file_data, btree_address as usize, offset_size, length_size)?;
+    let node = BTreeV1Node::parse_in(
+        file,
+        btree_address as usize as u64,
+        offset_size,
+        length_size,
+    )?;
 
     if node.node_type != 0 {
         return Err(FormatError::InvalidBTreeNodeType(node.node_type));
@@ -178,7 +210,7 @@ fn collect_symbol_table_nodes_inner(
         let mut result = Vec::new();
         for &child_addr in &node.children {
             let child_snods = collect_symbol_table_nodes_inner(
-                file_data,
+                file,
                 child_addr,
                 offset_size,
                 length_size,
@@ -317,4 +349,48 @@ mod tests {
         assert_eq!(node.entries_used, 1);
         assert_eq!(node.children, vec![0x50]);
     }
+
+    /// Nodes and trees, cut at every length, parse identically through a
+    /// `read_at`-only storage.
+    #[test]
+    fn storage_parse_matches_slice_parse() {
+        use crate::storage::CountingStorage;
+        let nodes = [
+            build_btree_node(0, 0, &[0, 5, 10], &[0x100, 0x200], None, None, 8),
+            build_btree_node(0, 0, &[0, 5], &[0x100], Some(0x40), Some(0x80), 4),
+            build_btree_node(1, 2, &[0, 5], &[0x100], None, Some(0x80), 8),
+        ];
+        for (n, node) in nodes.iter().enumerate() {
+            let os = if n == 1 { 4 } else { 8 };
+            for cut in 0..=node.len() {
+                let f = &node[..cut];
+                let storage = CountingStorage::new(f.to_vec());
+                let want = BTreeV1Node::parse(f, 0, os, 8);
+                let got = BTreeV1Node::parse_in(&storage, 0, os, 8);
+                assert_eq!(format!("{got:?}"), format!("{want:?}"));
+            }
+        }
+        let leaf1 = build_btree_node(0, 0, &[0, 5], &[0xA00], None, None, 8);
+        let leaf2 = build_btree_node(0, 0, &[5, 10], &[0xB00], None, None, 8);
+        let internal = build_btree_node(0, 1, &[0, 5, 10], &[0, 256], None, None, 8);
+        let mut file = vec![0u8; 512 + internal.len()];
+        file[..leaf1.len()].copy_from_slice(&leaf1);
+        file[256..256 + leaf2.len()].copy_from_slice(&leaf2);
+        file[512..].copy_from_slice(&internal);
+        for cut in [file.len(), 300, 260, 100, 10] {
+            let mut f = file.clone();
+            if cut < 512 {
+                // Truncate the leaves, keep the root.
+                f[cut..512].fill(0);
+            }
+            let storage = CountingStorage::new(f.clone());
+            assert_eq!(
+                collect_symbol_table_nodes_in(&storage, 512, 8, 8),
+                collect_symbol_table_nodes(&f, 512, 8, 8)
+            );
+        }
+        let storage = CountingStorage::new(file);
+        collect_symbol_table_nodes_in(&storage, 512, 8, 8).unwrap();
+        assert_eq!(storage.reads(), 6);
+    }
 }