use super::floor::Cutoff; use crate::types::{Epoch, Height, Round, View}; use commonware_codec::varint::UInt; use commonware_storage::{ Context, metadata::{self, Metadata}, }; use commonware_utils::sequence::U64; /// The key used to store the last processed height in the metadata store. const LATEST_KEY: U64 = U64::new(0xFF); /// The key used to store the epoch of the processed round floor. /// /// With [`CUTOFF_VIEW_KEY`] present, the round floor is [`Cutoff::Round`] at that epoch and view. /// When this key is alone, the round floor is the [`Cutoff::Epoch`] of that epoch. const CUTOFF_EPOCH_KEY: U64 = U64::new(0xFE); /// The key used to store the view of the processed round floor when it is a [`Cutoff::Round`]. const CUTOFF_VIEW_KEY: U64 = U64::new(0xFD); /// Last block acknowledged by the application. #[derive(Clone, Copy)] enum State { Unprocessed, Processed(Height), } impl State { const fn new(processed_height: Option) -> Self { match processed_height { Some(height) => Self::Processed(height), None => Self::Unprocessed, } } const fn processed_height(self) -> Option { match self { Self::Unprocessed => None, Self::Processed(height) => Some(height), } } const fn next_height(self) -> Height { match self { Self::Unprocessed => Height::zero(), Self::Processed(height) => height.next(), } } const fn acknowledge(&mut self, height: Height) { *self = Self::Processed(height); } } /// Application delivery stream progress, the processed round floor, and their durable metadata. pub(super) struct Stream { /// Holds the processed height and the round floor under fixed keys, so one sync persists them /// together. Every value is a varint `u64`, and a missing round floor reads as /// [`Cutoff::zero`]. metadata: Metadata>, /// Mirrors the processed height entry in `metadata`. state: State, /// Whether every buffered value has been synced. synced: bool, } impl Stream { /// Loads the processed height and round floor from the metadata store in /// `application_metadata_partition`. pub(super) async fn new(context: E, application_metadata_partition: &str) -> Self { let metadata: Metadata> = Metadata::init( context, metadata::Config { partition: application_metadata_partition.to_string(), codec_config: (), }, ) .await .expect("failed to initialize application metadata"); let state = State::new( metadata .get(&LATEST_KEY) .map(|height| Height::new(height.0)), ); Self { metadata, state, synced: true, } } /// Returns whether every buffered value has been synced. pub(super) const fn is_synced(&self) -> bool { self.synced } pub(super) const fn processed_height(&self) -> Option { self.state.processed_height() } pub(super) const fn next_height(&self) -> Height { self.state.next_height() } /// Returns the processed round floor, or [`Cutoff::zero`] if none has been set. A value /// buffered by [`Self::set_cutoff`] is returned before it is synced. pub(super) fn cutoff(&self) -> Cutoff { match ( self.metadata.get(&CUTOFF_EPOCH_KEY), self.metadata.get(&CUTOFF_VIEW_KEY), ) { (Some(epoch), Some(view)) => { Cutoff::Round(Round::new(Epoch::new(epoch.0), View::new(view.0))) } (Some(epoch), None) => Cutoff::Epoch(Epoch::new(epoch.0)), (None, _) => Cutoff::zero(), } } /// Buffers `height` as the last processed height. The next [`Self::sync`] makes it durable. pub(super) fn acknowledge(&mut self, height: Height) { self.state.acknowledge(height); self.metadata.put(LATEST_KEY, UInt(height.get())); self.synced = false; } /// Buffers `cutoff` as the processed round floor. The next [`Self::sync`] makes it durable. pub(super) fn set_cutoff(&mut self, cutoff: Cutoff) { match cutoff { Cutoff::Round(round) => { self.metadata .put(CUTOFF_EPOCH_KEY, UInt(round.epoch().get())); self.metadata.put(CUTOFF_VIEW_KEY, UInt(round.view().get())); } Cutoff::Epoch(epoch) => { self.metadata.put(CUTOFF_EPOCH_KEY, UInt(epoch.get())); self.metadata.remove(&CUTOFF_VIEW_KEY); } } self.synced = false; } /// Makes every buffered value durable and skips the metadata sync when nothing is buffered. pub(super) async fn sync(mut self) -> Result { if self.synced { return Ok(self); } self.metadata = self.metadata.sync().await?; self.synced = true; Ok(self) } } #[cfg(test)] mod tests { use super::*; use commonware_runtime::{Runner as _, Supervisor as _, deterministic}; const PARTITION: &str = "stream-metadata"; /// Writes a store whose only entry is `height`, encoded as a [`Height`] under the processed /// height key. async fn seed_processed_height(context: deterministic::Context, height: Height) { let mut metadata: Metadata = Metadata::init( context, metadata::Config { partition: PARTITION.to_string(), codec_config: (), }, ) .await .expect("metadata initialization must succeed"); metadata.put(LATEST_KEY, height); metadata.sync().await.expect("metadata sync must succeed"); } #[test] fn test_round_floor_round_trips_beside_processed_height() { let runner = deterministic::Runner::default(); runner.start(|context| async move { // A store holding only the processed height loads with the zero round floor. seed_processed_height(context.child("seed"), Height::new(7)).await; let mut stream = Stream::new(context.child("first"), PARTITION).await; assert_eq!(stream.processed_height(), Some(Height::new(7))); assert_eq!(stream.cutoff(), Cutoff::zero()); // A round floor set beside an acknowledgement survives a sync and reopen. let cutoff = Cutoff::Round(Round::new(Epoch::new(2), View::new(40))); stream.set_cutoff(cutoff); stream.acknowledge(Height::new(8)); let stream = stream.sync().await.expect("stream sync must succeed"); drop(stream); let stream = Stream::new(context.child("second"), PARTITION).await; assert_eq!(stream.processed_height(), Some(Height::new(8))); assert_eq!(stream.cutoff(), cutoff); }); } #[test] fn test_round_floor_kind_survives_reopen_across_transitions() { let runner = deterministic::Runner::default(); runner.start(|context| async move { // A round cutoff persists its epoch and view. let at_round = Cutoff::Round(Round::new(Epoch::new(2), View::new(40))); let mut stream = Stream::new(context.child("first"), PARTITION).await; stream.set_cutoff(at_round); let stream = stream.sync().await.expect("stream sync must succeed"); drop(stream); let mut stream = Stream::new(context.child("second"), PARTITION).await; assert_eq!(stream.cutoff(), at_round); // An epoch cutoff replaces it and reopens as an epoch cutoff, without the old view. let whole_epoch = Cutoff::Epoch(Epoch::new(2)); stream.set_cutoff(whole_epoch); assert_eq!(stream.cutoff(), whole_epoch); let stream = stream.sync().await.expect("stream sync must succeed"); drop(stream); let mut stream = Stream::new(context.child("third"), PARTITION).await; assert_eq!(stream.cutoff(), whole_epoch); // A round cutoff in a later epoch replaces the epoch cutoff and reopens with its view. let later = Cutoff::Round(Round::new(Epoch::new(3), View::new(5))); stream.set_cutoff(later); let stream = stream.sync().await.expect("stream sync must succeed"); drop(stream); let stream = Stream::new(context.child("fourth"), PARTITION).await; assert_eq!(stream.cutoff(), later); }); } }