polyc-eventlog 2026.10.0

//! Durable "prune pass" artifact (U-4b-2): the crash-safe intent/receipt
//! pair a partition's physical byte reclamation records around the one
//! operation this crate cannot undo.
//!
//! That operation is [`commonware_storage::journal::contiguous::Mutable::prune`].
//!
//! # Why one file, two stages
//!
//! A crash can land at any of four points: before the intent lands, after
//! the intent but before the primitive runs, after the primitive but before
//! the receipt lands, or after the receipt but before the caller's floor
//! bookkeeping advances. The first point needs no record (nothing durable
//! happened yet) and the last needs none either (the record is invalidated
//! as that step's own last act — see [`PrunePass::invalidate`]). The middle
//! two are what this format exists to tell apart: [`PrunePassOutcome::Intent`]
//! means the primitive may or may not have run (calling
//! [`commonware_storage::journal::contiguous::Mutable::prune`] again with the
//! same or a lower position is always safe — pruning is idempotent, and the
//! journal itself never prunes past what is asked), while
//! [`PrunePassOutcome::Receipt`] means the primitive has already run and
//! only the floor bookkeeping remains. One file rather than two keeps the two
//! stages from ever disagreeing about which partition, incarnation, or
//! checkpoint they bind to — a receipt is built FROM its own intent
//! ([`PrunePass::into_receipt`]), never assembled independently.
//!
//! # What the receipt records
//!
//! The applied boundary, never the requested one.
//! [`commonware_storage::journal::contiguous::Mutable::prune`] "prunes at
//! most what is asked, never more": some items below the requested boundary
//! may survive section or blob alignment. Reading the primitive's own
//! post-prune bounds is what tells a caller the boundary it actually reached
//! ([`crate::EventLog::prune`] does this internally), and that is the number
//! this record carries forward — a receipt that echoed the request instead
//! would let a caller advance its floor past bytes the primitive silently
//! kept.
//!
//! # Format
//!
//! Fixed-width, the same discipline [`crate::highwater`] uses and for the
//! same reason: every field here is a single scalar, 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)
//!   | requested boundary (8, BE) | checkpoint digest (32)
//!   | stage (1: 0 = intent, 1 = receipt)
//!   | applied boundary (8, BE, always written, 0 while stage is intent)
//!   | checksum (32)
//! ```

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"PPP1";

/// 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 BOUNDARY_LEN: usize = 8;
const CHECKPOINT_DIGEST_LEN: usize = 32;
const CHECKSUM_LEN: usize = 32;

/// Everything the checksum covers.
const BOUND_LEN: usize = MAGIC.len()
    + 1 // revision
    + PARTITION_DIGEST_LEN
    + INCARNATION_LEN
    + BOUNDARY_LEN // requested
    + CHECKPOINT_DIGEST_LEN
    + 1 // stage
    + BOUNDARY_LEN; // applied
const RECORD_LEN: usize = BOUND_LEN + CHECKSUM_LEN;

/// The artifact's file name inside the caller-supplied directory — the same
/// directory [`crate::floor_checkpoint::FloorCheckpoint`] and
/// [`crate::highwater::TailHighWater`] use.
pub const FILE_NAME: &str = "prunepass";
/// Where the artifact is staged before the atomic rename.
const TEMP_FILE_NAME: &str = "prunepass.writing";

const STAGE_INTENT: u8 = 0;
const STAGE_RECEIPT: u8 = 1;

/// A durable, crash-safe record of one partition's physical prune attempt.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PrunePass {
    partition_digest: [u8; PARTITION_DIGEST_LEN],
    incarnation: [u8; INCARNATION_LEN],
    requested_boundary: u64,
    checkpoint_digest: [u8; CHECKPOINT_DIGEST_LEN],
    applied_boundary: Option<u64>,
}

/// What reading a partition's prune-pass artifact found.
///
/// [`Self::Absent`] means no pass is in flight: either none was ever
/// attempted, or the last one finished and its record was invalidated as
/// that finish's own last act. [`Self::Intent`] and [`Self::Receipt`] are
/// the two mid-flight states a resume distinguishes — see the module doc.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PrunePassOutcome {
    /// No prune pass is currently in flight for this partition.
    Absent,
    /// An intent is durable; the primitive may or may not have run yet.
    Intent(Box<PrunePass>),
    /// The primitive has run and its receipt is durable; only the caller's
    /// floor bookkeeping remains.
    Receipt(Box<PrunePass>),
}

/// Why a [`PrunePass`] failed to write, or a read refused what it found.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum PruneRecordError {
    /// The write or read itself failed at the filesystem.
    #[error("prune pass 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("prune pass 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("prune pass carries another family's magic")]
    WrongMagic,
    /// The file declares a format revision this build does not read.
    #[error("prune pass 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("prune pass fails its checksum")]
    ChecksumMismatch,
    /// The artifact names a different partition than the caller asked for.
    #[error("prune pass was recorded for a different partition")]
    WrongPartition,
    /// The file declares a stage byte this build does not recognize.
    #[error("prune pass declares an unknown stage {found}")]
    UnknownStage {
        /// The byte the file declared.
        found: u8,
    },
}

