polyc-eventlog 2026.10.0

//! Durable "tail high-water" artifact (U-4a): the exact highest committed
//! state a partition's covering signed root and physical journal frontier
//! have ever proved durable.
//!
//! [`crate::checkpoint::EventCountCheckpoint`] is a MINIMUM floor: a crash
//! between a commit and the checkpoint's own sync leaves it stale (lower
//! than reality), and that staleness can only ever weaken detection, never
//! manufacture a false failure. This artifact is the opposite shape on
//! purpose — an EXACT high-water bound to one physical partition-journal
//! lineage (an incarnation), one committed logical head, one physical frame
//! frontier, and the covering signed root over them. It exists to catch a
//! failure the floor cannot: deleting or zeroing the NEWEST retained section
//! of a partition. A floor comparison (`actual < expected`) is a MINIMUM
//! check; it never rises past what a legitimate deliberate narrowing already
//! lowered it to, so it has no way to insist that a partition's CURRENT tail
//! still matches its LAST tail. Binding the exact head, frontier, and root
//! together is what closes that gap.
//!
//! # Why not `commonware_storage::metadata::Metadata`
//!
//! `Metadata`'s dual-blob rotation makes a single corrupted copy fall back to
//! the other transparently, and BOTH copies invalid or absent read back as an
//! *empty* store — indistinguishable from a store that was never written.
//! That is the right behavior for a MINIMUM floor: an empty floor is a safe,
//! permissive `0`. It is the wrong behavior for a high-water mark: a lost
//! artifact must never look like a fresh partition that never committed
//! anything. So this is a single plain file, atomically replaced (temp file,
//! `fsync`, rename, directory `fsync`), and reading it distinguishes
//! "genuinely never written" from "present but unreadable" as two different
//! constructible outcomes rather than normalizing the second into the first.
//!
//! # Format
//!
//! Fixed-width, no variable-length fields, so a torn write is detectable by
//! length alone before the checksum is even consulted:
//!
//! ```text
//! magic (4) | revision (1) | partition digest (32) | incarnation (32)
//!   | logical head (8, BE) | physical frontier (8, BE)
//!   | covering root (32) | covering leaf count (8, BE) | checksum (32)
//! ```
//!
//! The partition identity is bound as a SHA-256 digest of its logical name
//! rather than the raw (variable-length) name, so the record stays
//! fixed-width and [`TailHighWater::read`] can still refuse the artifact of
//! one partition being read for another: it takes the partition name the
//! caller expects and refuses a mismatch rather than trusting the directory
//! it was handed.

use std::{
    fs::File,
    io::Write as _,
    path::{Path, PathBuf},
};

use sha2::{Digest as _, Sha256};

/// The four bytes every artifact of this family starts with.
pub const MAGIC: [u8; 4] = *b"THW1";

/// The revision this build writes and the only one it reads.
pub const FORMAT_REVISION: u8 = 1;

const PARTITION_DIGEST_LEN: usize = 32;
const INCARNATION_LEN: usize = 32;
const HEAD_LEN: usize = 8;
const FRONTIER_LEN: usize = 8;
const ROOT_LEN: usize = 32;
const LEAF_COUNT_LEN: usize = 8;
const CHECKSUM_LEN: usize = 32;

/// Everything the checksum covers: magic, revision, partition digest,
/// incarnation, logical head, physical frontier, covering root, and covering
/// leaf count.
const BOUND_LEN: usize = MAGIC.len()
    + 1
    + PARTITION_DIGEST_LEN
    + INCARNATION_LEN
    + HEAD_LEN
    + FRONTIER_LEN
    + ROOT_LEN
    + LEAF_COUNT_LEN;
const RECORD_LEN: usize = BOUND_LEN + CHECKSUM_LEN;

