polyc-eventlog 2026.10.0

//! Durable "floor checkpoint" artifact (U-4b-1): a partition's folded state
//! as of a retention floor, so `load` can resume from it instead of
//! replaying from position 0.
//!
//! # Why
//!
//! Every partition open replays the whole partition and rebuilds the MMR
//! and the integrity chain from genesis. Physically discarding bytes below
//! a floor (a later change, not this one) breaks that replay unless a
//! partition can resume from a durable checkpoint at the floor. This
//! artifact is that checkpoint: enough to resume the MMR from pinned peaks,
//! trust the folded prefix's root with one signature check instead of
//! re-verifying every historical root, and fold state a later read still
//! depends on (receipts, intents) without re-reading the pruned prefix.
//!
//! This PR prunes nothing. A caller may write this artifact and resume from
//! it purely as a read-path optimization; the bytes below the floor remain
//! on disk and a full replay from position 0 remains available as ground
//! truth, which is what lets this PR's own tests prove resume-then-replay
//! reproduces a full replay's exact result.
//!
//! # Why not [`crate::highwater`]'s fixed-width shape
//!
//! [`crate::highwater::TailHighWater`] is fixed-width because every field it
//! carries is a single scalar. This artifact also carries a partition's
//! *pinned MMR peaks* (a handful of digests, bounded by
//! [`commonware_storage::merkle::MAX_PINNED_NODES`]) and an *opaque caller
//! payload* (the folded receipts, destruction/repair/excision intents, and
//! their fold order a resumed `state-service`-shaped partition still needs
//! to answer a retry identically to a full replay) — both variable-length
//! and both unbounded in the general case. So this format is length-framed
//! rather than fixed-width, with the same discipline: a torn write is
//! refused by a checksum over the whole record, never read as a shorter or
//! empty one.
//!
//! This crate does not interpret the payload. `state-service` writes and
//! reads it; to this crate it is just bytes bounded by
//! [`MAX_PAYLOAD_BYTES`].
//!
//! # Format
//!
//! ```text
//! magic (4) | revision (1) | partition digest (32) | incarnation (32)
//!   | floor position (8, BE) | floor journal position (8, BE)
//!   | floor leaf count (8, BE)
//!   | covering root (32) | covering leaf count (8, BE)
//!   | covering signature (64) | covering signer (32)
//!   | fence present (1) | fence value (8, BE, always written)
//!   | pinned peak count (1) | pinned peaks (32 each)
//!   | payload length (8, BE) | payload (variable)
//!   | checksum (32)
//! ```
//!
//! Three different numbers name "how far this checkpoint reaches," and the
//! format keeps all three because no one of them derives the others:
//! **floor position** is the contract position
//! (`polyc_state::revision::JournalPosition`) `state-service`'s own head and
//! feed floor are counted in; **floor journal position** is the RAW
//! physical event count the underlying
//! `commonware_storage::journal::contiguous::variable::Journal` counts in —
//! larger, because every commit's signed-root marker consumes a raw
//! position without advancing the contract head, and it is what a resuming
//! caller passes to replay the retained suffix; **floor leaf count** is the
//! MMR leaf count — smaller than the raw journal position for the same
//! reason, since a root marker is never an MMR leaf
//! (`polyc_eventlog_model::integrity::rebuild_from_events` filters them
//! out).
//!
//! The checksum covers every byte before it, whatever the pinned-peak count
//! or payload length turned out to be — the same discipline
//! [`crate::highwater`] uses, generalized to a variable record: read the
//! last 32 bytes as the checksum over everything else, and refuse before
//! trusting a single other field.

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

use commonware_storage::merkle::MAX_PINNED_NODES;
use sha2::{Digest as _, Sha256};

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

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

/// The longest opaque payload this artifact carries.
///
/// A generous ceiling on the folded receipts/intents a partition may still
/// need after resuming from its floor — high enough that a legitimate
/// partition never approaches it, low enough that a corrupt length field
/// cannot make a read attempt an enormous allocation. 64 MiB.
pub const MAX_PAYLOAD_BYTES: u64 = 64 * 1024 * 1024;

