format: resolve shared messages over Storage

Every shared-message entry point has an *_in(&dyn Storage, ..) core:
message_data(_with_sohm), resolve_shared_message(_with_sohm),
resolve_sohm_message, load_sohm_table, and the SMTB/SMLI parsers, which
read the signature and then all entries in one bounded read (the list's
open-ended &file_data[pos..] slice is gone). Object headers and the SOHM
fractal heap are read through their Storage parsers; a SOHM B-tree index
still needs the v2 B-tree over a slice, so over a backend without one it
is a clean ContiguousStorageRequired error. New signature::find_signature_in
probes the candidate offsets with 8-byte reads. The &[u8] functions are
wrappers.

New tests: SOHM tables and lists with 4- and 8-byte offsets, at two
offsets, cut at every length and with a bad signature, parse identically
through a read_at-only CountingStorage in at most two reads; the
signature search matches the slice search.

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

crates/clawhdf5-format/src/shared_message.rs

@@ -28,6 +28,7 @@ use crate::error::FormatError;
 use crate::fractal_heap::FractalHeapHeader;
 use crate::message_type::MessageType;
 use crate::object_header::ObjectHeader;
+use crate::storage::{Storage, Window, read_exact_at, require_contiguous};
 
 /// Fractal heap ID length for SOHM entries (fixed at 8 bytes).
 const FHEAP_ID_LEN: usize = 8;
