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<Option<_>>; 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 <[email protected]>

crates/clawhdf5/src/lazy.rs

@@ -436,19 +436,24 @@ impl<'f, R: HDF5Read> LazyDataset<'f, R> {
         )?)
     }
 
-    fn filter_pipeline(&self) -> Option<FilterPipeline> {
+    /// `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<Option<FilterPipeline>, 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<Vec<u8>, 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,

crates/clawhdf5/src/mmap_file.rs

@@ -377,19 +377,24 @@ impl<'f> MmapDataset<'f> {
         )?)
     }
 
-    fn filter_pipeline(&self) -> Option<FilterPipeline> {
+    /// `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<Option<FilterPipeline>, 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<Vec<u8>, 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,

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<FilterPipeline> {
+    /// `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<Option<FilterPipeline>, 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<Vec<u8>, 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

crates/clawhdf5/src/writer.rs

@@ -72,7 +72,7 @@ impl FileBuilder {
     /// Serialize and write the file to the given path.
     pub fn write<P: AsRef<std::path::Path>>(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<DatasetSpec>) -> Result<Vec<u8>, 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<String> {
+        let mut names: Vec<String> = 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");
+    }
+}