const PARTITION_DIGEST_LEN: usize = 32;
const INCARNATION_LEN: usize = 32;
const ROOT_LEN: usize = 32;
/// Encoded ed25519 signature bytes.
const SIGNATURE_LEN: usize = 64;
/// Encoded ed25519 public-key bytes.
const SIGNER_LEN: usize = 32;
const PEAK_LEN: usize = 32;
const CHECKSUM_LEN: usize = 32;

/// The fixed-width prefix every record starts with, before the variable
/// pinned-peaks and payload sections.
const FIXED_PREFIX_LEN: usize = MAGIC.len()
    + 1 // revision
    + PARTITION_DIGEST_LEN
    + INCARNATION_LEN
    + 8 // floor position
    + 8 // floor journal position
    + 8 // floor leaf count
    + ROOT_LEN
    + 8 // covering leaf count
    + SIGNATURE_LEN
    + SIGNER_LEN
    + 1 // fence present
    + 8 // fence value
    + 1; // pinned peak count

/// The artifact's file name inside the caller-supplied directory.
pub const FILE_NAME: &str = "floorcheckpoint";
/// Where the artifact is staged before the atomic rename.
const TEMP_FILE_NAME: &str = "floorcheckpoint.writing";

/// One partition's durable floor checkpoint: enough to resume from the floor
/// instead of replaying from position 0.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FloorCheckpoint {
    partition_digest: [u8; PARTITION_DIGEST_LEN],
    incarnation: [u8; INCARNATION_LEN],
    floor_position: u64,
    floor_journal_position: u64,
    floor_leaf_count: u64,
    covering_root: [u8; ROOT_LEN],
    covering_leaf_count: u64,
    covering_signature: [u8; SIGNATURE_LEN],
    covering_signer: [u8; SIGNER_LEN],
    fence: Option<u64>,
    pinned_peaks: Vec<[u8; PEAK_LEN]>,
    payload: Vec<u8>,
}

/// Why a [`FloorCheckpoint`] failed to build, write, or refused to read.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum FloorCheckpointError {
    /// The write or read itself failed at the filesystem.
    #[error("floor checkpoint i/o failure: {0}")]
    Io(#[from] std::io::Error),
    /// A caller tried to build a record with more pinned peaks than
    /// [`MAX_PINNED_NODES`] allows.
    #[error("floor checkpoint carries {found} pinned peaks, more than {max}")]
    TooManyPinnedPeaks {
        /// How many peaks the caller supplied.
        found: usize,
        /// The most this format ever carries.
        max: usize,
    },
    /// A caller tried to build a record with a payload longer than
    /// [`MAX_PAYLOAD_BYTES`].
    #[error("floor checkpoint payload is {found} byte(s), more than {max}")]
    PayloadTooLarge {
        /// How many bytes the caller supplied.
        found: u64,
        /// The most this format ever carries.
        max: u64,
    },
    /// The file is shorter than the fixed prefix plus a checksum could ever
    /// be — a torn write, a foreign file, or a hand-edited one. Never read
    /// as if it were absent.
    #[error("floor checkpoint is {found} byte(s), shorter than the minimum {minimum}")]
    TooShort {
        /// The shortest a valid record can be (empty peaks, empty payload).
        minimum: usize,
        /// The length the file actually was.
        found: usize,
    },
    /// The fixed-width prefix names a pinned-peak count or payload length
    /// that does not match how many bytes actually follow.
    #[error("floor checkpoint declares a length its own byte count does not match")]
    LengthMismatch,
    /// The file does not carry this family's magic.
    #[error("floor checkpoint carries another family's magic")]
    WrongMagic,
    /// The file declares a format revision this build does not read.
    #[error("floor checkpoint 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("floor checkpoint fails its checksum")]
    ChecksumMismatch,
    /// The artifact names a different partition than the caller asked for.
    #[error("floor checkpoint was recorded for a different partition")]
    WrongPartition,
}

/// What reading a partition's floor checkpoint found.
///
/// [`Self::Absent`] and every error are deliberately different
/// constructible states: a partition with no checkpoint yet always replays
/// from position 0 (legitimate — this PR prunes nothing, so absence is
/// never itself damage), while a present-but-unreadable checkpoint is
/// refused rather than silently treated as absent, which would resume a
/// caller from a floor that was never actually folded.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FloorCheckpointOutcome {
    /// No checkpoint has ever been written in this directory.
    Absent,
    /// The durable checkpoint this partition last recorded.
    Present(Box<FloorCheckpoint>),
}

