polyc-eventlog 2026.8.3

//! Durable "expected event count" side-channel (`#799` hardening).
//!
//! [`crate::EventLog`]'s replay can be silently shortened by the underlying
//! journal's own crash recovery: on open, `commonware-storage`'s
//! `variable::Journal` re-derives its still-open (active) section's item
//! boundaries by sequentially re-parsing it from byte zero, and if that scan
//! ever fails to decode an item it cannot tell "one item mid-section is
//! corrupt" apart from "this whole section may be an in-flight torn write" —
//! so it silently rewinds, which can invalidate the ENTIRE active section,
//! not just a truncated tail (empirically: even an event that PRECEDES the
//! corrupted one is lost when everything lives in one still-open section).
//! That is the right call for what the journal itself can see (a real crash
//! never durably committed content it cannot cleanly re-derive), but the
//! journal cannot tell "recovered from a crash" apart from "recovered from a
//! byte flipped in already-durable data" — both look identical on disk. Left
//! unchecked, a caller that only consults the journal after this self-heal
//! sees a shorter-than-expected (or empty) replay with no signal anything is
//! wrong, which is exactly the false pass a tamper-evidence check must not
//! produce.
//!
//! [`EventCountCheckpoint`] closes that gap by durably recording, in a
//! store SEPARATE from the partition's own journal files (a
//! [`commonware_storage::metadata::Metadata`] instance — CRC32-checked,
//! dual-blob rotation, so a single corrupted blob falls back to the other
//! still-valid one instead of silently truncating), the event count last
//! observed after a legitimate, durably-committed mutation. A later replay
//! whose count is lower than the last recorded checkpoint has lost
//! already-committed data — the signature of active-section corruption, not
//! an ordinary crash — and the caller must treat that as a hard failure.
//!
//! This is deliberately a MINIMUM bound, not an exact one: a crash between a
//! successful journal `commit()` and this checkpoint's own `sync()` leaves
//! the checkpoint stale (lower than reality), which only weakens detection
//! for that one commit — the checkpoint is only ever raised to a count the
//! caller has already durably observed, so a stale value can never manufacture
//! a false failure.

use commonware_storage::metadata::{Config as MetadataConfig, Metadata};
use commonware_utils::sequence::U64;

use crate::EventLogError;

/// Fixed key under which the expected event count is stored — one value per
/// checkpoint store, so a well-known key is all that is needed.
const EXPECTED_COUNT_KEY: U64 = U64::new(0);

/// Suffix distinguishing a partition's checkpoint store from its journal's
/// own `{partition}_data` / `{partition}_offsets` storage directories. Ends
/// without a `_data` suffix, so a directory listing keyed on that suffix
/// (partition discovery) never picks this side-store up as a logical
/// conversation partition.
const CHECKPOINT_PARTITION_SUFFIX: &str = "__eventcount_checkpoint";

/// A durable "last known committed event count" for one conversation
/// partition, stored independently of that partition's own journal files.
///
/// See the module docs for why this exists and the guarantee it provides.
pub struct EventCountCheckpoint<E: commonware_storage::Context> {
    store: Metadata<E, U64, u64>,
}

impl<E: commonware_storage::Context> EventCountCheckpoint<E> {
    /// Open (or create) the checkpoint store tracking `partition`.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Checkpoint`] if the underlying metadata store
    /// fails to initialize.
    pub async fn open(context: E, partition: &str) -> Result<Self, EventLogError> {
        let store = Metadata::init(
            context,
            MetadataConfig {
                partition: format!("{partition}{CHECKPOINT_PARTITION_SUFFIX}"),
                codec_config: (),
            },
        )
        .await?;
        Ok(Self { store })
    }

    /// The last durably recorded event count, or `0` if none was ever
    /// recorded — a fresh partition, or one whose checkpoint was cleared by
    /// [`Self::clear`].
    #[must_use]
    pub fn expected_count(&self) -> u64 {
        self.store.get(&EXPECTED_COUNT_KEY).copied().unwrap_or(0)
    }

    /// Durably record `count` as the new expected minimum.
    ///
    /// Callers must only pass a count already observed durably committed
    /// (see the module docs) — this store does not itself validate that.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Checkpoint`] if the durable sync fails.
    pub async fn record(&mut self, count: u64) -> Result<(), EventLogError> {
        self.store.put(EXPECTED_COUNT_KEY, count);
        self.store.sync().await?;
        Ok(())
    }

    /// Clear the checkpoint because the partition it tracks was destroyed or
    /// erased, so a subsequent [`Self::open`] of a reused partition name
    /// starts fresh at `0` rather than remembering content the partition no
    /// longer has.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Checkpoint`] if the durable sync fails.
    pub async fn clear(&mut self) -> Result<(), EventLogError> {
        self.store.clear();
        self.store.sync().await?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use commonware_runtime::{Runner as _, Supervisor as _, deterministic};

    /// A fresh checkpoint has no recorded expectation — verify must not treat
    /// "never recorded" as "corrupted".
    #[test]
    fn fresh_checkpoint_expects_zero() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let checkpoint = EventCountCheckpoint::open(context, "conv-fresh")
                .await
                .expect("open");
            assert_eq!(checkpoint.expected_count(), 0);
        });
    }

    /// A recorded count survives a reopen (durability), and `clear` resets it.
    #[test]
    fn record_survives_reopen_and_clear_resets_it() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            {
                let mut checkpoint =
                    EventCountCheckpoint::open(context.child("first"), "conv-durable")
                        .await
                        .expect("open");
                checkpoint.record(7).await.expect("record");
            }

            let mut checkpoint =
                EventCountCheckpoint::open(context.child("second"), "conv-durable")
                    .await
                    .expect("reopen");
            assert_eq!(checkpoint.expected_count(), 7, "survives reopen");

            checkpoint.clear().await.expect("clear");
            assert_eq!(checkpoint.expected_count(), 0, "cleared");

            let checkpoint = EventCountCheckpoint::open(context.child("third"), "conv-durable")
                .await
                .expect("reopen after clear");
            assert_eq!(checkpoint.expected_count(), 0, "clear is durable");
        });
    }
}