/// The artifact's file name inside the caller-supplied directory.
pub const FILE_NAME: &str = "tailhighwater";
/// Where the artifact is staged before the atomic rename.
///
/// A fixed name, not a unique one: a leftover from a crashed publish is
/// overwritten by the next attempt, and it never collides with
/// [`FILE_NAME`].
const TEMP_FILE_NAME: &str = "tailhighwater.writing";

/// One partition's durable high-water: the exact committed state its newest
/// covering signed root and physical frontier have proved.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TailHighWater {
    partition_digest: [u8; PARTITION_DIGEST_LEN],
    incarnation: [u8; INCARNATION_LEN],
    logical_head: u64,
    physical_frontier: u64,
    covering_root: [u8; ROOT_LEN],
    covering_leaf_count: u64,
}

/// What reading a partition's high-water artifact found.
///
/// [`Self::Fresh`] and a corrupt, foreign, stale, or wrong-incarnation
/// artifact are deliberately different constructible states, never folded
/// into one `None`: a caller that could not tell them apart could not tell a
/// partition that has never committed anything from one whose proof of what
/// it committed is gone.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TailHighWaterOutcome {
    /// No artifact has ever been written in this directory.
    Fresh,
    /// The durable high-water this partition last recorded.
    Present(TailHighWater),
}

/// Why a [`TailHighWater`] failed to write, or a read refused what it found.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum HighWaterError {
    /// The write or read itself failed at the filesystem.
    #[error("tail high-water i/o failure: {0}")]
    Io(#[from] std::io::Error),
    /// The file is not this format's exact length — a torn write, a foreign
    /// file, or a hand-edited one. Never read as if it were absent.
    #[error("tail high-water is {found} byte(s), not {expected}")]
    WrongLength {
        /// The exact length a valid record has.
        expected: usize,
        /// The length the file actually was.
        found: usize,
    },
    /// The file does not carry this family's magic.
    #[error("tail high-water carries another family's magic")]
    WrongMagic,
    /// The file declares a format revision this build does not read.
    #[error("tail high-water format revision {found} is not {expected}")]
    UnsupportedRevision {
        /// The revision this build writes and reads.
        expected: u8,
        /// The revision the file declared.
        found: u8,
    },
    /// The checksum does not cover the bytes it should — a torn write or a
    /// flipped bit.
    #[error("tail high-water fails its checksum")]
    ChecksumMismatch,
    /// The artifact names a different partition than the caller asked for.
    #[error("tail high-water was recorded for a different partition")]
    WrongPartition,
}

impl TailHighWater {
    /// Binds `partition`'s current durable high-water: the incarnation and
    /// logical head a commit just made durable, the raw physical journal
    /// length after that commit, and the root and leaf count of the
    /// covering signed root over it.
    #[must_use]
    pub fn new(
        partition: &str,
        incarnation: [u8; INCARNATION_LEN],
        logical_head: u64,
        physical_frontier: u64,
        covering_root: [u8; ROOT_LEN],
        covering_leaf_count: u64,
    ) -> Self {
        Self {
            partition_digest: digest_partition(partition),
            incarnation,
            logical_head,
            physical_frontier,
            covering_root,
            covering_leaf_count,
        }
    }

    /// Returns the physical lineage this high-water was bound to.
    #[must_use]
    pub const fn incarnation(&self) -> [u8; INCARNATION_LEN] {
        self.incarnation
    }

    /// Returns the committed logical head this high-water was bound to.
    #[must_use]
    pub const fn logical_head(&self) -> u64 {
        self.logical_head
    }

    /// Returns the raw physical journal length this high-water was bound to.
    #[must_use]
    pub const fn physical_frontier(&self) -> u64 {
        self.physical_frontier
    }

    /// Returns the covering signed root this high-water was bound to.
    #[must_use]
    pub const fn covering_root(&self) -> [u8; ROOT_LEN] {
        self.covering_root
    }

    /// Returns the covering signed root's leaf count.
    #[must_use]
    pub const fn covering_leaf_count(&self) -> u64 {
        self.covering_leaf_count
    }