impl FloorCheckpoint {
    /// Binds `partition`'s floor checkpoint.
    ///
    /// # Errors
    ///
    /// Returns [`FloorCheckpointError::TooManyPinnedPeaks`] if `pinned_peaks`
    /// carries more than [`MAX_PINNED_NODES`] entries. Returns
    /// [`FloorCheckpointError::PayloadTooLarge`] if `payload` is longer than
    /// [`MAX_PAYLOAD_BYTES`].
    #[allow(
        clippy::too_many_arguments,
        reason = "every field is load-bearing and named"
    )]
    pub fn try_new(
        partition: &str,
        incarnation: [u8; INCARNATION_LEN],
        floor_position: u64,
        floor_journal_position: u64,
        floor_leaf_count: u64,
        covering_root: [u8; ROOT_LEN],
        covering_leaf_count: u64,
        covering_signature: [u8; SIGNATURE_LEN],
        covering_signer: [u8; SIGNER_LEN],
        fence: Option<u64>,
        pinned_peaks: Vec<[u8; PEAK_LEN]>,
        payload: Vec<u8>,
    ) -> Result<Self, FloorCheckpointError> {
        if pinned_peaks.len() > MAX_PINNED_NODES {
            return Err(FloorCheckpointError::TooManyPinnedPeaks {
                found: pinned_peaks.len(),
                max: MAX_PINNED_NODES,
            });
        }
        let payload_len = u64::try_from(payload.len()).unwrap_or(u64::MAX);
        if payload_len > MAX_PAYLOAD_BYTES {
            return Err(FloorCheckpointError::PayloadTooLarge {
                found: payload_len,
                max: MAX_PAYLOAD_BYTES,
            });
        }
        Ok(Self {
            partition_digest: digest_partition(partition),
            incarnation,
            floor_position,
            floor_journal_position,
            floor_leaf_count,
            covering_root,
            covering_leaf_count,
            covering_signature,
            covering_signer,
            fence,
            pinned_peaks,
            payload,
        })
    }

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

    /// Returns the contract position of the checkpoint's own base state — the
    /// last position folded into it. Comparable to
    /// `polyc_state::revision::JournalPosition` (`state-service`'s own head
    /// and feed floor), never to a raw journal position — see
    /// [`Self::floor_journal_position`] for that.
    #[must_use]
    pub const fn floor_position(&self) -> u64 {
        self.floor_position
    }

    /// Returns the raw physical journal position (event count, root
    /// markers included) of the checkpoint's own base state. A resuming
    /// caller passes this to whatever replays the retained suffix from the
    /// underlying physical journal — never [`Self::floor_position`], which
    /// counts a different, smaller thing.
    #[must_use]
    pub const fn floor_journal_position(&self) -> u64 {
        self.floor_journal_position
    }

    /// Returns the MMR leaf count at that same position.
    #[must_use]
    pub const fn floor_leaf_count(&self) -> u64 {
        self.floor_leaf_count
    }

    /// Returns the covering signed root over the folded prefix.
    #[must_use]
    pub const fn covering_root(&self) -> [u8; ROOT_LEN] {
        self.covering_root
    }

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

    /// Returns the covering root's ed25519 signature.
    #[must_use]
    pub const fn covering_signature(&self) -> [u8; SIGNATURE_LEN] {
        self.covering_signature
    }

    /// Returns the covering root's ed25519 signer public key.
    #[must_use]
    pub const fn covering_signer(&self) -> [u8; SIGNER_LEN] {
        self.covering_signer
    }

    /// Returns the highest fence the folded prefix presented, where any
    /// command in it presented one.
    #[must_use]
    pub const fn fence(&self) -> Option<u64> {
        self.fence
    }

    /// Returns the pinned MMR peak digests needed to resume the tree at
    /// [`Self::floor_leaf_count`], in [`commonware_storage::merkle::Family::nodes_to_pin`]
    /// order.
    #[must_use]
    pub fn pinned_peaks(&self) -> &[[u8; PEAK_LEN]] {
        &self.pinned_peaks
    }

    /// Returns the caller-owned opaque payload — this crate never interprets
    /// it.
    #[must_use]
    pub fn payload(&self) -> &[u8] {
        &self.payload
    }

    fn to_bytes(&self) -> Vec<u8> {
        let mut bound = Vec::with_capacity(
            FIXED_PREFIX_LEN + self.pinned_peaks.len() * PEAK_LEN + 8 + self.payload.len(),
        );
        bound.extend_from_slice(&MAGIC);
        bound.push(FORMAT_REVISION);
        bound.extend_from_slice(&self.partition_digest);
        bound.extend_from_slice(&self.incarnation);
        bound.extend_from_slice(&self.floor_position.to_be_bytes());
        bound.extend_from_slice(&self.floor_journal_position.to_be_bytes());
        bound.extend_from_slice(&self.floor_leaf_count.to_be_bytes());
        bound.extend_from_slice(&self.covering_root);
        bound.extend_from_slice(&self.covering_leaf_count.to_be_bytes());
        bound.extend_from_slice(&self.covering_signature);
        bound.extend_from_slice(&self.covering_signer);
        bound.push(u8::from(self.fence.is_some()));
        bound.extend_from_slice(&self.fence.unwrap_or(0).to_be_bytes());
        #[allow(
            clippy::cast_possible_truncation,
            reason = "try_new already refused more than MAX_PINNED_NODES (< 256) peaks"
        )]
        bound.push(self.pinned_peaks.len() as u8);
        for peak in &self.pinned_peaks {
            bound.extend_from_slice(peak);
        }
        let payload_len = u64::try_from(self.payload.len()).unwrap_or(u64::MAX);
        bound.extend_from_slice(&payload_len.to_be_bytes());
        bound.extend_from_slice(&self.payload);

        let checksum = Sha256::digest(&bound);
        bound.extend_from_slice(&checksum);
        bound
    }

    #[allow(clippy::too_many_lines, reason = "one linear field-by-field decode")]
    fn from_bytes(bytes: &[u8], partition: &str) -> Result<Self, FloorCheckpointError> {
        let minimum = FIXED_PREFIX_LEN + 8 + CHECKSUM_LEN;
        if bytes.len() < minimum {
            return Err(FloorCheckpointError::TooShort {
                minimum,
                found: bytes.len(),
            });
        }
        let (bound, checksum) = bytes.split_at(bytes.len() - CHECKSUM_LEN);
        // The checksum is checked before a single other field is trusted: a
        // torn write can flip the magic, the revision, or a length field
        // just as easily as any other byte.
        let computed = Sha256::digest(bound);
        if computed.as_slice() != checksum {
            return Err(FloorCheckpointError::ChecksumMismatch);
        }
        if bound.len() < FIXED_PREFIX_LEN {
            return Err(FloorCheckpointError::LengthMismatch);
        }
        let mut at = 0;
        if bound[at..at + MAGIC.len()] != MAGIC {
            return Err(FloorCheckpointError::WrongMagic);
        }
        at += MAGIC.len();
        let revision = bound[at];
        at += 1;
        if revision != FORMAT_REVISION {
            return Err(FloorCheckpointError::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(FloorCheckpointError::WrongPartition);
        }
        let incarnation: [u8; INCARNATION_LEN] = bound[at..at + INCARNATION_LEN]
            .try_into()
            .expect("slice is exactly INCARNATION_LEN bytes");
        at += INCARNATION_LEN;
        let floor_position = read_u64(bound, &mut at);
        let floor_journal_position = read_u64(bound, &mut at);
        let floor_leaf_count = read_u64(bound, &mut at);
        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 = read_u64(bound, &mut at);
        let covering_signature: [u8; SIGNATURE_LEN] = bound[at..at + SIGNATURE_LEN]
            .try_into()
            .expect("slice is exactly SIGNATURE_LEN bytes");
        at += SIGNATURE_LEN;
        let covering_signer: [u8; SIGNER_LEN] = bound[at..at + SIGNER_LEN]
            .try_into()
            .expect("slice is exactly SIGNER_LEN bytes");
        at += SIGNER_LEN;
        let fence_present = bound[at] != 0;
        at += 1;
        let fence_value = read_u64(bound, &mut at);
        let fence = fence_present.then_some(fence_value);
        let peak_count = usize::from(bound[at]);
        at += 1;

        let expected_len = at
            .checked_add(peak_count.saturating_mul(PEAK_LEN))
            .and_then(|len| len.checked_add(8))
            .ok_or(FloorCheckpointError::LengthMismatch)?;
        if bound.len() < expected_len {
            return Err(FloorCheckpointError::LengthMismatch);
        }
        let mut pinned_peaks = Vec::with_capacity(peak_count);
        for _ in 0..peak_count {
            let peak: [u8; PEAK_LEN] = bound[at..at + PEAK_LEN]
                .try_into()
                .expect("slice is exactly PEAK_LEN bytes");
            pinned_peaks.push(peak);
            at += PEAK_LEN;
        }
        let payload_len = read_u64(bound, &mut at);
        let payload_len_usize = usize::try_from(payload_len).map_err(|_| {
            // Unreachable on any host where usize is at least 32 bits and the
            // length check below would already refuse; kept explicit rather
            // than panicking on a hostile length field.
            FloorCheckpointError::LengthMismatch
        })?;
        if bound.len() != at + payload_len_usize {
            return Err(FloorCheckpointError::LengthMismatch);
        }
        let payload = bound[at..at + payload_len_usize].to_vec();

        Ok(Self {
            partition_digest,
            incarnation,
            floor_position,
            floor_journal_position,
            floor_leaf_count,
            covering_root,
            covering_leaf_count,
            covering_signature,
            covering_signer,
            fence,
            pinned_peaks,
            payload,
        })
    }

    /// Writes this checkpoint 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 — the folded prefix and
    /// its covering signed root — is itself durable, and for serializing
    /// writes against the same directory: this performs no locking of its
    /// own.
    ///
    /// # Errors
    ///
    /// Returns [`FloorCheckpointError::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<(), FloorCheckpointError> {
        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 floor checkpoint from `dir`.
    ///
    /// A genuinely absent file reads as [`FloorCheckpointOutcome::Absent`] —
    /// always legitimate for this PR, which prunes nothing.
    ///
    /// # Errors
    ///
    /// Returns [`FloorCheckpointError::TooShort`] or
    /// [`FloorCheckpointError::LengthMismatch`] for a torn write, never read
    /// as a shorter or empty valid record. Returns
    /// [`FloorCheckpointError::ChecksumMismatch`] for a flipped bit, and
    /// [`FloorCheckpointError::WrongMagic`], [`FloorCheckpointError::UnsupportedRevision`],
    /// or [`FloorCheckpointError::WrongPartition`] as their names describe.
    pub fn read(
        dir: &Path,
        partition: &str,
    ) -> Result<FloorCheckpointOutcome, FloorCheckpointError> {
        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(FloorCheckpointOutcome::Absent);
            }
            Err(error) => return Err(error.into()),
        };
        Self::from_bytes(&bytes, partition)
            .map(|checkpoint| FloorCheckpointOutcome::Present(Box::new(checkpoint)))
    }

    /// Removes `partition`'s floor checkpoint, where one exists.
    ///
    /// Used to invalidate a checkpoint a repair or excision has made stale:
    /// absence is always a legitimate, load-bearing state for this format
    /// (a full replay from position 0 is always available), so deleting one
    /// forward-heals `load` to the safe default rather than leaving it
    /// resuming from a prefix that may no longer be what the checkpoint
    /// claims.
    ///
    /// # Errors
    ///
    /// Returns [`FloorCheckpointError::Io`] if the file exists and cannot be
    /// removed, or if the directory cannot be synced afterward. A file that
    /// was already absent is not an error.
    pub fn invalidate(dir: &Path) -> Result<(), FloorCheckpointError> {
        match std::fs::remove_file(dir.join(FILE_NAME)) {
            Ok(()) => {}
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
            Err(error) => return Err(error.into()),
        }
        File::open(dir)?.sync_all()?;
        Ok(())
    }
}