@@ -253,17 +254,31 @@ pub fn parse_sohm_table(
     nindexes: u8,
     offset_size: u8,
 ) -> Result<SohmTable, FormatError> {
-    ensure_len(file_data, table_addr, 4)?;
+    parse_sohm_table_in(&file_data, table_addr as u64, nindexes, offset_size)
-    if &file_data[table_addr..table_addr + 4] != b"SMTB" {
+}
+
+/// [`parse_sohm_table`] over any [`Storage`]: one read of the signature,
+/// one of every index entry.
+pub fn parse_sohm_table_in(
+    file: &dyn Storage,
+    table_addr: u64,
+    nindexes: u8,
+    offset_size: u8,
+) -> Result<SohmTable, FormatError> {
+    let sig = read_exact_at(file, table_addr, 4)?;
+    if *sig != *b"SMTB" {
         return Err(FormatError::InvalidSohmTableSignature);
     }
-    let mut pos = table_addr + 4;
     let os = offset_size as usize;
     let entry_size = 1 + 1 + 2 + 4 + 2 + 2 + 2 + os + os; // 14 + 2*offset_size
+    // Positions below are relative to the table.
+    let w = Window::read(file, table_addr, 4 + nindexes as usize * entry_size)?;
+    let file_data: &[u8] = &w.bytes;
+    let mut pos = 4;
 
     let mut indexes = Vec::with_capacity(nindexes as usize);
     for _ in 0..nindexes {
-        ensure_len(file_data, pos, entry_size)?;
+        w.ensure(pos, entry_size)?;
         let version = file_data[pos];
         if version != 0 {
             return Err(FormatError::InvalidSohmTableVersion(version));
@@ -369,16 +384,29 @@ pub fn parse_sohm_list(
     num_messages: u16,
     offset_size: u8,
 ) -> Result<Vec<SohmEntry>, FormatError> {
-    ensure_len(file_data, list_addr, 4)?;
+    parse_sohm_list_in(&file_data, list_addr as u64, num_messages, offset_size)
-    if &file_data[list_addr..list_addr + 4] != b"SMLI" {
+}
+
+/// [`parse_sohm_list`] over any [`Storage`]: one read of the signature, one
+/// of every entry.
+pub fn parse_sohm_list_in(
+    file: &dyn Storage,
+    list_addr: u64,
+    num_messages: u16,
+    offset_size: u8,
+) -> Result<Vec<SohmEntry>, FormatError> {
+    let sig = read_exact_at(file, list_addr, 4)?;
+    if *sig != *b"SMLI" {
         return Err(FormatError::InvalidSohmListSignature);
     }
     let entry_sz = sohm_entry_size(offset_size);
-    let mut pos = list_addr + 4;
+    // Positions below are relative to the list.
+    let w = Window::read(file, list_addr, 4 + num_messages as usize * entry_sz)?;
+    let mut pos = 4;
     let mut entries = Vec::with_capacity(num_messages as usize);
     for _ in 0..num_messages {
-        ensure_len(file_data, pos, entry_sz)?;
+        w.ensure(pos, entry_sz)?;
-        let entry = parse_sohm_entry(&file_data[pos..], offset_size)?;
+        let entry = parse_sohm_entry(&w.bytes[pos..], offset_size)?;
         entries.push(entry);
         pos += entry_sz;
     }
@@ -392,6 +420,20 @@ pub fn parse_sohm_btree_entries(
     offset_size: u8,
     length_size: u8,
 ) -> Result<Vec<SohmEntry>, FormatError> {
+    parse_sohm_btree_entries_in(&file_data, btree_addr as u64, offset_size, length_size)
+}
+
+/// [`parse_sohm_btree_entries`] over any [`Storage`]. The v2 B-tree is not
+/// read over [`Storage`] yet, so this needs the whole file in memory
+/// ([`FormatError::ContiguousStorageRequired`] otherwise).
+pub fn parse_sohm_btree_entries_in(
+    file: &dyn Storage,
+    btree_addr: u64,
+    offset_size: u8,
+    length_size: u8,
+) -> Result<Vec<SohmEntry>, FormatError> {
+    let file_data = require_contiguous(file, "a shared-message B-tree index")?;
+    let btree_addr = usize::try_from(btree_addr).unwrap_or(usize::MAX);
     let header = BTreeV2Header::parse(file_data, btree_addr, offset_size, length_size)?;
     let records = collect_btree_v2_records(file_data, &header, offset_size, length_size)?;
     let mut entries = Vec::with_capacity(records.len());
@@ -413,15 +455,24 @@ pub fn load_sohm_table(
     offset_size: u8,
     length_size: u8,
 ) -> Result<Option<SohmTable>, FormatError> {
-    let sig = crate::signature::find_signature(file_data)?;
+    load_sohm_table_in(&file_data, offset_size, length_size)
-    let sb = crate::superblock::Superblock::parse(file_data, sig)?;
+}
+
+/// [`load_sohm_table`] over any [`Storage`].
+pub fn load_sohm_table_in(
+    file_data: &dyn Storage,
+    offset_size: u8,
+    length_size: u8,
+) -> Result<Option<SohmTable>, FormatError> {
+    let sig = crate::signature::find_signature_in(file_data)?;
+    let sb = crate::superblock::Superblock::parse_in(file_data, sig)?;
     let Some(ext_addr) = sb
         .superblock_extension_address
         .filter(|&a| !is_undefined(a, offset_size))
     else {
         return Ok(None);
     };
-    let ext = ObjectHeader::parse(file_data, ext_addr as usize, offset_size, length_size)?;
+    let ext = ObjectHeader::parse_in(file_data, ext_addr, offset_size, length_size)?;
     let Some(msg) = ext
         .messages
         .iter()
@@ -430,9 +481,9 @@ pub fn load_sohm_table(
         return Ok(None);
     };
     let table_msg = parse_sohm_table_message(&msg.data, offset_size)?;
-    parse_sohm_table(
+    parse_sohm_table_in(
         file_data,
-        table_msg.table_address as usize,
+        table_msg.table_address,
         table_msg.nindexes,
         offset_size,
     )
@@ -446,17 +497,27 @@ pub fn message_data_with_sohm<'a>(
     msg: &'a crate::object_header::HeaderMessage,
     offset_size: u8,
     length_size: u8,
+) -> Result<Cow<'a, [u8]>, FormatError> {
+    message_data_with_sohm_in(&file_data, msg, offset_size, length_size)
+}
+
+/// [`message_data_with_sohm`] over any [`Storage`].
+pub fn message_data_with_sohm_in<'a>(
+    file_data: &dyn Storage,
+    msg: &'a crate::object_header::HeaderMessage,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Cow<'a, [u8]>, FormatError> {
     if !is_shared(msg.flags) {
         return Ok(Cow::Borrowed(&msg.data));
     }
     let shared_ref = parse_shared_ref_sized(&msg.data, offset_size, length_size)?;
     let table = if shared_ref.heap_id.is_some() {
-        load_sohm_table(file_data, offset_size, length_size)?
+        load_sohm_table_in(file_data, offset_size, length_size)?
     } else {
         None
     };
-    resolve_shared_message_with_sohm(
+    resolve_shared_message_with_sohm_in(
         file_data,
         &shared_ref,
         msg.msg_type,
@@ -495,6 +556,25 @@ pub fn resolve_sohm_message(
     target_msg_type: MessageType,
     offset_size: u8,
     length_size: u8,
+) -> Result<Vec<u8>, FormatError> {
+    resolve_sohm_message_in(
+        &file_data,
+        heap_id,
+        sohm_table,
+        target_msg_type,
+        offset_size,
+        length_size,
+    )
+}
+
+/// [`resolve_sohm_message`] over any [`Storage`].
+pub fn resolve_sohm_message_in(
+    file_data: &dyn Storage,
+    heap_id: &[u8; FHEAP_ID_LEN],
+    sohm_table: &SohmTable,
+    target_msg_type: MessageType,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Vec<u8>, FormatError> {
     let index = find_index_for_msg_type(sohm_table, target_msg_type)
         .ok_or(FormatError::InvalidSharedMessageVersion(2))?;
@@ -503,13 +583,9 @@ pub fn resolve_sohm_message(
         return Err(FormatError::InvalidSharedMessageVersion(2));
     }
 
-    let fh_header = FractalHeapHeader::parse(
+    let fh_header =
-        file_data,
+        FractalHeapHeader::parse_in(file_data, index.heap_addr, offset_size, length_size)?;
-        index.heap_addr as usize,
+    fh_header.read_managed_object_in(file_data, heap_id, offset_size)
-        offset_size,
-        length_size,
-    )?;
-    fh_header.read_managed_object(file_data, heap_id, offset_size)
 }
 
 /// The payload of an object-header message, following the indirection if the
@@ -526,12 +602,22 @@ pub fn message_data<'a>(
     msg: &'a crate::object_header::HeaderMessage,
     offset_size: u8,
     length_size: u8,
+) -> Result<Cow<'a, [u8]>, FormatError> {
+    message_data_in(&file_data, msg, offset_size, length_size)
+}
+
+/// [`message_data`] over any [`Storage`].
+pub fn message_data_in<'a>(
+    file_data: &dyn Storage,
+    msg: &'a crate::object_header::HeaderMessage,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Cow<'a, [u8]>, FormatError> {
     if !is_shared(msg.flags) {
         return Ok(Cow::Borrowed(&msg.data));
     }
     let shared_ref = parse_shared_ref_sized(&msg.data, offset_size, length_size)?;
-    resolve_shared_message(
+    resolve_shared_message_in(
         file_data,
         &shared_ref,
         msg.msg_type,
@@ -553,13 +639,30 @@ pub fn resolve_shared_message(
     target_msg_type: MessageType,
     offset_size: u8,
     length_size: u8,
+) -> Result<Vec<u8>, FormatError> {
+    resolve_shared_message_in(
+        &file_data,
+        shared_ref,
+        target_msg_type,
+        offset_size,
+        length_size,
+    )
+}
+
+/// [`resolve_shared_message`] over any [`Storage`].
+pub fn resolve_shared_message_in(
+    file_data: &dyn Storage,
+    shared_ref: &SharedMessageRef,
+    target_msg_type: MessageType,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Vec<u8>, FormatError> {
     let table = if shared_ref.heap_id.is_some() {
-        load_sohm_table(file_data, offset_size, length_size)?
+        load_sohm_table_in(file_data, offset_size, length_size)?
     } else {
         None
     };
-    resolve_shared_message_with_sohm(
+    resolve_shared_message_with_sohm_in(
         file_data,
         shared_ref,
         target_msg_type,
@@ -577,6 +680,25 @@ pub fn resolve_shared_message_with_sohm(
     offset_size: u8,
     length_size: u8,
     sohm_table: Option<&SohmTable>,
+) -> Result<Vec<u8>, FormatError> {
+    resolve_shared_message_with_sohm_in(
+        &file_data,
+        shared_ref,
+        target_msg_type,
+        offset_size,
+        length_size,
+        sohm_table,
+    )
+}
+
+/// [`resolve_shared_message_with_sohm`] over any [`Storage`].
+pub fn resolve_shared_message_with_sohm_in(
+    file_data: &dyn Storage,
+    shared_ref: &SharedMessageRef,
+    target_msg_type: MessageType,
+    offset_size: u8,
+    length_size: u8,
+    sohm_table: Option<&SohmTable>,
 ) -> Result<Vec<u8>, FormatError> {
     // Dispatch on what the reference carries rather than on `ref_type`: v1/v2
     // references are always an object-header address whatever their type
@@ -586,8 +708,7 @@ pub fn resolve_shared_message_with_sohm(
         shared_ref.heap_id.as_ref(),
     ) {
         (Some(addr), _) => {
-            let target_header =
+            let target_header = ObjectHeader::parse_in(file_data, addr, offset_size, length_size)?;
-                ObjectHeader::parse(file_data, addr as usize, offset_size, length_size)?;
             for msg in &target_header.messages {
                 if msg.msg_type == target_msg_type && !is_shared(msg.flags) {
                     return Ok(msg.data.clone());
@@ -614,7 +735,7 @@ pub fn resolve_shared_message_with_sohm(
         }
         (None, Some(heap_id)) => {
             let table = sohm_table.ok_or(FormatError::InvalidSharedMessageVersion(2))?;
-            resolve_sohm_message(
+            resolve_sohm_message_in(
                 file_data,
                 heap_id,
                 table,
@@ -1052,4 +1173,81 @@ mod tests {
         // With 2-byte offsets: OH=2+2=4, heap=12, entry=1+4+12=17
         assert_eq!(sohm_entry_size(2), 17);
     }
+
+    /// SOHM tables and lists parse identically through a read_at-only
+    /// CountingStorage: at two offsets, with 4- and 8-byte offsets, cut at
+    /// every length and with a bad signature.
+    #[test]
+    fn storage_reads_match_slice_reads() {
+        use crate::storage::CountingStorage;
+        let idx = |t: u8, n: u16| SohmIndex {
+            index_type: t,
+            mesg_types: 0x0008,
+            min_mesg_size: 50,
+            list_max: 50,
+            btree_min: 40,
+            num_messages: n,
+            index_addr: 0x3000,
+            heap_addr: 0x4000,
+        };
+        let heap_entry = |h: u32| SohmEntry {
+            location: 0,
+            hash: h,
+            heap_id: Some([1, 2, 3, 4, 5, 6, 7, h as u8]),
+            ref_count: Some(h),
+            mesg_index: None,
+            oh_addr: None,
+        };
+        let oh_entry = SohmEntry {
+            location: 1,
+            hash: 9,
+            heap_id: None,
+            ref_count: None,
+            mesg_index: Some(3),
+            oh_addr: Some(0x7000),
+        };
+        let mut compared = 0;
+        for os in [4u8, 8] {
+            let smtb = build_smtb(&[idx(0, 2), idx(1, 7)], os);
+            let smli = build_smli(&[heap_entry(1), oh_entry.clone(), heap_entry(2)], os);
+            for (body, n) in [(smtb, 2u16), (smli, 3)] {
+                let is_table = &body[..4] == b"SMTB";
+                for at in [0usize, 0x40] {
+                    let mut full = vec![0u8; at];
+                    full.extend_from_slice(&body);
+                    let mut files = Vec::new();
+                    for cut in at..=full.len() {
+                        files.push(full[..cut].to_vec());
+                    }
+                    let mut bad = full.clone();
+                    bad[at] = b'X';
+                    files.push(bad);
+                    for f in files {
+                        let st = CountingStorage::new(f.clone());
+                        let (want, got) = if is_table {
+                            (
+                                format!("{:?}", parse_sohm_table(&f, at, n as u8, os)),
+                                format!("{:?}", parse_sohm_table_in(&st, at as u64, n as u8, os)),
+                            )
+                        } else {
+                            (
+                                format!("{:?}", parse_sohm_list(&f, at, n, os)),
+                                format!("{:?}", parse_sohm_list_in(&st, at as u64, n, os)),
+                            )
+                        };
+                        assert_eq!(got, want, "{} bytes", f.len());
+                        assert!(st.reads() <= 2);
+                        compared += 1;
+                    }
+                }
+            }
+        }
+        assert!(compared > 200);
+        // The B-tree index is not read over Storage yet: a clean error.
+        let st = CountingStorage::new(vec![0u8; 64]);
+        assert_eq!(
+            parse_sohm_btree_entries_in(&st, 0, 8, 8).unwrap_err(),
+            FormatError::ContiguousStorageRequired("a shared-message B-tree index")
+        );
+    }
 }

crates/clawhdf5-format/src/signature.rs

@@ -1,6 +1,7 @@
 //! HDF5 file signature (magic bytes) detection.
 
 use crate::error::FormatError;
+use crate::storage::{Storage, read_exact_at};
 
 /// The 8-byte HDF5 magic signature.
 pub const HDF5_SIGNATURE: [u8; 8] = [0x89, b'H', b'D', b'F', b'\r', b'\n', 0x1A, b'\n'];
@@ -39,6 +40,20 @@ pub fn find_signature(data: &[u8]) -> Result<usize, FormatError> {
     Err(FormatError::SignatureNotFound)
 }
 
+/// [`find_signature`] over any [`Storage`]: one 8-byte read per candidate
+/// offset.
+pub fn find_signature_in(file: &dyn Storage) -> Result<u64, FormatError> {
+    let len = file.len();
+    let mut offset = 0u64;
+    while offset.checked_add(8).is_some_and(|end| end <= len) {
+        if *read_exact_at(file, offset, 8)? == HDF5_SIGNATURE {
+            return Ok(offset);
+        }
+        offset = if offset == 0 { 512 } else { offset * 2 };
+    }
+    Err(FormatError::SignatureNotFound)
+}
+
 /// Split a file into its user block and its HDF5 bytes.
 ///
 /// Returns `(user_block, hdf5)`: `user_block` is everything before the
@@ -132,4 +147,26 @@ mod tests {
         data[512..520].copy_from_slice(&HDF5_SIGNATURE);
         assert_eq!(find_signature(&data), Ok(0));
     }
+
+    #[test]
+    fn find_signature_in_matches_slice_search() {
+        use crate::storage::CountingStorage;
+        for (len, at) in [
+            (0, None),
+            (7, None),
+            (8, Some(0)),
+            (600, Some(512)),
+            (5000, Some(4096)),
+            (3000, Some(2048)),
+            (3000, None),
+        ] {
+            let mut data = vec![0u8; len];
+            if let Some(at) = at {
+                data[at..at + 8].copy_from_slice(&HDF5_SIGNATURE);
+            }
+            let want = find_signature(&data).map(|o| o as u64);
+            let got = find_signature_in(&CountingStorage::new(data));
+            assert_eq!(got, want, "{len} {at:?}");
+        }
+    }
 }