    fn to_bytes(self) -> [u8; RECORD_LEN] {
        let mut bound = [0_u8; BOUND_LEN];
        let mut at = 0;
        bound[at..at + MAGIC.len()].copy_from_slice(&MAGIC);
        at += MAGIC.len();
        bound[at] = FORMAT_REVISION;
        at += 1;
        bound[at..at + PARTITION_DIGEST_LEN].copy_from_slice(&self.partition_digest);
        at += PARTITION_DIGEST_LEN;
        bound[at..at + INCARNATION_LEN].copy_from_slice(&self.incarnation);
        at += INCARNATION_LEN;
        bound[at..at + HEAD_LEN].copy_from_slice(&self.logical_head.to_be_bytes());
        at += HEAD_LEN;
        bound[at..at + FRONTIER_LEN].copy_from_slice(&self.physical_frontier.to_be_bytes());
        at += FRONTIER_LEN;
        bound[at..at + ROOT_LEN].copy_from_slice(&self.covering_root);
        at += ROOT_LEN;
        bound[at..at + LEAF_COUNT_LEN].copy_from_slice(&self.covering_leaf_count.to_be_bytes());
        at += LEAF_COUNT_LEN;
        debug_assert_eq!(at, BOUND_LEN);

        let checksum = Sha256::digest(bound);
        let mut record = [0_u8; RECORD_LEN];
        record[..BOUND_LEN].copy_from_slice(&bound);
        record[BOUND_LEN..].copy_from_slice(&checksum);
        record
    }

    fn from_bytes(bytes: &[u8], partition: &str) -> Result<Self, HighWaterError> {
        if bytes.len() != RECORD_LEN {
            return Err(HighWaterError::WrongLength {
                expected: RECORD_LEN,
                found: bytes.len(),
            });
        }
        let (bound, checksum) = bytes.split_at(BOUND_LEN);
        // The checksum is checked before a single other field is trusted: a
        // torn write can flip the magic or the revision just as easily as
        // any other byte, and only the checksum tells a complete record from
        // a damaged one.
        let computed = Sha256::digest(bound);
        if computed.as_slice() != checksum {
            return Err(HighWaterError::ChecksumMismatch);
        }
        if bound[..MAGIC.len()] != MAGIC {
            return Err(HighWaterError::WrongMagic);
        }
        let mut at = MAGIC.len();
        let revision = bound[at];
        at += 1;
        if revision != FORMAT_REVISION {
            return Err(HighWaterError::UnsupportedRevision {
                expected: FORMAT_REVISION,
                found: revision,
            });
        }
        let partition_digest: [u8; PARTITION_DIGEST_LEN] = bound[at..at + PARTITION_DIGEST_LEN]
            .try_into()
            .expect("slice is exactly PARTITION_DIGEST_LEN bytes");
        at += PARTITION_DIGEST_LEN;
        if partition_digest != digest_partition(partition) {
            return Err(HighWaterError::WrongPartition);
        }
        let incarnation: [u8; INCARNATION_LEN] = bound[at..at + INCARNATION_LEN]
            .try_into()
            .expect("slice is exactly INCARNATION_LEN bytes");
        at += INCARNATION_LEN;
        let logical_head = u64::from_be_bytes(
            bound[at..at + HEAD_LEN]
                .try_into()
                .expect("slice is exactly HEAD_LEN bytes"),
        );
        at += HEAD_LEN;
        let physical_frontier = u64::from_be_bytes(
            bound[at..at + FRONTIER_LEN]
                .try_into()
                .expect("slice is exactly FRONTIER_LEN bytes"),
        );
        at += FRONTIER_LEN;
        let covering_root: [u8; ROOT_LEN] = bound[at..at + ROOT_LEN]
            .try_into()
            .expect("slice is exactly ROOT_LEN bytes");
        at += ROOT_LEN;
        let covering_leaf_count = u64::from_be_bytes(
            bound[at..at + LEAF_COUNT_LEN]
                .try_into()
                .expect("slice is exactly LEAF_COUNT_LEN bytes"),
        );
        at += LEAF_COUNT_LEN;
        debug_assert_eq!(at, BOUND_LEN);

        Ok(Self {
            partition_digest,
            incarnation,
            logical_head,
            physical_frontier,
            covering_root,
            covering_leaf_count,
        })
    }