impl PrunePass {
    /// Opens a new intent: `partition`'s prune pass is starting, bound to
    /// `incarnation`, requesting `requested_boundary`, under the eligible
    /// checkpoint whose bytes hash to `checkpoint_digest`.
    #[must_use]
    pub fn intent(
        partition: &str,
        incarnation: [u8; INCARNATION_LEN],
        requested_boundary: u64,
        checkpoint_digest: [u8; CHECKPOINT_DIGEST_LEN],
    ) -> Self {
        Self {
            partition_digest: digest_partition(partition),
            incarnation,
            requested_boundary,
            checkpoint_digest,
            applied_boundary: None,
        }
    }

    /// Turns this intent into its receipt: the same binding, plus the
    /// boundary the primitive actually applied.
    ///
    /// Building the receipt from the intent, rather than constructing one
    /// independently, is what keeps the two from ever disagreeing about
    /// which partition, incarnation, or checkpoint they cover.
    #[must_use]
    pub const fn into_receipt(self, applied_boundary: u64) -> Self {
        Self {
            applied_boundary: Some(applied_boundary),
            ..self
        }
    }

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

    /// Returns the raw physical journal position this pass requested.
    #[must_use]
    pub const fn requested_boundary(&self) -> u64 {
        self.requested_boundary
    }

    /// Returns the digest of the floor checkpoint this pass ran under.
    #[must_use]
    pub const fn checkpoint_digest(&self) -> [u8; CHECKPOINT_DIGEST_LEN] {
        self.checkpoint_digest
    }

