polyc-eventlog 2026.9.0

//! 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, returning the
    /// healthy store to keep.
    ///
    /// Callers must only pass a count already observed durably committed
    /// (see the module docs) — this store does not itself validate that.
    ///
    /// # Poisoning
    ///
    /// Consuming, and that is the point. Commonware states that an error from
    /// a mutable storage operation — `put`, `delete`, or `sync` — is
    /// unrecoverable, and the caller must not use that database instance
    /// again. `Metadata::sync` advances its in-memory cursor and next version
    /// before the writes that can fail, so a failed instance describes a
    /// durable state that does not exist.
    ///
    /// Taking `self` by value makes the rule structural rather than
    /// documented: on success the caller gets a store back, and on failure
    /// there is nothing to put back. The next attempt has to
    /// [`Self::open`] from durable storage.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Checkpoint`] if the durable sync fails, and
    /// drops the poisoned store with it.
    pub async fn record(mut self, count: u64) -> Result<Self, EventLogError> {
        self.store.put(EXPECTED_COUNT_KEY, count);
        self.store.sync().await?;
        Ok(self)
    }

    /// Clear the checkpoint because the partition it tracks was emptied 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. Returns the healthy store to keep.
    ///
    /// # Poisoning
    ///
    /// Consuming, for the reason [`Self::record`] gives.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Checkpoint`] if the durable sync fails, and
    /// drops the poisoned store with it.
    pub async fn clear(mut self) -> Result<Self, EventLogError> {
        self.store.clear();
        self.store.sync().await?;
        Ok(self)
    }
}

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

    /// 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 checkpoint = EventCountCheckpoint::open(context.child("first"), "conv-durable")
                    .await
                    .expect("open");
                checkpoint.record(7).await.expect("record");
            }

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

            let checkpoint = 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");
        });
    }

    /// A REAL Commonware `Metadata` sync failure consumes the store, and the
    /// durable floor is whatever last synced — never what the failed instance
    /// held in memory.
    ///
    /// Commonware's rule is that an error from a mutable storage operation is
    /// unrecoverable and the instance must not be used again. `Metadata::sync`
    /// advances its in-memory cursor and next version before the writes that
    /// can fail, so a reused instance would answer for a durable state that
    /// does not exist.
    ///
    /// The fault is injected into the deterministic runtime's storage, so the
    /// failure happens inside `Metadata::sync` itself. A test seam that
    /// returned before Commonware wrote anything would prove nothing about
    /// this rule.
    #[test]
    fn a_failed_sync_consumes_the_store_and_leaves_the_durable_floor_alone() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let checkpoint = EventCountCheckpoint::open(context.child("first"), "conv-poison")
                .await
                .expect("open");
            let checkpoint = checkpoint.record(5).await.expect("first record");

            *context.storage_fault_config().write() = FaultConfig::default().sync(1.0);
            // `match`, not `expect_err`: the success arm would carry the
            // store, which has no `Debug`, precisely because nothing should
            // ever print or inspect one.
            let Err(error) = checkpoint.record(9).await else {
                panic!("the durable sync must fail")
            };
            assert!(
                matches!(error, EventLogError::Checkpoint(_)),
                "expected a checkpoint storage error, got {error}"
            );
            // There is no `checkpoint` binding left to misuse: `record`
            // consumed it, and the failure path dropped it with the error
            // rather than handing it back. That is the eviction, and it is
            // structural rather than a rule a caller has to remember.

            *context.storage_fault_config().write() = FaultConfig::default();
            let reopened = EventCountCheckpoint::open(context.child("second"), "conv-poison")
                .await
                .expect("reopen");
            assert_eq!(
                reopened.expected_count(),
                5,
                "the floor is the last count that actually synced"
            );
        });
    }
}