    /// Writes this high-water to `dir` atomically: temp file, `fsync`,
    /// rename, then `fsync` the directory.
    ///
    /// Replaces whatever `dir` held before. The caller is responsible for
    /// calling this only after the state it binds is itself durable (see the
    /// module doc), and for serializing writes against the same directory —
    /// this performs no locking of its own.
    ///
    /// # Errors
    ///
    /// Returns [`HighWaterError::Io`] if the directory cannot be created, the
    /// temp file cannot be written, synced, or renamed into place, or the
    /// directory cannot be synced afterward.
    pub fn write_atomic(self, dir: &Path) -> Result<(), HighWaterError> {
        std::fs::create_dir_all(dir)?;
        let bytes = self.to_bytes();
        let temp = dir.join(TEMP_FILE_NAME);
        {
            let mut file = File::create(&temp)?;
            file.write_all(&bytes)?;
            file.sync_all()?;
        }
        std::fs::rename(&temp, dir.join(FILE_NAME))?;
        File::open(dir)?.sync_all()?;
        Ok(())
    }

    /// Reads `partition`'s high-water artifact from `dir`.
    ///
    /// A genuinely absent file reads as [`TailHighWaterOutcome::Fresh`] —
    /// distinct from every other outcome, which is an error. Whether
    /// `Fresh` is an acceptable answer for an already-established partition
    /// is a judgment this function cannot make on its own: it knows only
    /// this one file, not whether the partition's journal already holds a
    /// signed root. The caller decides that against what the journal
    /// separately proves.
    ///
    /// # Errors
    ///
    /// Returns [`HighWaterError::WrongLength`] for anything other than a
    /// genuinely missing file or an exact-length record — a torn write is
    /// refused, never read as a shorter (or empty) valid record.
    /// Returns [`HighWaterError::ChecksumMismatch`] for a flipped bit.
    /// Returns [`HighWaterError::WrongMagic`] for another family's file.
    /// Returns [`HighWaterError::UnsupportedRevision`] for a revision this
    /// build does not read. Returns [`HighWaterError::WrongPartition`] when
    /// the artifact names a different partition than `partition`.
    pub fn read(dir: &Path, partition: &str) -> Result<TailHighWaterOutcome, HighWaterError> {
        let path: PathBuf = dir.join(FILE_NAME);
        let bytes = match std::fs::read(&path) {
            Ok(bytes) => bytes,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                return Ok(TailHighWaterOutcome::Fresh);
            }
            Err(error) => return Err(error.into()),
        };
        Self::from_bytes(&bytes, partition).map(TailHighWaterOutcome::Present)
    }
}

fn digest_partition(partition: &str) -> [u8; PARTITION_DIGEST_LEN] {
    Sha256::digest(partition.as_bytes()).into()
}

#[cfg(test)]
mod tests {
    use super::*;

    fn sample(partition: &str) -> TailHighWater {
        TailHighWater::new(partition, [7_u8; 32], 42, 44, [9_u8; 32], 43)
    }