    /// Returns the boundary the primitive actually applied, once this pass
    /// has reached its receipt stage.
    #[must_use]
    pub const fn applied_boundary(&self) -> Option<u64> {
        self.applied_boundary
    }

    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 + BOUNDARY_LEN].copy_from_slice(&self.requested_boundary.to_be_bytes());
        at += BOUNDARY_LEN;
        bound[at..at + CHECKPOINT_DIGEST_LEN].copy_from_slice(&self.checkpoint_digest);
        at += CHECKPOINT_DIGEST_LEN;
        bound[at] = if self.applied_boundary.is_some() {
            STAGE_RECEIPT
        } else {
            STAGE_INTENT
        };
        at += 1;
        bound[at..at + BOUNDARY_LEN]
            .copy_from_slice(&self.applied_boundary.unwrap_or(0).to_be_bytes());
        at += BOUNDARY_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, PruneRecordError> {
        if bytes.len() != RECORD_LEN {
            return Err(PruneRecordError::WrongLength {
                expected: RECORD_LEN,
                found: bytes.len(),
            });
        }
        let (bound, checksum) = bytes.split_at(BOUND_LEN);
        let computed = Sha256::digest(bound);
        if computed.as_slice() != checksum {
            return Err(PruneRecordError::ChecksumMismatch);
        }
        let mut at = 0;
        if bound[at..at + MAGIC.len()] != MAGIC {
            return Err(PruneRecordError::WrongMagic);
        }
        at += MAGIC.len();
        let revision = bound[at];
        at += 1;
        if revision != FORMAT_REVISION {
            return Err(PruneRecordError::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(PruneRecordError::WrongPartition);
        }
        let incarnation: [u8; INCARNATION_LEN] = bound[at..at + INCARNATION_LEN]
            .try_into()
            .expect("slice is exactly INCARNATION_LEN bytes");
        at += INCARNATION_LEN;
        let requested_boundary = read_u64(bound, &mut at);
        let checkpoint_digest: [u8; CHECKPOINT_DIGEST_LEN] = bound[at..at + CHECKPOINT_DIGEST_LEN]
            .try_into()
            .expect("slice is exactly CHECKPOINT_DIGEST_LEN bytes");
        at += CHECKPOINT_DIGEST_LEN;
        let stage = bound[at];
        at += 1;
        let applied_boundary_value = read_u64(bound, &mut at);
        let applied_boundary = match stage {
            STAGE_INTENT => None,
            STAGE_RECEIPT => Some(applied_boundary_value),
            found => return Err(PruneRecordError::UnknownStage { found }),
        };

        Ok(Self {
            partition_digest,
            incarnation,
            requested_boundary,
            checkpoint_digest,
            applied_boundary,
        })
    }

    /// Writes this pass to `dir` atomically: temp file, `fsync`, rename,
    /// then `fsync` the directory.
    ///
    /// Replaces whatever `dir` held before. The caller serializes writes
    /// against the same directory; this performs no locking of its own.
    ///
    /// # Errors
    ///
    /// Returns [`PruneRecordError::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<(), PruneRecordError> {
        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 prune pass from `dir`.
    ///
    /// A genuinely absent file reads as [`PrunePassOutcome::Absent`]: no
    /// pass is in flight.
    ///
    /// # Errors
    ///
    /// Returns [`PruneRecordError::WrongLength`] for a torn write, never
    /// read as a shorter or empty valid record. Returns
    /// [`PruneRecordError::ChecksumMismatch`] for a flipped bit, and
    /// [`PruneRecordError::WrongMagic`], [`PruneRecordError::UnsupportedRevision`],
    /// [`PruneRecordError::WrongPartition`], or [`PruneRecordError::UnknownStage`]
    /// as their names describe.
    pub fn read(dir: &Path, partition: &str) -> Result<PrunePassOutcome, PruneRecordError> {
        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(PrunePassOutcome::Absent);
            }
            Err(error) => return Err(error.into()),
        };
        let pass = Self::from_bytes(&bytes, partition)?;
        Ok(match pass.applied_boundary {
            None => PrunePassOutcome::Intent(Box::new(pass)),
            Some(_) => PrunePassOutcome::Receipt(Box::new(pass)),
        })
    }

    /// Removes `partition`'s prune pass, where one exists.
    ///
    /// This is a finished pass's own last durable act: once the caller's
    /// floor bookkeeping has advanced, nothing about this pass needs
    /// resuming, and invalidating it is what lets the NEXT pass start from
    /// [`PrunePassOutcome::Absent`] rather than resuming a pass that already
    /// finished.
    ///
    /// # Errors
    ///
    /// Returns [`PruneRecordError::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<(), PruneRecordError> {
        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_intent(partition: &str) -> PrunePass {
        PrunePass::intent(partition, [7_u8; 32], 42, [9_u8; 32])
    }

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

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

        let outcome = PrunePass::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, PrunePassOutcome::Intent(Box::new(written)));
    }

    #[test]
    fn a_receipt_built_from_its_intent_round_trips_as_receipt() {
        let dir = tempfile::tempdir().expect("temp dir");
        let receipt = sample_intent("conv-1").into_receipt(40);
        receipt.write_atomic(dir.path()).expect("write");

        let outcome = PrunePass::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, PrunePassOutcome::Receipt(Box::new(receipt)));
        assert_eq!(receipt.applied_boundary(), Some(40));
    }

    #[test]
    fn a_receipt_overwrites_its_own_intent_rather_than_appending() {
        let dir = tempfile::tempdir().expect("temp dir");
        let intent = sample_intent("conv-1");
        intent.write_atomic(dir.path()).expect("write");
        let receipt = intent.into_receipt(41);
        receipt.write_atomic(dir.path()).expect("write");

        let outcome = PrunePass::read(dir.path(), "conv-1").expect("read");
        assert_eq!(outcome, PrunePassOutcome::Receipt(Box::new(receipt)));
        assert!(!dir.path().join(TEMP_FILE_NAME).exists());
    }

    #[test]
    fn invalidate_removes_a_present_pass_and_is_a_no_op_when_absent() {
        let dir = tempfile::tempdir().expect("temp dir");
        PrunePass::invalidate(dir.path()).expect("invalidate an absent pass");

        sample_intent("conv-1")
            .write_atomic(dir.path())
            .expect("write");
        PrunePass::invalidate(dir.path()).expect("invalidate a present pass");
        assert_eq!(
            PrunePass::read(dir.path(), "conv-1").unwrap(),
            PrunePassOutcome::Absent
        );
    }

    #[test]
    fn a_truncated_record_is_refused_not_read_as_shorter_or_empty() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample_intent("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 = PrunePass::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, PruneRecordError::WrongLength { .. }));
    }

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

    #[test]
    fn another_familys_magic_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample_intent("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[..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 = PrunePass::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, PruneRecordError::WrongMagic));
    }

    #[test]
    fn an_unsupported_revision_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample_intent("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[..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 = PrunePass::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(
            error,
            PruneRecordError::UnsupportedRevision {
                found: 2,
                expected: 1
            }
        ));
    }

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

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

    #[test]
    fn an_unknown_stage_byte_is_refused() {
        let dir = tempfile::tempdir().expect("temp dir");
        sample_intent("conv-1")
            .write_atomic(dir.path())
            .expect("write");
        let mut bytes = std::fs::read(dir.path().join(FILE_NAME)).expect("read raw");
        let stage_offset = BOUND_LEN - BOUNDARY_LEN - 1;
        bytes[stage_offset] = 2;
        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 unknown stage");

        let error = PrunePass::read(dir.path(), "conv-1").expect_err("must refuse");
        assert!(matches!(error, PruneRecordError::UnknownStage { found: 2 }));
    }

    #[test]
    fn accessors_return_exactly_what_was_bound() {
        let pass = sample_intent("conv-1");
        assert_eq!(pass.incarnation(), [7_u8; 32]);
        assert_eq!(pass.requested_boundary(), 42);
        assert_eq!(pass.checkpoint_digest(), [9_u8; 32]);
        assert_eq!(pass.applied_boundary(), None);
        let receipt = pass.into_receipt(40);
        assert_eq!(receipt.applied_boundary(), Some(40));
        assert_eq!(receipt.requested_boundary(), 42);
        assert_eq!(receipt.checkpoint_digest(), [9_u8; 32]);
    }
}