use commonware_storage::metadata::{Config as MetadataConfig, Metadata};
use commonware_utils::sequence::U64;
use crate::EventLogError;
const EXPECTED_COUNT_KEY: U64 = U64::new(0);
const CHECKPOINT_PARTITION_SUFFIX: &str = "__eventcount_checkpoint";
pub struct EventCountCheckpoint<E: commonware_storage::Context> {
store: Metadata<E, U64, u64>,
}
impl<E: commonware_storage::Context> EventCountCheckpoint<E> {
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 })
}
#[must_use]
pub fn expected_count(&self) -> u64 {
self.store.get(&EXPECTED_COUNT_KEY).copied().unwrap_or(0)
}
pub async fn record(mut self, count: u64) -> Result<Self, EventLogError> {
self.store.put(EXPECTED_COUNT_KEY, count);
self.store.sync().await?;
Ok(self)
}
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,
};
#[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);
});
}
#[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");
});
}
#[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);
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}"
);
*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"
);
});
}
}