From bf8bbec87e829501e5585950d18b7f38a61e8d0c Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 06:14:30 -0700 Subject: [PATCH] fix(clawhdf5): surface filter-pipeline parse errors; write files atomically - Dataset::filter_pipeline() (reader, lazy and mmap variants) swallowed parse errors with `.ok()`, so a malformed pipeline message silently became "no filters" and the still-compressed chunk bytes were returned as the data. It now returns Result>; a present-but-unparseable pipeline is Error::Format. - FileBuilder::write used std::fs::write, which truncates the destination first: a crash mid-write destroyed the existing file. It now writes a sibling temp file, syncs it, renames it over the target and syncs the directory, cleaning the temp file up on failure. Co-Authored-By: Claude Fable 5.1 --- crates/clawhdf5/src/lazy.rs | 11 +++-- crates/clawhdf5/src/mmap_file.rs | 11 +++-- crates/clawhdf5/src/reader.rs | 13 +++-- crates/clawhdf5/src/writer.rs | 85 +++++++++++++++++++++++++++++++- 4 files changed, 109 insertions(+), 11 deletions(-) diff --git a/crates/clawhdf5/src/lazy.rs b/crates/clawhdf5/src/lazy.rs index e561deb..192a05c 100644 --- a/crates/clawhdf5/src/lazy.rs +++ b/crates/clawhdf5/src/lazy.rs @@ -436,19 +436,24 @@ impl<'f, R: HDF5Read> LazyDataset<'f, R> { )?) } - fn filter_pipeline(&self) -> Option { + /// `Ok(None)` means the dataset has no filter pipeline. A pipeline message + /// that is present but unparseable is an error: treating it as "no + /// filters" would hand the caller the still-compressed bytes as if they + /// were the data. + fn filter_pipeline(&self) -> Result, Error> { self.header .messages .iter() .find(|m| m.msg_type == MessageType::FilterPipeline) - .and_then(|msg| FilterPipeline::parse(&msg.data).ok()) + .map(|msg| FilterPipeline::parse(&msg.data).map_err(Error::Format)) + .transpose() } fn read_raw(&self) -> Result, Error> { let dt = self.datatype()?; let ds = self.dataspace()?; let dl = self.data_layout()?; - let pipeline = self.filter_pipeline(); + let pipeline = self.filter_pipeline()?; let data = self.file.reader.as_bytes(); Ok(data_read::read_raw_data_full( data, diff --git a/crates/clawhdf5/src/mmap_file.rs b/crates/clawhdf5/src/mmap_file.rs index 972501d..1573780 100644 --- a/crates/clawhdf5/src/mmap_file.rs +++ b/crates/clawhdf5/src/mmap_file.rs @@ -377,19 +377,24 @@ impl<'f> MmapDataset<'f> { )?) } - fn filter_pipeline(&self) -> Option { + /// `Ok(None)` means the dataset has no filter pipeline. A pipeline message + /// that is present but unparseable is an error: treating it as "no + /// filters" would hand the caller the still-compressed bytes as if they + /// were the data. + fn filter_pipeline(&self) -> Result, Error> { self.header .messages .iter() .find(|m| m.msg_type == MessageType::FilterPipeline) - .and_then(|msg| FilterPipeline::parse(&msg.data).ok()) + .map(|msg| FilterPipeline::parse(&msg.data).map_err(Error::Format)) + .transpose() } fn read_raw(&self) -> Result, Error> { let dt = self.datatype()?; let ds = self.dataspace()?; let dl = self.data_layout()?; - let pipeline = self.filter_pipeline(); + let pipeline = self.filter_pipeline()?; Ok(data_read::read_raw_data_full( self.file.reader.as_bytes(), &dl, diff --git a/crates/clawhdf5/src/reader.rs b/crates/clawhdf5/src/reader.rs index 223664d..fee9594 100644 --- a/crates/clawhdf5/src/reader.rs +++ b/crates/clawhdf5/src/reader.rs @@ -447,7 +447,7 @@ impl<'f> Dataset<'f> { let dt = self.datatype()?; let ds = self.dataspace()?; let dl = self.data_layout()?; - let pipeline = self.filter_pipeline(); + let pipeline = self.filter_pipeline()?; Ok(data_read::read_raw_data_selection( self.file.data.as_bytes(), &dl, @@ -743,19 +743,24 @@ impl<'f> Dataset<'f> { )?) } - fn filter_pipeline(&self) -> Option { + /// `Ok(None)` means the dataset has no filter pipeline. A pipeline message + /// that is present but unparseable is an error: treating it as "no + /// filters" would hand the caller the still-compressed bytes as if they + /// were the data. + fn filter_pipeline(&self) -> Result, Error> { self.header .messages .iter() .find(|m| m.msg_type == MessageType::FilterPipeline) - .and_then(|msg| FilterPipeline::parse(&msg.data).ok()) + .map(|msg| FilterPipeline::parse(&msg.data).map_err(Error::Format)) + .transpose() } fn read_raw(&self) -> Result, Error> { let dt = self.datatype()?; let ds = self.dataspace()?; let dl = self.data_layout()?; - let pipeline = self.filter_pipeline(); + let pipeline = self.filter_pipeline()?; // Virtual datasets are assembled from source datasets; the per-file // chunk cache does not apply. Route them through the resolver path so diff --git a/crates/clawhdf5/src/writer.rs b/crates/clawhdf5/src/writer.rs index a9aa64d..59904ba 100644 --- a/crates/clawhdf5/src/writer.rs +++ b/crates/clawhdf5/src/writer.rs @@ -72,7 +72,7 @@ impl FileBuilder { /// Serialize and write the file to the given path. pub fn write>(self, path: P) -> Result<(), Error> { let bytes = self.finish()?; - std::fs::write(path, bytes).map_err(Error::Io) + write_file_atomically(path.as_ref(), &bytes).map_err(Error::Io) } } @@ -186,3 +186,86 @@ pub fn create_datasets_parallel(specs: Vec) -> Result, Erro let bytes = clawhdf5_format::file_writer::finalize_parallel(blocks)?; Ok(bytes) } + +/// Write `bytes` to `path` so that a crash or power loss leaves either the old +/// file or the complete new one — never a truncated mix. `std::fs::write` +/// truncates the destination first, so dying mid-write used to destroy the +/// existing file. +fn write_file_atomically(path: &std::path::Path, bytes: &[u8]) -> std::io::Result<()> { + use std::io::Write; + + // Same directory as the target, so the rename stays on one filesystem. + let mut tmp_name = path + .file_name() + .ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "path has no file name") + })? + .to_os_string(); + tmp_name.push(format!(".tmp-{}", std::process::id())); + let tmp_path = path.with_file_name(tmp_name); + + let result = (|| { + let mut f = std::fs::File::create(&tmp_path)?; + f.write_all(bytes)?; + f.sync_all()?; + std::fs::rename(&tmp_path, path) + })(); + if result.is_err() { + let _ = std::fs::remove_file(&tmp_path); + return result; + } + // Make the rename itself durable. Best-effort: not every filesystem + // supports syncing a directory, and the new file is already in place. + #[cfg(unix)] + if let Some(dir) = path.parent() { + let dir = if dir.as_os_str().is_empty() { + std::path::Path::new(".") + } else { + dir + }; + if let Ok(d) = std::fs::File::open(dir) { + let _ = d.sync_all(); + } + } + Ok(()) +} + +#[cfg(test)] +mod atomic_write_tests { + use super::write_file_atomically; + + fn entries(dir: &std::path::Path) -> Vec { + let mut names: Vec = std::fs::read_dir(dir) + .unwrap() + .map(|e| e.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + names.sort(); + names + } + + #[test] + fn replaces_existing_file_and_leaves_no_temp_behind() { + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().join("out.h5"); + std::fs::write(&path, b"old contents").unwrap(); + + write_file_atomically(&path, b"new").unwrap(); + + assert_eq!(std::fs::read(&path).unwrap(), b"new"); + assert_eq!(entries(dir.path()), ["out.h5"]); + } + + #[test] + fn failure_leaves_the_existing_file_untouched() { + let dir = tempfile::TempDir::new().unwrap(); + // The target is a directory, so the final rename cannot succeed. + let path = dir.path().join("taken"); + std::fs::create_dir(&path).unwrap(); + std::fs::write(path.join("keep"), b"x").unwrap(); + + assert!(write_file_atomically(&path, b"new").is_err()); + + assert!(path.is_dir()); + assert_eq!(entries(dir.path()), ["taken"], "temp file cleaned up"); + } +}