haematite 0.7.0

//! Node-file publication and the zstd codec shared with the vacuum's mark walk.

use std::io::{ErrorKind, Write};
use std::path::{Path, PathBuf};

use super::error::StoreError;

/// Atomically publish a compressed node file, syncing its DATA but NOT its
/// parent directory entry.
///
/// The temp file's data is fsync'd (`temp_file.sync_all`) before the atomic
/// `persist_noclobber` rename, so the node bytes are durable. The directory
/// entry created by the rename is deliberately NOT fsync'd here: doing so per
/// node would fsync the same subdirectory once per node. Instead the caller
/// records the returned parent directory and the commit path fsyncs each
/// DISTINCT directory ONCE, as a batched barrier, strictly before the WAL marker
/// (see `super::DiskStore::sync_dirty_dirs`).
///
/// Returns `Some(parent_dir)` when a NEW file was persisted (its directory entry
/// still needs the barrier), or `None` when the file already existed (the
/// directory entry is already durable from the prior write).
///
/// D1 exception: the prefix subdirectory is established by
/// [`super::barrier::ensure_prefix_dir`] BEFORE this call (so the store-root
/// retention obligation is recorded first); this publish therefore creates no
/// directory. A vanished prefix subdir surfaces as the natural I/O error from
/// `tempfile_in`.
pub fn publish_compressed_node(
    final_path: &Path,
    compressed: &[u8],
) -> Result<Option<PathBuf>, StoreError> {
    let Some(parent_dir) = final_path.parent() else {
        return Err(StoreError::MissingParentDirectory {
            path: final_path.to_path_buf(),
        });
    };

    let mut temp_file = tempfile::Builder::new()
        .prefix(".node-")
        .suffix(".tmp")
        .tempfile_in(parent_dir)
        .map_err(StoreError::Io)?;
    temp_file.write_all(compressed).map_err(StoreError::Io)?;
    temp_file.as_file_mut().sync_all().map_err(StoreError::Io)?;

    match temp_file.persist_noclobber(final_path) {
        Ok(_file) => Ok(Some(parent_dir.to_path_buf())),
        Err(error) if error.error.kind() == ErrorKind::AlreadyExists => Ok(None),
        Err(error) => Err(StoreError::Io(error.error)),
    }
}

pub fn path_exists(path: &Path) -> Result<bool, StoreError> {
    path.try_exists().map_err(StoreError::Io)
}

pub fn compress_node(serialised: &[u8]) -> Result<Vec<u8>, StoreError> {
    zstd::stream::encode_all(serialised, 0)
        .map_err(|error| StoreError::Compression(error.to_string()))
}

/// Decompress a stored node file's bytes. Shared with the vacuum's verified
/// mark walk (STORAGE-VACUUM.md §4 ⟨r2, B2⟩), which recomputes every traversed
/// node's content hash from these exact decoded bytes — the same decode the
/// production read path uses, so the two cannot drift.
pub fn decompress_node(compressed: &[u8]) -> Result<Vec<u8>, StoreError> {
    zstd::stream::decode_all(compressed).map_err(|error| StoreError::Compression(error.to_string()))
}