move around

Author: patrick-ogrady

consensus/src/marshal/config.rs

@@ -17,12 +17,8 @@ use std::num::{NonZeroU64, NonZeroUsize};
 /// supersedes the configured anchor.
 pub enum Start<S: Scheme, C: Digest, B> {
     /// Start from the height-zero genesis block.
-    ///
-    /// Genesis is emitted to the application unless durable metadata shows it
-    /// was already acknowledged.
     Genesis(B),
-    /// Start from a finalized commitment, delivering the floor block to the
-    /// application if it has not already been acknowledged.
+    /// Start from a finalized commitment.
     Floor(Finalization<S, C>),
 }
 

consensus/src/marshal/core/actor.rs

@@ -4,6 +4,7 @@ use super::{
     delivery::PendingVerification,
     floor::Floor,
     mailbox::{CommitmentFallback, Mailbox, Message},
+    stream::Stream,
     subscriptions::{Key as SubscriptionKey, KeyFor as SubscriptionKeyFor, Subscriptions},
     variant::OptionalBuffer,
     Buffer, Variant,
@@ -39,23 +40,18 @@ use commonware_runtime::{
 };
 use commonware_storage::{
     archive::Identifier as ArchiveID,
-    metadata::{self, Metadata},
 };
 use commonware_utils::{
     acknowledgement::Exact,
     channel::{fallible::OneshotExt, oneshot},
     futures::AbortablePool,
-    sequence::U64,
     Acknowledgement, BoxedError,
 };
 use futures::{future::join_all, try_join};
 use rand_core::CryptoRngCore;
 use std::{collections::BTreeMap, future::Future, num::NonZeroUsize, sync::Arc};
 use tracing::{debug, warn};
 
-/// The key used to store the last processed height in the metadata store.
-const LATEST_KEY: U64 = U64::new(0xFF);
-
 // Resolver request keys are expressed in the variant commitment type, which
 // may differ from the block digest for coded variants.
 type ResolverRequestFor<V> = Key<<V as Variant>::Commitment>;
@@ -68,47 +64,6 @@ struct ResolverDelivery<V: Variant> {
     response: oneshot::Sender<bool>,
 }
 
-/// Last block acknowledged by the application.
-#[derive(Clone, Copy)]
-enum ApplicationFloor {
-    BeforeGenesis,
-    Acknowledged(Height),
-}
-
-impl ApplicationFloor {
-    const fn from_metadata(height: Option<Height>) -> Self {
-        match height {
-            Some(height) => Self::Acknowledged(height),
-            None => Self::BeforeGenesis,
-        }
-    }
-
-    const fn processed_height(self) -> Height {
-        match self {
-            Self::BeforeGenesis => Height::zero(),
-            Self::Acknowledged(height) => height,
-        }
-    }
-
-    const fn round_restore_height(self) -> Height {
-        match self {
-            Self::BeforeGenesis => Height::zero(),
-            Self::Acknowledged(height) => height.next(),
-        }
-    }
-
-    const fn next_dispatch_height(self) -> Height {
-        match self {
-            Self::BeforeGenesis => Height::zero(),
-            Self::Acknowledged(height) => height.next(),
-        }
-    }
-
-    const fn acknowledge(&mut self, height: Height) {
-        *self = Self::Acknowledged(height);
-    }
-}
-
 /// The [Actor] is responsible for receiving uncertified blocks from the broadcast mechanism,
 /// receiving notarizations and finalizations from consensus, and reconstructing a total order
 /// of blocks.
@@ -163,7 +118,7 @@ where
     // Current processed floor and any pending floor update
     floor: Floor<P::Scheme, V::Commitment>,
     // Application delivery cursor
-    application_floor: ApplicationFloor,
+    stream: Stream<E>,
     // Pending application acknowledgements
     pending_acks: PendingAcks<V, A>,
     // Highest known finalized height
@@ -174,8 +129,6 @@ where
     // ---------- Storage ----------
     // Prunable cache
     cache: cache::Manager<E, V, P::Scheme>,
-    // Metadata tracking application progress
-    application_metadata: Metadata<E, U64, Height>,
     // Finalizations stored by height
     finalizations_by_height: FC,
     // Finalized blocks stored by height
@@ -226,19 +179,14 @@ where
         )
         .await;
 