fn read_u64(bound: &[u8], at: &mut usize) -> u64 {
    let value = u64::from_be_bytes(
        bound[*at..*at + 8]
            .try_into()
            .expect("slice is exactly 8 bytes"),
    );
    *at += 8;
    value
}

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, peaks: usize, payload: &[u8]) -> FloorCheckpoint {
        FloorCheckpoint::try_new(
            partition,
            [7_u8; 32],
            42,
            45,
            40,
            [9_u8; 32],
            41,
            [3_u8; SIGNATURE_LEN],
            [4_u8; SIGNER_LEN],
            Some(5),
            (0..peaks)
                .map(|i| [u8::try_from(i).expect("test peak counts stay below 256"); PEAK_LEN])
                .collect(),
            payload.to_vec(),
        )
        .expect("a within-bounds checkpoint builds")
    }

    #[test]
    fn a_fresh_directory_reads_as_absent_not_an_error() {
        let dir = tempfile::tempdir().expect("temp dir");
        let outcome = FloorCheckpoint::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, FloorCheckpointOutcome::Absent);
    }

    #[test]
    fn a_written_checkpoint_survives_a_round_trip_with_peaks_and_payload() {
        let dir = tempfile::tempdir().expect("temp dir");
        let written = sample("conv-1", 5, b"folded-receipts-and-intents");
        written.write_atomic(dir.path()).expect("write");

        let outcome = FloorCheckpoint::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, FloorCheckpointOutcome::Present(Box::new(written)));
    }

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

        let outcome = FloorCheckpoint::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, FloorCheckpointOutcome::Present(Box::new(written)));
    }

    #[test]
    fn a_second_write_replaces_the_first_rather_than_appending() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1", 3, b"first")
            .write_atomic(dir.path())
            .expect("write");
        let second = sample("conv-1", 1, b"second-and-different-length-payload");
        second.write_atomic(dir.path()).expect("write");

        let outcome = FloorCheckpoint::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, FloorCheckpointOutcome::Present(Box::new(second)));
        assert!(!dir.path().join(TEMP_FILE_NAME).exists());
    }

    #[test]
    fn too_many_pinned_peaks_refuses_construction() {
        let error = FloorCheckpoint::try_new(
            "conv-1",
            [0; 32],
            0,
            0,
            0,
            [0; 32],
            0,
            [0; SIGNATURE_LEN],
            [0; SIGNER_LEN],
            None,
            vec![[0; PEAK_LEN]; MAX_PINNED_NODES + 1],
            vec![],
        )
        .expect_err("too many peaks must refuse");
        assert!(matches!(
            error,
            FloorCheckpointError::TooManyPinnedPeaks { found, max }
                if found == MAX_PINNED_NODES + 1 && max == MAX_PINNED_NODES
        ));
    }

    #[test]
    fn a_payload_past_the_ceiling_refuses_construction() {
        let error = FloorCheckpoint::try_new(
            "conv-1",
            [0; 32],
            0,
            0,
            0,
            [0; 32],
            0,
            [0; SIGNATURE_LEN],
            [0; SIGNER_LEN],
            None,
            vec![],
            vec![0_u8; usize::try_from(MAX_PAYLOAD_BYTES).unwrap() + 1],
        )
        .expect_err("an oversized payload must refuse");
        assert!(matches!(
            error,
            FloorCheckpointError::PayloadTooLarge { .. }
        ));
    }

    #[test]
    fn a_truncated_record_is_refused_not_read_as_shorter_or_empty() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1", 2, b"payload")
            .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 = FloorCheckpoint::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(
            matches!(error, FloorCheckpointError::ChecksumMismatch),
            "a byte dropped from the end changes the checksummed region; got {error}"
        );
    }

    #[test]
    fn a_record_shorter_than_the_fixed_prefix_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        std::fs::create_dir_all(dir.path()).unwrap();
        std::fs::write(dir.path().join(FILE_NAME), [0_u8; 4]).expect("write short file");

        let error = FloorCheckpoint::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, FloorCheckpointError::TooShort { .. }));
    }

    #[test]
    fn a_flipped_checksum_byte_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1", 2, b"payload")
            .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 = FloorCheckpoint::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, FloorCheckpointError::ChecksumMismatch));
    }

    #[test]
    fn a_flipped_body_byte_is_refused_by_the_checksum() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1", 2, b"payload")
            .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 = FloorCheckpoint::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, FloorCheckpointError::ChecksumMismatch));
    }

    #[test]
    fn another_familys_magic_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1", 0, b"")
            .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[..bytes.len() - CHECKSUM_LEN]);
        let new_len = bytes.len() - CHECKSUM_LEN;
        bytes[new_len..].copy_from_slice(&checksum);
        std::fs::write(dir.path().join(FILE_NAME), &bytes).expect("write foreign magic");

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

    #[test]
    fn an_unsupported_revision_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample("conv-1", 0, b"")
            .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[..bytes.len() - CHECKSUM_LEN]);
        let new_len = bytes.len() - CHECKSUM_LEN;
        bytes[new_len..].copy_from_slice(&checksum);
        std::fs::write(dir.path().join(FILE_NAME), &bytes).expect("write future revision");

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

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

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

    /// A peak count or payload length that the checksum verifies (so it is
    /// not a torn write) but that does not match how many bytes actually
    /// follow — the shape a hand-edited length field leaves.
    #[test]
    fn an_internally_inconsistent_length_is_refused_not_misread() {
        let dir = tempfile::tempdir().expect("temp dir");
        let written = sample("conv-1", 2, b"payload");
        written.write_atomic(dir.path()).expect("write");
        let mut bytes = std::fs::read(dir.path().join(FILE_NAME)).expect("read raw");
        // The peak-count byte sits right after the fence value, at a fixed
        // offset from the start of the fixed prefix.
        let peak_count_offset = FIXED_PREFIX_LEN - 1;
        bytes[peak_count_offset] = 60; // claims 60 peaks; far fewer bytes follow
        let checksum = Sha256::digest(&bytes[..bytes.len() - CHECKSUM_LEN]);
        let new_len = bytes.len() - CHECKSUM_LEN;
        bytes[new_len..].copy_from_slice(&checksum);
        std::fs::write(dir.path().join(FILE_NAME), &bytes).expect("write inconsistent length");

        let error = FloorCheckpoint::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, FloorCheckpointError::LengthMismatch));
    }

    #[test]
    fn invalidate_removes_a_present_checkpoint_and_is_a_no_op_when_absent() {
        let dir = tempfile::tempdir().expect("temp dir");
        assert_eq!(
            FloorCheckpoint::read(dir.path(), "conv-1").unwrap(),
            FloorCheckpointOutcome::Absent
        );
        FloorCheckpoint::invalidate(dir.path()).expect("invalidate an absent checkpoint");

        sample("conv-1", 1, b"x")
            .write_atomic(dir.path())
            .expect("write");
        assert!(matches!(
            FloorCheckpoint::read(dir.path(), "conv-1").unwrap(),
            FloorCheckpointOutcome::Present(_)
        ));
        FloorCheckpoint::invalidate(dir.path()).expect("invalidate a present checkpoint");
        assert_eq!(
            FloorCheckpoint::read(dir.path(), "conv-1").unwrap(),
            FloorCheckpointOutcome::Absent
        );
    }

    #[test]
    fn accessors_return_exactly_what_was_bound() {
        let checkpoint = sample("conv-1", 2, b"payload");
        assert_eq!(checkpoint.incarnation(), [7_u8; 32]);
        assert_eq!(checkpoint.floor_position(), 42);
        assert_eq!(checkpoint.floor_journal_position(), 45);
        assert_eq!(checkpoint.floor_leaf_count(), 40);
        assert_eq!(checkpoint.covering_root(), [9_u8; 32]);
        assert_eq!(checkpoint.covering_leaf_count(), 41);
        assert_eq!(checkpoint.covering_signature(), [3_u8; SIGNATURE_LEN]);
        assert_eq!(checkpoint.covering_signer(), [4_u8; SIGNER_LEN]);
        assert_eq!(checkpoint.fence(), Some(5));
        assert_eq!(
            checkpoint.pinned_peaks(),
            &[[0_u8; PEAK_LEN], [1_u8; PEAK_LEN]]
        );
        assert_eq!(checkpoint.payload(), b"payload");
    }

    #[test]
    fn an_absent_fence_round_trips_as_absent() {
        let dir = tempfile::tempdir().expect("temp dir");
        let checkpoint = FloorCheckpoint::try_new(
            "conv-1",
            [1; 32],
            1,
            1,
            1,
            [1; 32],
            1,
            [1; SIGNATURE_LEN],
            [1; SIGNER_LEN],
            None,
            vec![],
            vec![],
        )
        .unwrap();
        checkpoint.write_atomic(dir.path()).expect("write");
        let FloorCheckpointOutcome::Present(read_back) =
            FloorCheckpoint::read(dir.path(), "conv-1").unwrap()
        else {
            panic!("expected a present checkpoint");
        };
        assert_eq!(read_back.fence(), None);
    }
}