    #[test]
    fn a_fresh_directory_reads_as_fresh_not_an_error() {
        let dir = tempfile::tempdir().expect("temp dir");
        let outcome = TailHighWater::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, TailHighWaterOutcome::Fresh);
    }

    #[test]
    fn a_written_high_water_survives_a_round_trip() {
        let dir = tempfile::tempdir().expect("temp dir");
        let written = sample("conv-1");
        written.write_atomic(dir.path()).expect("write");

        let outcome = TailHighWater::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, TailHighWaterOutcome::Present(written));
    }

    #[test]
    fn a_second_write_replaces_the_first_rather_than_appending() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1").write_atomic(dir.path()).expect("write");
        let second = TailHighWater::new("conv-1", [7_u8; 32], 99, 101, [9_u8; 32], 100);
        second.write_atomic(dir.path()).expect("write");

        let outcome = TailHighWater::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, TailHighWaterOutcome::Present(second));
        // No temp file left behind.
        assert!(!dir.path().join(TEMP_FILE_NAME).exists());
    }

    #[test]
    fn a_truncated_record_is_refused_not_read_as_shorter_or_empty() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1").write_atomic(dir.path()).expect("write");
        let mut bytes = std::fs::read(dir.path().join(FILE_NAME)).expect("read raw");
        bytes.truncate(bytes.len() - 1);
        std::fs::write(dir.path().join(FILE_NAME), &bytes).expect("write truncated");

        let error = TailHighWater::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(
            matches!(error, HighWaterError::WrongLength { .. }),
            "expected WrongLength, got {error}"
        );
    }

    #[test]
    fn a_flipped_checksum_byte_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1").write_atomic(dir.path()).expect("write");
        let mut bytes = std::fs::read(dir.path().join(FILE_NAME)).expect("read raw");
        let last = bytes.len() - 1;
        bytes[last] ^= 0xFF;
        std::fs::write(dir.path().join(FILE_NAME), &bytes).expect("write flipped");

        let error = TailHighWater::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, HighWaterError::ChecksumMismatch));
    }

    #[test]
    fn a_flipped_body_byte_is_refused_by_the_checksum() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1").write_atomic(dir.path()).expect("write");
        let mut bytes = std::fs::read(dir.path().join(FILE_NAME)).expect("read raw");
        bytes[10] ^= 0xFF;
        std::fs::write(dir.path().join(FILE_NAME), &bytes).expect("write flipped");

        let error = TailHighWater::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, HighWaterError::ChecksumMismatch));
    }

    #[test]
    fn another_familys_magic_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1").write_atomic(dir.path()).expect("write");
        let mut bytes = std::fs::read(dir.path().join(FILE_NAME)).expect("read raw");
        bytes[0] ^= 0xFF;
        let checksum = Sha256::digest(&bytes[..BOUND_LEN]);
        bytes[BOUND_LEN..].copy_from_slice(&checksum);
        std::fs::write(dir.path().join(FILE_NAME), &bytes).expect("write foreign magic");

        let error = TailHighWater::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, HighWaterError::WrongMagic));
    }

    #[test]
    fn an_unsupported_revision_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1").write_atomic(dir.path()).expect("write");
        let mut bytes = std::fs::read(dir.path().join(FILE_NAME)).expect("read raw");
        bytes[MAGIC.len()] = FORMAT_REVISION + 1;
        let checksum = Sha256::digest(&bytes[..BOUND_LEN]);
        bytes[BOUND_LEN..].copy_from_slice(&checksum);
        std::fs::write(dir.path().join(FILE_NAME), &bytes).expect("write future revision");

        let error = TailHighWater::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(
            error,
            HighWaterError::UnsupportedRevision {
                found: 2,
                expected: 1
            }
        ));
    }

    /// The artifact of one partition read for another must be refused, never
    /// silently accepted because the bytes happen to be well-formed.
    #[test]
    fn a_valid_artifact_for_a_different_partition_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1").write_atomic(dir.path()).expect("write");

        let error = TailHighWater::read(dir.path(), "conv-2").expect_err("must refuse");
        assert!(matches!(error, HighWaterError::WrongPartition));
    }

    #[test]
    fn accessors_return_exactly_what_was_bound() {
        let high_water = sample("conv-1");
        assert_eq!(high_water.incarnation(), [7_u8; 32]);
        assert_eq!(high_water.logical_head(), 42);
        assert_eq!(high_water.physical_frontier(), 44);
        assert_eq!(high_water.covering_root(), [9_u8; 32]);
        assert_eq!(high_water.covering_leaf_count(), 43);
    }
}