use std::collections::BTreeMap;
use std::fs;
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::Path;
use crate::checksum::jenkins_lookup3;
use crate::dataspace::{Dataspace, DataspaceType};
use crate::error::Error;
use crate::file_lock::{self, FileLocking};
use crate::file_space_info::{FileSpaceInfo, FileSpaceStrategy};
use crate::file_writer::{LENGTH_SIZE, OFFSET_SIZE, build_dataset_oh, make_link};
use crate::free_space::FreeList;
use crate::free_space_manager::{self, FreeSection, FsmHeader, fshd_len, serialize_file_fsm};
use crate::group_v2::resolve_group_entries;
use crate::link_message::{LinkMessage, LinkTarget};
use crate::message_type::MessageType;
use crate::object_header::ObjectHeader;
use crate::signature;
use crate::superblock::Superblock;
use crate::type_builders::{AttrValue, DatasetBuilder, build_attr_message};
const UNDEF: u64 = u64::MAX;
const MAX_COMPACT_ATTRS: usize = 8;
const MAX_COPY_DEPTH: u32 = 1000;
const MAX_OH_CHUNKS: usize = 256;
type PathKey = Vec<String>;
pub struct EditSession {
handle: fs::File,
data: Vec<u8>,
sb_sig_off: usize,
superblock: Superblock,
pending_datasets: Vec<(PathKey, DatasetBuilder)>,
pending_groups: Vec<PathKey>,
pending_group_attrs: Vec<(PathKey, GroupAttrOp)>,
pending_deletes: Vec<PathKey>,
pending_copies: Vec<(PathKey, PathKey)>,
free: FreeList,
persist: Option<PersistState>,
}
struct PersistState {
strategy: FileSpaceStrategy,
threshold: u64,
page_size: u64,
old_blocks: Vec<(u64, u64)>,
}
impl EditSession {
pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, Error> {
Self::open_with_locking(path, FileLocking::Enabled)
}
pub fn open_with_locking<P: AsRef<Path>>(path: P, locking: FileLocking) -> Result<Self, Error> {
let path = path.as_ref();
let mut handle = fs::OpenOptions::new()
.read(true)
.write(true)
.open(path)
.map_err(Error::Io)?;
file_lock::acquire_exclusive(&handle, locking, path)?;
let mut data = Vec::new();
handle.read_to_end(&mut data).map_err(Error::Io)?;
let sb_sig_off = signature::find_signature(&data)?;
let superblock = Superblock::parse(&data, sb_sig_off)?;
if superblock.version > 3 {
return Err(Error::EditUnsupported("unsupported superblock version"));
}
if superblock.offset_size != OFFSET_SIZE || superblock.length_size != LENGTH_SIZE {
return Err(Error::EditUnsupported(
"only 8-byte offsets and lengths are supported for in-place editing",
));
}
if superblock.base_address != 0 {
return Err(Error::EditUnsupported(
"files with a userblock (non-zero base address) are not editable in place yet",
));
}
let mut session = Self {
handle,
data,
sb_sig_off,
superblock,
pending_datasets: Vec::new(),
pending_groups: Vec::new(),
pending_group_attrs: Vec::new(),
pending_deletes: Vec::new(),
pending_copies: Vec::new(),
free: FreeList::new(),
persist: None,
};
session.load_persisted_free_space();
Ok(session)
}
fn load_persisted_free_space(&mut self) {
if self.superblock.version < 2 {
return; }
let Some(ext_rel) = self.superblock.superblock_extension_address else {
return;
};
if ext_rel == UNDEF {
return;
}
let Ok(ext_addr) = usize::try_from(ext_rel) else {
return;
};
let Some(info) = self.extension_fsinfo(ext_addr) else {
return;
};
if !info.persist {
return;
}
let os = self.superblock.offset_size;
if let Ok(mut sections) =
free_space_manager::read_persisted_sections(&self.data, &info.manager_addrs, 0, os)
{
let file_len = self.data.len() as u64;
sections.sort_by_key(|s| s.addr);
let mut prev_end = 0u64;
for s in sections {
let Some(end) = s.addr.checked_add(s.size) else {
continue;
};
if s.size == 0 || end > file_len || s.addr < prev_end {
continue;
}
prev_end = end;
self.free.free(s.addr, s.size);
}
}
let mut old_blocks = Vec::new();
if let Ok(spans) = self.oh_chunk_spans(ext_addr) {
old_blocks.extend(spans);
}
for &m in &info.manager_addrs {
if m == UNDEF {
continue;
}
let Ok(m_us) = usize::try_from(m) else {
continue;
};
let Some(slice) = self.data.get(m_us..) else {
continue;
};
if let Ok(h) = FsmHeader::parse(slice, os) {
old_blocks.push((m, fshd_len(os)));
if h.fsse_addr != UNDEF
&& h.fsse_addr
.checked_add(h.fsse_used)
.is_some_and(|end| end <= self.data.len() as u64)
{
old_blocks.push((h.fsse_addr, h.fsse_used));
}
}
}
self.persist = Some(PersistState {
strategy: info.strategy,
threshold: info.threshold,
page_size: info.page_size,
old_blocks,
});
}
fn extension_fsinfo(&self, ext_addr: usize) -> Option<FileSpaceInfo> {
let os = self.superblock.offset_size;
let ls = self.superblock.length_size;
let base = self.superblock.base_address;
let oh = ObjectHeader::parse_with_base(&self.data, ext_addr, os, ls, base).ok()?;
let msg = oh
.messages
.iter()
.find(|m| m.msg_type == MessageType::FileSpaceInfo)?;
FileSpaceInfo::parse(&msg.data, os, ls).ok()
}
pub fn create_dataset(&mut self, path: &str) -> &mut DatasetBuilder {
let mut comps = split_path(path);
let leaf = comps.pop().unwrap_or_default();
self.pending_datasets
.push((comps, DatasetBuilder::new(&leaf)));
&mut self.pending_datasets.last_mut().unwrap().1
}
pub fn create_group(&mut self, path: &str) {
self.pending_groups.push(split_path(path));
}
pub fn set_group_attr(&mut self, path: &str, name: &str, value: AttrValue) -> &mut Self {
self.pending_group_attrs.push((
split_path(path),
GroupAttrOp::Set {
name: name.to_string(),
value,
},
));
self
}
pub fn remove_group_attr(&mut self, path: &str, name: &str) -> &mut Self {
self.pending_group_attrs.push((
split_path(path),
GroupAttrOp::Remove {
name: name.to_string(),
},
));
self
}
pub fn delete(&mut self, path: &str) {
self.pending_deletes.push(split_path(path));
}
pub fn copy(&mut self, src: &str, dst: &str) {
self.pending_copies.push((split_path(src), split_path(dst)));
}
pub fn commit(&mut self) -> Result<(), Error> {
if self.pending_datasets.is_empty()
&& self.pending_groups.is_empty()
&& self.pending_group_attrs.is_empty()
&& self.pending_deletes.is_empty()
&& self.pending_copies.is_empty()
{
return Ok(());
}
let mut nodes: BTreeMap<PathKey, Node> = BTreeMap::new();
nodes.entry(PathKey::new()).or_default(); let mut add_targets: Vec<PathKey> = Vec::new();
let mut attr_targets: Vec<PathKey> = Vec::new();
for path in std::mem::take(&mut self.pending_groups) {
if path.is_empty() {
return Err(Error::EditUnsupported("cannot create the root group"));
}
ensure_ancestors(&mut nodes, &path);
nodes.entry(path.clone()).or_default().is_new = true;
add_targets.push(path);
}
for (parent, db) in std::mem::take(&mut self.pending_datasets) {
let mut full = parent.clone();
full.push(db.name.clone());
add_targets.push(full);
ensure_ancestors(&mut nodes, &parent);
nodes.entry(parent).or_default().datasets.push(db);
}
for (path, op) in std::mem::take(&mut self.pending_group_attrs) {
ensure_ancestors(&mut nodes, &path);
nodes.entry(path.clone()).or_default().attr_ops.push(op);
attr_targets.push(path);
}
for (src, dst) in std::mem::take(&mut self.pending_copies) {
if src.is_empty() {
return Err(Error::EditUnsupported("cannot copy the root group"));
}
if dst.is_empty() {
return Err(Error::EditUnsupported("copy destination path is empty"));
}
if is_prefix(&src, &dst) {
return Err(Error::EditUnsupported(
"cannot copy an object into itself or its own subtree",
));
}
let src_str = src.join("/");
let src_addr =
crate::group_v2::resolve_path_any(&self.data, &self.superblock, &src_str)
.map_err(|_| Error::EditUnsupported("copy source does not exist"))?;
let src_addr = usize::try_from(src_addr)
.map_err(|_| Error::EditUnsupported("source address exceeds this platform"))?;
self.plan_copy(src_addr, 0)?;
add_targets.push(dst.clone());
let leaf = dst.last().unwrap().clone();
let parent = dst[..dst.len() - 1].to_vec();
ensure_ancestors(&mut nodes, &parent);
nodes
.entry(parent)
.or_default()
.copies
.push((leaf, src_addr as u64));
}
let deletes = std::mem::take(&mut self.pending_deletes);
let mut deleted_addrs: Vec<usize> = Vec::new();
for (i, d) in deletes.iter().enumerate() {
if d.is_empty() {
return Err(Error::EditUnsupported("cannot delete the root group"));
}
let path_str = d.join("/");
let del_addr =
crate::group_v2::resolve_path_any(&self.data, &self.superblock, &path_str)
.map_err(|_| Error::EditUnsupported("nothing to delete at the given path"))?;
if let Ok(a) = usize::try_from(del_addr) {
deleted_addrs.push(a);
}
for t in &add_targets {
if is_prefix(d, t) || is_prefix(t, d) {
return Err(Error::EditUnsupported(
"a deletion overlaps an addition in the same commit; use separate commits",
));
}
}
for t in &attr_targets {
if is_prefix(d, t) {
return Err(Error::EditUnsupported(
"a deletion overlaps a group-attribute edit in the same commit; use separate commits",
));
}
}
for (j, d2) in deletes.iter().enumerate() {
if i != j && is_prefix(d, d2) {
return Err(Error::EditUnsupported(
"overlapping deletions in one commit; delete the common parent only",
));
}
}
let parent = d[..d.len() - 1].to_vec();
ensure_ancestors(&mut nodes, &parent);
nodes
.entry(parent)
.or_default()
.deletes
.push(d.last().unwrap().clone());
}
let keys: Vec<PathKey> = nodes.keys().cloned().collect();
let mut superseded_addrs: Vec<usize> = Vec::new();
for key in &keys {
let is_new = nodes[key].is_new;
if is_new {
nodes.get_mut(key).unwrap().base_region = fresh_group_region();
} else {
let path_str = key.join("/");
let addr =
crate::group_v2::resolve_path_any(&self.data, &self.superblock, &path_str)
.map_err(|_| {
Error::EditUnsupported(
"a target group does not exist; create it first in this session",
)
})?;
let addr = usize::try_from(addr)
.map_err(|_| Error::EditUnsupported("group address exceeds this platform"))?;
let info = self.inspect_group(addr)?;
superseded_addrs.push(addr);
let node = nodes.get_mut(key).unwrap();
node.base_region = info.region;
node.existing_links = info.link_names;
}
}
for key in &keys {
let node = nodes.get_mut(key).unwrap();
let ops = std::mem::take(&mut node.attr_ops);
if !ops.is_empty() {
let region = std::mem::take(&mut node.base_region);
node.base_region = apply_group_attr_ops(®ion, &ops)?;
}
}
let mut children: BTreeMap<PathKey, Vec<PathKey>> = BTreeMap::new();
for key in &keys {
if !key.is_empty() {
let parent = key[..key.len() - 1].to_vec();
children.entry(parent).or_default().push(key.clone());
}
}
for key in &keys {
let node = &nodes[key];
let mut adding: Vec<&str> = Vec::new();
for db in &node.datasets {
adding.push(&db.name);
}
for child in children.get(key).into_iter().flatten() {
if nodes[child].is_new {
adding.push(child.last().unwrap());
}
}
for (leaf, _) in &node.copies {
adding.push(leaf);
}
for (i, name) in adding.iter().enumerate() {
if node.existing_links.iter().any(|n| n == name) || adding[..i].contains(name) {
return Err(Error::EditUnsupported(
"a link with this name already exists in the target group",
));
}
}
}
let mut flat: BTreeMap<PathKey, Vec<FlatDataset>> = BTreeMap::new();
for key in &keys {
let dbs = std::mem::take(&mut nodes.get_mut(key).unwrap().datasets);
let mut v = Vec::with_capacity(dbs.len());
for db in dbs {
v.push(flatten_dataset(db)?);
}
flat.insert(key.clone(), v);
}
let mut to_free: Vec<(u64, u64)> = Vec::new();
for &a in &deleted_addrs {
self.collect_free_spans(a, 0, &mut to_free);
}
for &a in &superseded_addrs {
if let Ok(spans) = self.oh_chunk_spans(a) {
to_free.extend(spans);
}
}
let mut new_addr: BTreeMap<PathKey, u64> = BTreeMap::new();
let mut by_depth = keys.clone();
by_depth.sort_by_key(|k| std::cmp::Reverse(k.len())); for key in &by_depth {
let (mut region, deletes, copies) = {
let node = nodes.get_mut(key).unwrap();
(
std::mem::take(&mut node.base_region),
std::mem::take(&mut node.deletes),
std::mem::take(&mut node.copies),
)
};
for name in &deletes {
region = remove_link_from_region(®ion, name)?;
}
for (leaf, src_addr) in copies {
#[expect(
clippy::cast_possible_truncation,
reason = "src_addr is an offset into self.data, the in-memory file image \
(a Vec<u8>), so it always fits usize on the running target"
)]
let root = self.perform_copy(src_addr as usize, 0)?;
region.extend_from_slice(&encode_link_message(&leaf, root));
}
for fd in flat.remove(key).into_iter().flatten() {
let data_addr = self.alloc_or_append(&fd.raw)?;
let oh = build_dataset_oh(
&fd.dt,
&fd.ds,
data_addr,
fd.raw.len() as u64,
&fd.attrs,
None,
);
let oh_addr = self.alloc_or_append(&oh)?;
region.extend_from_slice(&encode_link_message(&fd.name, oh_addr));
}
for child in children.get(key).into_iter().flatten() {
let child_name = child.last().unwrap();
let child_addr = new_addr[child];
if nodes[child].is_new {
region.extend_from_slice(&encode_link_message(child_name, child_addr));
} else {
patch_link_target(&mut region, child_name, child_addr)?;
}
}
let oh = build_v2_object_header(®ion);
let addr = self.alloc_or_append(&oh)?;
new_addr.insert(key.clone(), addr);
}
let new_root = new_addr[&PathKey::new()];
if self.persist.is_some() {
return self.commit_persisting(new_root, to_free);
}
for (a, l) in to_free.drain(..) {
self.free.free(a, l);
}
let cur_eof = self.data.len() as u64;
let trunc_to = self.free.take_trailing(cur_eof);
let new_eof = trunc_to.unwrap_or(cur_eof);
self.handle.sync_all().map_err(Error::Io)?;
if self.superblock.version >= 2 {
let mut new_sb = self.superblock.clone();
new_sb.root_group_address = new_root;
new_sb.eof_address = new_eof;
new_sb.consistency_flags = 0;
let sb_bytes = new_sb.serialize();
self.write_at(self.sb_sig_off, &sb_bytes)?;
self.handle.sync_all().map_err(Error::Io)?;
self.superblock = new_sb;
} else {
self.repoint_v0v1_root(new_root, new_eof)?;
self.handle.sync_all().map_err(Error::Io)?;
self.superblock.root_group_address = new_root;
self.superblock.eof_address = new_eof;
}
if let Some(cut) = trunc_to {
self.handle.set_len(cut).map_err(Error::Io)?;
#[expect(
clippy::cast_possible_truncation,
reason = "cut is a shrink target <= the current file length, which equals \
self.data.len() (a usize)"
)]
self.data.truncate(cut as usize);
self.handle.sync_all().map_err(Error::Io)?;
}
Ok(())
}
fn commit_persisting(&mut self, new_root: u64, to_free: Vec<(u64, u64)>) -> Result<(), Error> {
let os = self.superblock.offset_size;
let (strategy, threshold, page_size, old_blocks) = {
let ps = self
.persist
.as_ref()
.expect("commit_persisting is only called when persistence is armed");
(
ps.strategy,
ps.threshold,
ps.page_size,
ps.old_blocks.clone(),
)
};
let mut post = self.free.clone();
for &(a, l) in &to_free {
post.free(a, l);
}
for &(a, l) in &old_blocks {
post.free(a, l);
}
let sections: Vec<FreeSection> = post
.sections()
.into_iter()
.map(|(addr, size)| FreeSection { addr, size })
.collect();
let old_ext_rel = self
.superblock
.superblock_extension_address
.filter(|&a| a != UNDEF)
.ok_or(Error::EditUnsupported(
"a persisting file has no superblock extension to update",
))?;
let old_ext_addr = usize::try_from(old_ext_rel)
.map_err(|_| Error::EditUnsupported("extension address exceeds this platform"))?;
let placeholder =
FileSpaceInfo::persistent_single_manager(strategy, threshold, page_size, 0, 0);
let ext_len =
build_v2_object_header(&self.rewrite_extension_region(old_ext_addr, &placeholder)?)
.len() as u64;
let ext_addr = self.data.len() as u64;
let fshd_addr = ext_addr + ext_len;
let (ext_oh, fsm_blocks, final_eof) = if sections.is_empty() {
let info = FileSpaceInfo::persistent_empty(strategy, threshold, page_size);
let ext_oh =
build_v2_object_header(&self.rewrite_extension_region(old_ext_addr, &info)?);
let final_eof = ext_addr + ext_oh.len() as u64;
(ext_oh, None, final_eof)
} else {
let fsse_addr = fshd_addr + fshd_len(os);
let eoa_pre_fsm = fshd_addr;
let info = FileSpaceInfo::persistent_single_manager(
strategy,
threshold,
page_size,
fshd_addr,
eoa_pre_fsm,
);
let ext_oh =
build_v2_object_header(&self.rewrite_extension_region(old_ext_addr, &info)?);
debug_assert_eq!(
ext_oh.len() as u64,
ext_len,
"extension length must be stable across the placeholder and real messages"
);
let (fshd, fsse) = serialize_file_fsm(§ions, fshd_addr, fsse_addr, os);
let final_eof = fsse_addr + fsse.len() as u64;
(ext_oh, Some((fshd, fsse)), final_eof)
};
let written_ext = self.append(&ext_oh)?;
debug_assert_eq!(written_ext, ext_addr);
let mut new_old_blocks = vec![(ext_addr, ext_oh.len() as u64)];
if let Some((fshd, fsse)) = fsm_blocks {
let wf = self.append(&fshd)?;
debug_assert_eq!(wf, fshd_addr);
new_old_blocks.push((fshd_addr, fshd.len() as u64));
let ws = self.append(&fsse)?;
new_old_blocks.push((ws, fsse.len() as u64));
}
self.handle.sync_all().map_err(Error::Io)?;
let mut new_sb = self.superblock.clone();
new_sb.root_group_address = new_root;
new_sb.eof_address = final_eof;
new_sb.superblock_extension_address = Some(ext_addr);
new_sb.consistency_flags = 0;
let sb_bytes = new_sb.serialize();
self.write_at(self.sb_sig_off, &sb_bytes)?;
self.handle.sync_all().map_err(Error::Io)?;
self.superblock = new_sb;
self.free = post;
self.persist = Some(PersistState {
strategy,
threshold,
page_size,
old_blocks: new_old_blocks,
});
Ok(())
}
fn rewrite_extension_region(
&self,
ext_addr: usize,
info: &FileSpaceInfo,
) -> Result<Vec<u8>, Error> {
let region = self.gather_oh_messages(ext_addr)?;
let new_body = info.serialize();
let new_len = u16::try_from(new_body.len())
.map_err(|_| Error::EditUnsupported("File Space Info message too large"))?;
let mut out = Vec::with_capacity(region.len());
let mut p = 0;
let mut replaced = false;
while let Some((msg_type, _body, body_end)) = next_message(®ion, p)? {
if msg_type == MessageType::FileSpaceInfo {
out.push(region[p]); out.extend_from_slice(&new_len.to_le_bytes());
out.push(region[p + 3]); out.extend_from_slice(&new_body);
replaced = true;
} else {
out.extend_from_slice(®ion[p..body_end]);
}
p = body_end;
}
if !replaced {
return Err(Error::EditUnsupported(
"a persisting file's superblock extension has no File Space Info message",
));
}
Ok(out)
}
fn repoint_v0v1_root(&mut self, new_root: u64, new_eof: u64) -> Result<(), Error> {
let os = self.superblock.offset_size as usize;
let var_start = if self.superblock.version == 0 { 24 } else { 28 };
let base = self.sb_sig_off + var_start;
let eof_off = base + 2 * os;
let ste = base + 4 * os;
let oh_addr_off = ste + os;
let cache_off = ste + 2 * os;
self.write_at(eof_off, &new_eof.to_le_bytes()[..os])?;
self.write_at(cache_off, &[0u8; 4])?; self.write_at(cache_off + 8, &[0u8; 16])?; self.write_at(oh_addr_off, &new_root.to_le_bytes()[..os])?;
Ok(())
}
fn oh_region(&self, addr: usize) -> Result<(usize, usize), Error> {
let d = &self.data;
if d.len() < addr + 6 || &d[addr..addr + 4] != b"OHDR" || d[addr + 4] != 2 {
return Err(Error::EditUnsupported(
"an object does not use a version 2 object header",
));
}
let flags = d[addr + 5];
if flags & 0x04 != 0 {
return Err(Error::EditUnsupported(
"an object tracks message creation order (not supported in place yet)",
));
}
let mut pos = addr + 6;
if flags & 0x20 != 0 {
pos += 16; }
if flags & 0x10 != 0 {
pos += 4; }
let size_width = match flags & 0x03 {
0 => 1usize,
1 => 2,
2 => 4,
_ => 8,
};
if d.len() < pos + size_width {
return Err(Error::EditUnsupported("truncated object header"));
}
let chunk0_size = read_le(&d[pos..pos + size_width]);
pos += size_width;
let region_start = pos;
let region_end = region_start
.checked_add(chunk0_size)
.filter(|&e| e + 4 <= d.len())
.ok_or(Error::EditUnsupported("truncated object header"))?;
Ok((region_start, region_end))
}
fn gather_oh_messages(&self, addr: usize) -> Result<Vec<u8>, Error> {
let (rs, re) = self.oh_region(addr)?;
let d = &self.data;
let mut out = Vec::new();
let mut chunks: Vec<(usize, usize)> = vec![(rs, re)];
let mut i = 0;
while i < chunks.len() {
if chunks.len() > MAX_OH_CHUNKS {
return Err(Error::EditUnsupported(
"object header has too many continuation chunks",
));
}
let (cs, ce) = chunks[i];
i += 1;
let region = &d[..ce];
let mut p = cs;
while let Some((msg_type, body, body_end)) = next_message(region, p)? {
if msg_type == MessageType::ObjectHeaderContinuation {
if body_end - body < (OFFSET_SIZE + LENGTH_SIZE) as usize {
return Err(Error::EditUnsupported("malformed continuation message"));
}
let off = u64::from_le_bytes(d[body..body + 8].try_into().unwrap());
let len = u64::from_le_bytes(d[body + 8..body + 16].try_into().unwrap());
let off = usize::try_from(off).map_err(|_| {
Error::EditUnsupported("continuation address exceeds this platform")
})?;
let len = usize::try_from(len).map_err(|_| {
Error::EditUnsupported("continuation length exceeds this platform")
})?;
let blk_end = off
.checked_add(len)
.filter(|&e| e <= d.len() && len >= 8)
.ok_or(Error::EditUnsupported("continuation block out of bounds"))?;
if &d[off..off + 4] != b"OCHK" {
return Err(Error::EditUnsupported(
"invalid continuation block signature",
));
}
chunks.push((off + 4, blk_end - 4));
} else {
out.extend_from_slice(®ion[p..body_end]);
}
p = body_end;
}
}
Ok(out)
}
fn reconstruct_v1_group(&self, addr: usize) -> Result<GroupInfo, Error> {
let os = self.superblock.offset_size;
let ls = self.superblock.length_size;
let base = self.superblock.base_address;
let oh = ObjectHeader::parse_with_base(&self.data, addr, os, ls, base)?;
if oh
.messages
.iter()
.any(|m| m.msg_type == MessageType::DataLayout)
{
return Err(Error::EditUnsupported(
"a target path names a dataset, not a group",
));
}
let entries = resolve_group_entries(&self.data, &oh, os, ls, base)?;
let mut region = fresh_group_region();
let mut link_names = Vec::with_capacity(entries.len());
for e in &entries {
region.extend_from_slice(&encode_link_message(&e.name, e.object_header_address));
link_names.push(e.name.clone());
}
for m in &oh.messages {
if m.msg_type == MessageType::Attribute {
if m.flags != 0 {
return Err(Error::EditUnsupported(
"a v0/v1 group has a shared attribute message (not convertible in place yet)",
));
}
if m.data.len() > u16::MAX as usize {
return Err(Error::EditUnsupported(
"a v0/v1 group attribute is too large to convert in place",
));
}
#[expect(
clippy::cast_possible_truncation,
reason = "message type ids are a small enum that fits the 1-byte v2 type field"
)]
region.push(MessageType::Attribute.to_u16() as u8);
#[expect(
clippy::cast_possible_truncation,
reason = "attribute body length fits the 2-byte message-size field (oversized \
bodies are rejected above)"
)]
region.extend_from_slice(&(m.data.len() as u16).to_le_bytes());
region.push(0); region.extend_from_slice(&m.data);
}
}
Ok(GroupInfo { region, link_names })
}
fn inspect_group(&self, addr: usize) -> Result<GroupInfo, Error> {
if self.data.len() < addr + 4 || self.data[addr..addr + 4] != *b"OHDR" {
return self.reconstruct_v1_group(addr);
}
let mut region = self.gather_oh_messages(addr)?;
let mut p = 0;
let mut has_link_info = false;
let mut link_names = Vec::new();
while let Some((msg_type, body, body_end)) = next_message(®ion, p)? {
match msg_type {
MessageType::LinkInfo => {
has_link_info = true;
let mut q = body + 2;
if body_end - body >= 2 && region[body + 1] & 0x01 != 0 {
q += 8;
}
if q + 8 <= body_end {
let heap_addr = u64::from_le_bytes(region[q..q + 8].try_into().unwrap());
if heap_addr != u64::MAX {
return Err(Error::EditUnsupported(
"a target group uses dense (fractal-heap) link storage (not supported in place yet)",
));
}
}
}
MessageType::Link => {
if let Ok(link) = LinkMessage::parse(®ion[body..body_end], OFFSET_SIZE) {
link_names.push(link.name);
}
}
MessageType::DataLayout => {
return Err(Error::EditUnsupported(
"a target path names a dataset, not a group",
));
}
_ => {}
}
p = body_end;
}
if !has_link_info {
return Err(Error::EditUnsupported(
"a target group's object header has no link-info message",
));
}
ensure_group_info(&mut region)?;
Ok(GroupInfo { region, link_names })
}
fn read_object(&self, addr: usize) -> Result<ObjModel, Error> {
let region = self.gather_oh_messages(addr)?;
let mut layout: Option<(usize, usize)> = None; let mut has_link_info = false;
let mut children: Vec<(String, u64)> = Vec::new();
let mut non_link: Vec<u8> = Vec::new();
let mut p = 0;
while let Some((msg_type, body, body_end)) = next_message(®ion, p)? {
match msg_type {
MessageType::AttributeInfo => {
return Err(Error::EditUnsupported(
"an object uses dense (fractal-heap) attribute storage (not supported in place yet)",
));
}
MessageType::LinkInfo => {
has_link_info = true;
let mut q = body + 2;
if body_end - body >= 2 && region[body + 1] & 0x01 != 0 {
q += 8;
}
if q + 8 <= body_end {
let heap_addr = u64::from_le_bytes(region[q..q + 8].try_into().unwrap());
if heap_addr != u64::MAX {
return Err(Error::EditUnsupported(
"a group uses dense (fractal-heap) link storage (not supported in place yet)",
));
}
}
}
MessageType::Link => match LinkMessage::parse(®ion[body..body_end], OFFSET_SIZE)
{
Ok(LinkMessage {
name,
link_target:
LinkTarget::Hard {
object_header_address,
},
..
}) => children.push((name, object_header_address)),
_ => {
return Err(Error::EditUnsupported(
"a group contains a soft/external link (not copyable in place yet)",
));
}
},
MessageType::DataLayout => layout = Some((body, body_end - body)),
_ => {}
}
if msg_type != MessageType::Link {
non_link.extend_from_slice(®ion[p..body_end]);
}
p = body_end;
}
if let Some((lbody, lsize)) = layout {
let version = region[lbody];
if !(version == 3 || version == 4) || lsize < 2 {
return Err(Error::EditUnsupported(
"an unsupported data-layout version cannot be copied in place yet",
));
}
let class = region[lbody + 1];
match class {
0 => Ok(ObjModel::DatasetVerbatim { region }),
1 => {
if lbody + 18 > region.len() {
return Err(Error::EditUnsupported("malformed contiguous data layout"));
}
let data_addr =
u64::from_le_bytes(region[lbody + 2..lbody + 10].try_into().unwrap());
let data_size =
u64::from_le_bytes(region[lbody + 10..lbody + 18].try_into().unwrap());
Ok(ObjModel::DatasetContiguous {
region,
addr_off: lbody + 2,
data_addr,
data_size,
})
}
_ => Err(Error::EditUnsupported(
"chunked datasets cannot be copied in place yet",
)),
}
} else if has_link_info {
ensure_group_info(&mut non_link)?;
Ok(ObjModel::Group {
non_link_region: non_link,
children,
})
} else {
Err(Error::EditUnsupported(
"an object is neither a contiguous/compact dataset nor a group",
))
}
}
fn plan_copy(&self, addr: usize, depth: u32) -> Result<(), Error> {
if depth >= MAX_COPY_DEPTH {
return Err(Error::EditUnsupported(
"copy source nests too deeply (possible hard-link cycle)",
));
}
if let ObjModel::Group { children, .. } = self.read_object(addr)? {
for (_, child) in children {
let child = usize::try_from(child)
.map_err(|_| Error::EditUnsupported("child address exceeds this platform"))?;
self.plan_copy(child, depth + 1)?;
}
}
Ok(())
}
fn perform_copy(&mut self, addr: usize, depth: u32) -> Result<u64, Error> {
if depth >= MAX_COPY_DEPTH {
return Err(Error::EditUnsupported(
"copy source nests too deeply (possible hard-link cycle)",
));
}
match self.read_object(addr)? {
ObjModel::DatasetVerbatim { region } => {
let oh = build_v2_object_header(®ion);
self.alloc_or_append(&oh)
}
ObjModel::DatasetContiguous {
mut region,
addr_off,
data_addr,
data_size,
} => {
let start = usize::try_from(data_addr)
.map_err(|_| Error::EditUnsupported("data address exceeds this platform"))?;
let len = usize::try_from(data_size)
.map_err(|_| Error::EditUnsupported("data size exceeds this platform"))?;
let end = start
.checked_add(len)
.filter(|&e| e <= self.data.len())
.ok_or(Error::EditUnsupported("dataset data is out of bounds"))?;
let data = self.data[start..end].to_vec();
let new_data_addr = self.alloc_or_append(&data)?;
region[addr_off..addr_off + 8].copy_from_slice(&new_data_addr.to_le_bytes());
let oh = build_v2_object_header(®ion);
self.alloc_or_append(&oh)
}
ObjModel::Group {
non_link_region,
children,
} => {
let mut region = non_link_region;
for (name, child) in children {
let child = usize::try_from(child).map_err(|_| {
Error::EditUnsupported("child address exceeds this platform")
})?;
let new_child = self.perform_copy(child, depth + 1)?;
region.extend_from_slice(&encode_link_message(&name, new_child));
}
let oh = build_v2_object_header(®ion);
self.alloc_or_append(&oh)
}
}
}
fn append(&mut self, bytes: &[u8]) -> Result<u64, Error> {
let addr = self.data.len() as u64;
self.handle.seek(SeekFrom::Start(addr)).map_err(Error::Io)?;
self.handle.write_all(bytes).map_err(Error::Io)?;
self.data.extend_from_slice(bytes);
Ok(addr)
}
fn write_at(&mut self, offset: usize, bytes: &[u8]) -> Result<(), Error> {
self.handle
.seek(SeekFrom::Start(offset as u64))
.map_err(Error::Io)?;
self.handle.write_all(bytes).map_err(Error::Io)?;
self.data[offset..offset + bytes.len()].copy_from_slice(bytes);
Ok(())
}
fn alloc_or_append(&mut self, bytes: &[u8]) -> Result<u64, Error> {
if let Some(addr) = self.free.alloc(bytes.len() as u64) {
self.write_at(
usize::try_from(addr).map_err(|_| {
Error::EditUnsupported("free-region address exceeds this platform")
})?,
bytes,
)?;
Ok(addr)
} else {
self.append(bytes)
}
}
fn oh_chunk_spans(&self, addr: usize) -> Result<Vec<(u64, u64)>, Error> {
let (rs, re) = self.oh_region(addr)?;
let d = &self.data;
let mut spans: Vec<(u64, u64)> = vec![(addr as u64, (re + 4 - addr) as u64)];
let mut chunks: Vec<(usize, usize)> = vec![(rs, re)];
let mut i = 0;
while i < chunks.len() {
if chunks.len() > MAX_OH_CHUNKS {
return Err(Error::EditUnsupported(
"object header has too many continuation chunks",
));
}
let (cs, ce) = chunks[i];
i += 1;
let region = &d[..ce];
let mut p = cs;
while let Some((msg_type, body, body_end)) = next_message(region, p)? {
if msg_type == MessageType::ObjectHeaderContinuation {
if body_end - body < (OFFSET_SIZE + LENGTH_SIZE) as usize {
return Err(Error::EditUnsupported("malformed continuation message"));
}
let off = u64::from_le_bytes(d[body..body + 8].try_into().unwrap());
let len = u64::from_le_bytes(d[body + 8..body + 16].try_into().unwrap());
let off_us = usize::try_from(off).map_err(|_| {
Error::EditUnsupported("continuation address exceeds this platform")
})?;
let len_us = usize::try_from(len).map_err(|_| {
Error::EditUnsupported("continuation length exceeds this platform")
})?;
let blk_end = off_us
.checked_add(len_us)
.filter(|&e| e <= d.len() && len_us >= 8)
.ok_or(Error::EditUnsupported("continuation block out of bounds"))?;
if d[off_us..off_us + 4] != *b"OCHK" {
return Err(Error::EditUnsupported(
"invalid continuation block signature",
));
}
spans.push((off, len));
chunks.push((off_us + 4, blk_end - 4));
}
p = body_end;
}
}
Ok(spans)
}
fn collect_free_spans(&self, addr: usize, depth: u32, out: &mut Vec<(u64, u64)>) {
if depth >= MAX_COPY_DEPTH {
return;
}
let spans = match self.oh_chunk_spans(addr) {
Ok(s) => s,
Err(_) => return,
};
match self.read_object(addr) {
Ok(ObjModel::DatasetVerbatim { .. }) => out.extend(spans),
Ok(ObjModel::DatasetContiguous {
data_addr,
data_size,
..
}) => {
out.extend(spans);
if data_addr != u64::MAX && data_size > 0 {
if let (Ok(start), Ok(len)) =
(usize::try_from(data_addr), usize::try_from(data_size))
{
if start.checked_add(len).is_some_and(|e| e <= self.data.len()) {
out.push((data_addr, data_size));
}
}
}
}
Ok(ObjModel::Group { children, .. }) => {
out.extend(spans);
for (_, child) in children {
if let Ok(c) = usize::try_from(child) {
self.collect_free_spans(c, depth + 1, out);
}
}
}
Err(_) => {}
}
}
}
#[derive(Default)]
struct Node {
is_new: bool,
datasets: Vec<DatasetBuilder>,
attr_ops: Vec<GroupAttrOp>,
deletes: Vec<String>,
copies: Vec<(String, u64)>,
base_region: Vec<u8>,
existing_links: Vec<String>,
}
enum GroupAttrOp {
Set { name: String, value: AttrValue },
Remove { name: String },
}
enum ObjModel {
DatasetVerbatim { region: Vec<u8> },
DatasetContiguous {
region: Vec<u8>,
addr_off: usize,
data_addr: u64,
data_size: u64,
},
Group {
non_link_region: Vec<u8>,
children: Vec<(String, u64)>,
},
}
struct GroupInfo {
region: Vec<u8>,
link_names: Vec<String>,
}
struct FlatDataset {
name: String,
dt: crate::datatype::Datatype,
ds: Dataspace,
raw: Vec<u8>,
attrs: Vec<crate::attribute::AttributeMessage>,
}
fn split_path(path: &str) -> PathKey {
path.split('/')
.filter(|s| !s.is_empty())
.map(String::from)
.collect()
}
fn ensure_ancestors(nodes: &mut BTreeMap<PathKey, Node>, path: &[String]) {
for len in 0..=path.len() {
nodes.entry(path[..len].to_vec()).or_default();
}
}
fn flatten_dataset(db: DatasetBuilder) -> Result<FlatDataset, Error> {
if db.name.is_empty() {
return Err(Error::EditUnsupported("dataset path has an empty name"));
}
if db.chunk_options.is_chunked() || db.maxshape.is_some() {
return Err(Error::EditUnsupported(
"chunked / compressed / extensible datasets cannot be added in place yet",
));
}
if db.reference_targets.is_some() {
return Err(Error::EditUnsupported(
"object-reference datasets cannot be added in place yet",
));
}
#[cfg(feature = "provenance")]
if db.provenance.is_some() {
return Err(Error::EditUnsupported(
"provenance datasets cannot be added in place yet",
));
}
let dt = db
.datatype
.ok_or(Error::EditUnsupported("dataset has no datatype/data"))?;
let shape = db
.shape
.ok_or(Error::EditUnsupported("dataset has no shape"))?;
if shape.contains(&0) {
return Err(Error::EditUnsupported(
"empty (zero-element) datasets cannot be added in place yet",
));
}
let raw = db
.data
.ok_or(Error::EditUnsupported("dataset has no data"))?;
let elem = dt.type_size() as u64;
if elem > 0 {
let expected = shape.iter().product::<u64>().saturating_mul(elem);
if raw.len() as u64 != expected {
return Err(Error::EditUnsupported(
"dataset data length does not match its shape",
));
}
}
if db
.attrs
.iter()
.any(|(_, v)| matches!(v, AttrValue::VarLenAsciiArray(_)))
{
return Err(Error::EditUnsupported(
"variable-length attributes cannot be added in place yet",
));
}
if db.attrs.len() > MAX_COMPACT_ATTRS {
return Err(Error::EditUnsupported(
"datasets with dense (many) attributes cannot be added in place yet",
));
}
if make_link(&db.name, 0).serialize(OFFSET_SIZE).len() > u16::MAX as usize {
return Err(Error::EditUnsupported(
"dataset name is too long to encode as a link message",
));
}
let ds = Dataspace {
space_type: if shape.is_empty() {
DataspaceType::Scalar
} else {
DataspaceType::Simple
},
#[expect(
clippy::cast_possible_truncation,
reason = "dataspace rank fits the 1-byte dimensionality field (HDF5 caps rank at 32)"
)]
rank: shape.len() as u8,
dimensions: shape,
max_dimensions: None,
};
let attrs = db
.attrs
.iter()
.map(|(n, v)| build_attr_message(n, v))
.collect();
Ok(FlatDataset {
name: db.name,
dt,
ds,
raw,
attrs,
})
}
const GROUP_INFO_BODY: [u8; 2] = [0, 0];
fn region_message(msg_type: MessageType, body: &[u8]) -> Vec<u8> {
let mut m = Vec::with_capacity(4 + body.len());
#[expect(
clippy::cast_possible_truncation,
reason = "message type ids are a small enum that fits the 1-byte v2 type field"
)]
m.push(msg_type.to_u16() as u8);
#[expect(
clippy::cast_possible_truncation,
reason = "callers pass bodies that fit the 2-byte message-size field (see doc comment)"
)]
m.extend_from_slice(&(body.len() as u16).to_le_bytes());
m.push(0); m.extend_from_slice(body);
m
}
fn fresh_group_region() -> Vec<u8> {
let mut li = Vec::with_capacity(18);
li.push(0); li.push(0); li.extend_from_slice(&u64::MAX.to_le_bytes()); li.extend_from_slice(&u64::MAX.to_le_bytes()); let mut region = region_message(MessageType::LinkInfo, &li);
region.extend_from_slice(®ion_message(MessageType::GroupInfo, &GROUP_INFO_BODY));
region
}
fn ensure_group_info(region: &mut Vec<u8>) -> Result<(), Error> {
let mut p = 0;
while let Some((msg_type, _body, body_end)) = next_message(region, p)? {
if msg_type == MessageType::GroupInfo {
return Ok(());
}
p = body_end;
}
region.extend_from_slice(®ion_message(MessageType::GroupInfo, &GROUP_INFO_BODY));
Ok(())
}
fn encode_link_message(name: &str, addr: u64) -> Vec<u8> {
let body = make_link(name, addr).serialize(OFFSET_SIZE);
region_message(MessageType::Link, &body)
}
fn patch_link_target(region: &mut [u8], name: &str, new_addr: u64) -> Result<(), Error> {
let mut p = 0;
while let Some((msg_type, body, body_end)) = next_message(region, p)? {
if msg_type == MessageType::Link {
if let Ok(link) = LinkMessage::parse(®ion[body..body_end], OFFSET_SIZE) {
if link.name == name {
return match link.link_target {
LinkTarget::Hard { .. } => {
let ofs = body_end - OFFSET_SIZE as usize;
region[ofs..body_end].copy_from_slice(&new_addr.to_le_bytes());
Ok(())
}
_ => Err(Error::EditUnsupported(
"a group on the edited path is reached by a soft/external link",
)),
};
}
}
}
p = body_end;
}
Err(Error::EditUnsupported(
"expected child link not found in parent group",
))
}
fn remove_link_from_region(region: &[u8], name: &str) -> Result<Vec<u8>, Error> {
let mut out = Vec::with_capacity(region.len());
let mut p = 0;
let mut removed = false;
while let Some((msg_type, body, body_end)) = next_message(region, p)? {
let mut skip = false;
if msg_type == MessageType::Link {
if let Ok(link) = LinkMessage::parse(®ion[body..body_end], OFFSET_SIZE) {
if link.name == name {
skip = true;
removed = true;
}
}
}
if !skip {
out.extend_from_slice(®ion[p..body_end]);
}
p = body_end;
}
if p < region.len() {
out.extend_from_slice(®ion[p..]);
}
if !removed {
return Err(Error::EditUnsupported(
"link to delete not found in its parent group",
));
}
Ok(out)
}
fn apply_group_attr_ops(region: &[u8], ops: &[GroupAttrOp]) -> Result<Vec<u8>, Error> {
let mut out = region.to_vec();
let mut wrote_attr = false;
for op in ops {
out = match op {
GroupAttrOp::Set { name, value } => {
wrote_attr = true;
set_attr_in_region(&out, name, value)?
}
GroupAttrOp::Remove { name } => remove_attr_from_region(&out, name)?,
};
}
if wrote_attr && compact_attr_count(&out)? > MAX_COMPACT_ATTRS {
return Err(Error::EditUnsupported(
"group attributes would exceed compact storage; dense attribute edits are not supported in place yet",
));
}
Ok(out)
}
fn set_attr_in_region(region: &[u8], name: &str, value: &AttrValue) -> Result<Vec<u8>, Error> {
let new_msg = encode_attr_message(name, value)?;
let mut out = Vec::with_capacity(region.len() + new_msg.len());
let mut p = 0;
while let Some((msg_type, body, body_end)) = next_message(region, p)? {
match msg_type {
MessageType::AttributeInfo => {
return Err(Error::EditUnsupported(
"a target group uses dense (fractal-heap) attribute storage (not supported in place yet)",
));
}
MessageType::Attribute => {
let attr_name = parse_compact_attr_name(region, p, body, body_end)?;
if attr_name == name {
p = body_end;
continue;
}
}
_ => {}
}
out.extend_from_slice(®ion[p..body_end]);
p = body_end;
}
out.extend_from_slice(&new_msg);
if p < region.len() {
out.extend_from_slice(®ion[p..]);
}
Ok(out)
}
fn remove_attr_from_region(region: &[u8], name: &str) -> Result<Vec<u8>, Error> {
let mut out = Vec::with_capacity(region.len());
let mut p = 0;
let mut removed = false;
while let Some((msg_type, body, body_end)) = next_message(region, p)? {
let mut skip = false;
match msg_type {
MessageType::AttributeInfo => {
return Err(Error::EditUnsupported(
"a target group uses dense (fractal-heap) attribute storage (not supported in place yet)",
));
}
MessageType::Attribute => {
let attr_name = parse_compact_attr_name(region, p, body, body_end)?;
if attr_name == name {
skip = true;
removed = true;
}
}
_ => {}
}
if !skip {
out.extend_from_slice(®ion[p..body_end]);
}
p = body_end;
}
if p < region.len() {
out.extend_from_slice(®ion[p..]);
}
if !removed {
return Err(Error::EditUnsupported(
"group attribute to remove was not found",
));
}
Ok(out)
}
fn compact_attr_count(region: &[u8]) -> Result<usize, Error> {
let mut count = 0usize;
let mut p = 0;
while let Some((msg_type, _body, body_end)) = next_message(region, p)? {
if msg_type == MessageType::AttributeInfo {
return Err(Error::EditUnsupported(
"a target group uses dense (fractal-heap) attribute storage (not supported in place yet)",
));
}
if msg_type == MessageType::Attribute {
count += 1;
}
p = body_end;
}
Ok(count)
}
fn parse_compact_attr_name(
region: &[u8],
msg_start: usize,
body: usize,
body_end: usize,
) -> Result<String, Error> {
if region[msg_start + 3] != 0 {
return Err(Error::EditUnsupported(
"a target group has a shared attribute message (not editable in place yet)",
));
}
crate::attribute::AttributeMessage::parse(®ion[body..body_end], LENGTH_SIZE)
.map(|attr| attr.name)
.map_err(|_| Error::EditUnsupported("a target group has an unreadable attribute message"))
}
fn encode_attr_message(name: &str, value: &AttrValue) -> Result<Vec<u8>, Error> {
if matches!(value, AttrValue::VarLenAsciiArray(_)) {
return Err(Error::EditUnsupported(
"variable-length group attributes cannot be edited in place yet",
));
}
let body = build_attr_message(name, value).serialize(LENGTH_SIZE);
if body.len() > u16::MAX as usize {
return Err(Error::EditUnsupported(
"group attribute is too large to encode in place",
));
}
Ok(region_message(MessageType::Attribute, &body))
}
fn is_prefix(a: &[String], b: &[String]) -> bool {
a.len() <= b.len() && b[..a.len()] == *a
}
fn next_message(region: &[u8], p: usize) -> Result<Option<(MessageType, usize, usize)>, Error> {
if p + 4 > region.len() {
return Ok(None);
}
let msg_type = MessageType::from_u16(region[p] as u16);
let msg_size = u16::from_le_bytes([region[p + 1], region[p + 2]]) as usize;
let body = p + 4;
let body_end = body + msg_size;
if body_end > region.len() {
return Err(Error::EditUnsupported("malformed object header message"));
}
Ok(Some((msg_type, body, body_end)))
}
fn build_v2_object_header(region: &[u8]) -> Vec<u8> {
let total = region.len();
let (flags, width) = if total <= 255 {
(0u8, 1usize)
} else if total <= 65535 {
(1u8, 2)
} else {
(2u8, 4)
};
let mut buf = Vec::with_capacity(8 + total + 4);
buf.extend_from_slice(b"OHDR");
buf.push(2); buf.push(flags);
#[expect(
clippy::cast_possible_truncation,
reason = "width was selected just above to be the smallest field that holds total"
)]
match width {
1 => buf.push(total as u8),
2 => buf.extend_from_slice(&(total as u16).to_le_bytes()),
_ => buf.extend_from_slice(&(total as u32).to_le_bytes()),
}
buf.extend_from_slice(region);
let checksum = jenkins_lookup3(&buf);
buf.extend_from_slice(&checksum.to_le_bytes());
buf
}
#[expect(
clippy::cast_possible_truncation,
reason = "callers parse in-file sizes/offsets bounded by the in-memory image; downstream \
slicing is length-checked, so a malformed oversized field errors rather than reads OOB"
)]
fn read_le(bytes: &[u8]) -> usize {
let mut v = 0u64;
for (i, &b) in bytes.iter().enumerate() {
v |= (b as u64) << (8 * i);
}
v as usize
}
#[cfg(test)]
mod tests {
use super::*;
fn region_types(region: &[u8]) -> Vec<MessageType> {
let mut out = Vec::new();
let mut p = 0;
while let Some((mt, _, end)) = next_message(region, p).unwrap() {
out.push(mt);
p = end;
}
out
}
#[test]
fn fresh_group_region_pairs_link_info_with_group_info() {
let types = region_types(&fresh_group_region());
assert_eq!(types, vec![MessageType::LinkInfo, MessageType::GroupInfo]);
}
#[test]
fn ensure_group_info_appends_when_missing() {
let li_body = {
let mut b = vec![0u8, 0];
b.extend_from_slice(&u64::MAX.to_le_bytes());
b.extend_from_slice(&u64::MAX.to_le_bytes());
b
};
let mut region = region_message(MessageType::LinkInfo, &li_body);
ensure_group_info(&mut region).unwrap();
assert_eq!(
region_types(®ion),
vec![MessageType::LinkInfo, MessageType::GroupInfo]
);
let mut p = 0;
while let Some((mt, body, end)) = next_message(®ion, p).unwrap() {
if mt == MessageType::GroupInfo {
assert_eq!(®ion[body..end], &GROUP_INFO_BODY);
}
p = end;
}
}
#[test]
fn ensure_group_info_is_idempotent() {
let mut region = fresh_group_region();
let before = region.clone();
ensure_group_info(&mut region).unwrap();
assert_eq!(region, before);
}
#[test]
fn commit_clears_a_stale_consistency_flag() {
use crate::writer::FileBuilder;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("stale_flag.h5");
let mut b = FileBuilder::new();
b.create_dataset("d").with_i32_data(&[1, 2, 3]);
b.write(&path).unwrap();
{
let mut data = std::fs::read(&path).unwrap();
let off = signature::find_signature(&data).unwrap();
let mut sb = Superblock::parse(&data, off).unwrap();
assert!(
sb.version >= 2,
"FileBuilder should emit a v2/v3 superblock"
);
sb.consistency_flags = 0x05;
let bytes = sb.serialize();
data[off..off + bytes.len()].copy_from_slice(&bytes);
std::fs::write(&path, &data).unwrap();
assert_eq!(
Superblock::parse(&data, off).unwrap().consistency_flags,
0x05
);
}
{
let mut s = EditSession::open(&path).unwrap();
s.create_dataset("e").with_i32_data(&[4, 5]);
s.commit().unwrap();
}
let data = std::fs::read(&path).unwrap();
let off = signature::find_signature(&data).unwrap();
assert_eq!(
Superblock::parse(&data, off).unwrap().consistency_flags,
0,
"commit must clear the stale consistency flag"
);
}
}