-        // Initialize metadata tracking application progress
-        let application_metadata = Metadata::init(
+        let stream = Stream::new(
             context.child("application_metadata"),
-            metadata::Config {
-                partition: format!("{}-application-metadata", config.partition_prefix),
-                codec_config: (),
-            },
+            &config.partition_prefix,
         )
-        .await
-        .expect("failed to initialize application metadata");
-        let application_floor =
-            ApplicationFloor::from_metadata(application_metadata.get(&LATEST_KEY).copied());
-        let last_processed_height = application_floor.processed_height();
+        .await;
+        let last_processed_height = stream
+            .processed_height()
+            .unwrap_or_else(Height::zero);
 
         // Genesis is a local anchor. A floor finalization is verified and
         // resolved after `run` receives the resolver and buffer.
@@ -260,7 +208,7 @@ where
         // one height ahead restores the anchor's round across the crash window.
         let last_processed_round = Self::latest_processed_round(
             &finalizations_by_height,
-            application_floor.round_restore_height(),
+            stream.next_height(),
         )
         .await;
 
@@ -289,12 +237,11 @@ where
                 strategy: config.strategy,
                 last_proposed_block: None,
                 floor,
-                application_floor,
+                stream,
                 pending_acks: PendingAcks::new(config.max_pending_acks.get()),
                 tip: Height::zero(),
                 block_subscriptions: Subscriptions::new(),
                 cache,
-                application_metadata,
                 finalizations_by_height,
                 finalized_blocks,
                 finalized_height,
@@ -532,7 +479,7 @@ where
         };
 
         // Persist buffered progress updates once after draining all ready acks.
-        self.application_metadata
+        self.stream
             .sync()
             .await
             .expect("failed to sync application progress");
@@ -1196,7 +1143,7 @@ where
         self.update_processed_height(dispatch_floor, resolver);
         self.update_processed_round_floor(dispatch_floor, round, resolver)
             .await;
-        self.application_metadata
+        self.stream
             .sync()
             .await
             .expect("failed to sync floor metadata");
@@ -1617,7 +1564,7 @@ where
     ///
     /// Iteration M (ack handler, M > N):
     ///   ack handler       ->  update_processed_height  ->  metadata buffered
-    ///   application_metadata.sync ->  metadata durable
+    ///   stream.sync       ->  metadata durable
     /// ```
     async fn try_dispatch_blocks(
         &mut self,
@@ -1631,7 +1578,7 @@ where
         while self.pending_acks.has_capacity() {
             let next_height = self
                 .pending_acks
-                .next_dispatch_height(self.application_floor.next_dispatch_height());
+                .next_dispatch_height(self.stream.next_height());
             let Some(block) = self.get_finalized_block(next_height).await else {
                 return;
             };
@@ -2087,8 +2034,7 @@ where
             PublicKey = <P::Scheme as CertificateScheme>::PublicKey,
         >,
     ) {
-        self.application_floor.acknowledge(height);
-        self.application_metadata.put(LATEST_KEY, height);
+        self.stream.acknowledge(height);
         self.floor.set_processed_height(height);
         let _ = self
             .processed_height

consensus/src/marshal/core/mod.rs

@@ -48,6 +48,7 @@ mod acks;
 pub(crate) mod cache;
 mod delivery;
 mod floor;
+mod stream;
 
 mod mailbox;
 pub use mailbox::{CommitmentFallback, DigestFallback, Mailbox};

consensus/src/marshal/core/stream.rs

@@ -0,0 +1,82 @@
+use crate::types::Height;
+use commonware_storage::{
+    metadata::{self, Metadata},
+    Context,
+};
+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);
+
+/// Last block acknowledged by the application.
+#[derive(Clone, Copy)]
+enum State {
+    Unprocessed,
+    Processed(Height),
+}
+
+impl State {
+    const fn new(processed_height: Option<Height>) -> Self {
+        match processed_height {
+            Some(height) => Self::Processed(height),
+            None => Self::Unprocessed,
+        }
+    }
+
+    const fn processed_height(self) -> Option<Height> {
+        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 and durable metadata.
+pub(super) struct Stream<E: Context> {
+    metadata: Metadata<E, U64, Height>,
+    state: State,
+}
+
+impl<E: Context> Stream<E> {
+    pub(super) async fn new(context: E, partition_prefix: &str) -> Self {
+        let metadata = Metadata::init(
+            context,
+            metadata::Config {
+                partition: format!("{partition_prefix}-application-metadata"),
+                codec_config: (),
+            },
+        )
+        .await
+        .expect("failed to initialize application metadata");
+        let state = State::new(metadata.get(&LATEST_KEY).copied());
+        Self { metadata, state }
+    }
+
+    pub(super) const fn processed_height(&self) -> Option<Height> {
+        self.state.processed_height()
+    }
+
+    pub(super) const fn next_height(&self) -> Height {
+        self.state.next_height()
+    }
+
+    pub(super) fn acknowledge(&mut self, height: Height) {
+        self.state.acknowledge(height);
+        self.metadata.put(LATEST_KEY, height);
+    }
+
+    pub(super) async fn sync(&self) -> Result<(), metadata::Error> {
+        self.metadata.sync().await
+    }
+}