[marshal] Send Tip Updates (#2065)

patrick-ogrady · web-flow · commit 03c0661dd3ac · 2025-10-30T17:59:27.000-07:00
diff --git a/consensus/src/marshal/actor.rs b/consensus/src/marshal/actor.rs
@@ -10,7 +10,7 @@ use super::{
     SchemeProvider,
 };
 use crate::{
-    marshal::ingress::mailbox::Identifier as BlockID,
+    marshal::{ingress::mailbox::Identifier as BlockID, Update},
     simplex::{
         signing_scheme::Scheme,
         types::{Finalization, Notarization},
@@ -35,7 +35,6 @@ use governor::clock::Clock as GClock;
 use prometheus_client::metrics::gauge::Gauge;
 use rand::{CryptoRng, Rng};
 use std::{
-    cmp::max,
     collections::{btree_map::Entry, BTreeMap},
     time::Instant,
 };
@@ -95,6 +94,8 @@ pub struct Actor<
     // ---------- State ----------
     // Last view processed
     last_processed_round: Round,
+    // Highest known finalized height
+    tip: u64,
     // Outstanding subscriptions for blocks
     block_subscriptions: BTreeMap<B::Commitment, BlockSubscription<B>>,
 
@@ -238,6 +239,7 @@ impl<
                 block_codec_config: config.block_codec_config,
                 partition_prefix: config.partition_prefix,
                 last_processed_round: Round::new(0, 0),
+                tip: 0,
                 block_subscriptions: BTreeMap::new(),
                 cache,
                 finalizations_by_height,
@@ -252,7 +254,7 @@ impl<
     /// Start the actor.
     pub fn start<R, K>(
         mut self,
-        application: impl Reporter<Activity = B>,
+        application: impl Reporter<Activity = Update<B>>,
         buffer: buffered::Mailbox<K, B>,
         resolver: (mpsc::Receiver<handler::Message<B>>, R),
     ) -> Handle<()>
@@ -266,7 +268,7 @@ impl<
     /// Run the application actor.
     async fn run<R, K>(
         mut self,
-        application: impl Reporter<Activity = B>,
+        mut application: impl Reporter<Activity = Update<B>>,
         mut buffer: buffered::Mailbox<K, B>,
         (mut resolver_rx, mut resolver): (mpsc::Receiver<handler::Message<B>>, R),
     ) where
@@ -280,7 +282,7 @@ impl<
         let finalizer = Finalizer::new(
             self.context.with_label("finalizer"),
             format!("{}-finalizer", self.partition_prefix.clone()),
-            application,
+            application.clone(),
             orchestrator,
             notifier_rx,
         )
@@ -290,6 +292,14 @@ impl<
         // Create a local pool for waiter futures
         let mut waiters = AbortablePool::<(B::Commitment, B)>::default();
 
+        // Get tip and send to application
+        let tip = self.get_latest().await;
+        if let Some((height, commitment)) = tip {
+            application.report(Update::Tip(height, commitment)).await;
+            self.tip = height;
+            self.finalized_height.set(height as i64);
+        }
+
         // Handle messages
         loop {
             // Remove any dropped subscribers. If all subscribers dropped, abort the waiter.
@@ -369,7 +379,7 @@ impl<
                             if let Some(block) = self.find_block(&mut buffer, commitment).await {
                                 // If found, persist the block
                                 let height = block.height();
-                                self.finalize(height, commitment, block, Some(finalization), &mut notifier_tx).await;
+                                self.finalize(height, commitment, block, Some(finalization), &mut notifier_tx, &mut application).await;
                                 debug!(?round, height, "finalized block stored");
                             } else {
                                 // Otherwise, fetch the block from the network.
@@ -504,7 +514,7 @@ impl<
                                 let commitment = cursor.parent();
                                 if let Some(block) = self.find_block(&mut buffer, commitment).await {
                                     let finalization = self.cache.get_finalization_for(commitment).await;
-                                    self.finalize(block.height(), commitment, block.clone(), finalization, &mut notifier_tx).await;
+                                    self.finalize(block.height(), commitment, block.clone(), finalization, &mut notifier_tx, &mut application).await;
                                     debug!(height = block.height(), "repaired block");
                                     cursor = block;
                                 } else {
@@ -595,7 +605,7 @@ impl<
                                     // Persist the block, also persisting the finalization if we have it
                                     let height = block.height();
                                     let finalization = self.cache.get_finalization_for(commitment).await;
-                                    self.finalize(height, commitment, block, finalization, &mut notifier_tx).await;
+                                    self.finalize(height, commitment, block, finalization, &mut notifier_tx, &mut application).await;
                                     debug!(?commitment, height, "received block");
                                     let _ = response.send(true);
                                 },
@@ -629,7 +639,7 @@ impl<
                                     // Valid finalization received
                                     debug!(height, "received finalization");
                                     let _ = response.send(true);
-                                    self.finalize(height, block.commitment(), block, Some(finalization), &mut notifier_tx).await;
+                                    self.finalize(height, block.commitment(), block, Some(finalization), &mut notifier_tx, &mut application).await;
                                 },
                                 Request::Notarized { round } => {
                                     let Some(scheme) = self.scheme_provider.scheme(round.epoch()) else {
@@ -670,7 +680,7 @@ impl<
                                     // the request for the block.
                                     let height = block.height();
                                     if let Some(finalization) = self.cache.get_finalization_for(commitment).await {
-                                        self.finalize(height, commitment, block.clone(), Some(finalization), &mut notifier_tx).await;
+                                        self.finalize(height, commitment, block.clone(), Some(finalization), &mut notifier_tx, &mut application).await;
                                     }
 
                                     // Cache the notarization and block
@@ -746,6 +756,7 @@ impl<
         block: B,
         finalization: Option<Finalization<S, B::Commitment>>,
         notifier: &mut mpsc::Sender<()>,
+        application: &mut impl Reporter<Activity = Update<B>>,
     ) {
         self.notify_subscribers(commitment, &block).await;
 
@@ -766,13 +777,15 @@ impl<
             panic!("failed to finalize: {e}");
         }
 
-        // Update metrics
-        let new_value: i64 = height as i64;
-        let old_value: i64 = self.finalized_height.get();
-        self.finalized_height.set(max(new_value, old_value));
-
         // Notify the finalizer
         let _ = notifier.try_send(());
+
+        // Update metrics and send tip update to application
+        if height > self.tip {
+            application.report(Update::Tip(height, commitment)).await;
+            self.tip = height;
+            self.finalized_height.set(height as i64);
+        }
     }
 
     /// Get the latest finalized block information (height and commitment tuple).
diff --git a/consensus/src/marshal/finalizer.rs b/consensus/src/marshal/finalizer.rs
@@ -1,4 +1,7 @@
-use crate::{marshal::ingress::orchestrator::Orchestrator, Block, Reporter};
+use crate::{
+    marshal::{ingress::orchestrator::Orchestrator, Update},
+    Block, Reporter,
+};
 use commonware_runtime::{spawn_cell, Clock, ContextCell, Handle, Metrics, Spawner, Storage};
 use commonware_storage::metadata::{self, Metadata};
 use commonware_utils::{fixed_bytes, sequence::FixedBytes};
@@ -13,7 +16,11 @@ const LATEST_KEY: FixedBytes<1> = fixed_bytes!("00");
 ///
 /// Stores the highest height for which the application has processed. This allows resuming
 /// processing from the last processed height after a restart.
-pub struct Finalizer<B: Block, R: Spawner + Clock + Metrics + Storage, Z: Reporter<Activity = B>> {
+pub struct Finalizer<
+    B: Block,
+    R: Spawner + Clock + Metrics + Storage,
+    Z: Reporter<Activity = Update<B>>,
+> {
     context: ContextCell<R>,
 
     // Application that processes the finalized blocks.
@@ -29,7 +36,7 @@ pub struct Finalizer<B: Block, R: Spawner + Clock + Metrics + Storage, Z: Report
     metadata: Metadata<R, FixedBytes<1>, u64>,
 }
 
-impl<B: Block, R: Spawner + Clock + Metrics + Storage, Z: Reporter<Activity = B>>
+impl<B: Block, R: Spawner + Clock + Metrics + Storage, Z: Reporter<Activity = Update<B>>>
     Finalizer<B, R, Z>
 {
     /// Initialize the finalizer.
@@ -89,7 +96,7 @@ impl<B: Block, R: Spawner + Clock + Metrics + Storage, Z: Reporter<Activity = B>
                 // height is processed by the application), it is possible that the application may
                 // be asked to process a block it has already seen (which it can simply ignore).
                 let commitment = block.commitment();
-                self.application.report(block).await;
+                self.application.report(Update::Block(block)).await;
 
                 // Record that we have processed up through this height.
                 latest = height;
diff --git a/consensus/src/marshal/mocks/application.rs b/consensus/src/marshal/mocks/application.rs
@@ -1,4 +1,4 @@
-use crate::{Block, Reporter};
+use crate::{marshal::Update, Block, Reporter};
 use std::{
     collections::BTreeMap,
     sync::{Arc, Mutex},
@@ -8,12 +8,15 @@ use std::{
 #[derive(Clone)]
 pub struct Application<B: Block> {
     blocks: Arc<Mutex<BTreeMap<u64, B>>>,
+    #[allow(clippy::type_complexity)]
+    tip: Arc<Mutex<Option<(u64, B::Commitment)>>>,
 }
 
 impl<B: Block> Default for Application<B> {
     fn default() -> Self {
         Self {
             blocks: Default::default(),
+            tip: Default::default(),
         }
     }
 }
@@ -23,15 +26,24 @@ impl<B: Block> Application<B> {
     pub fn blocks(&self) -> BTreeMap<u64, B> {
         self.blocks.lock().unwrap().clone()
     }
+
+    /// Returns the tip.
+    pub fn tip(&self) -> Option<(u64, B::Commitment)> {
+        *self.tip.lock().unwrap()
+    }
 }
 
 impl<B: Block> Reporter for Application<B> {
-    type Activity = B;
+    type Activity = Update<B>;
 
     async fn report(&mut self, activity: Self::Activity) {
-        self.blocks
-            .lock()
-            .unwrap()
-            .insert(activity.height(), activity);
+        match activity {
+            Update::Block(block) => {
+                self.blocks.lock().unwrap().insert(block.height(), block);
+            }
+            Update::Tip(height, commitment) => {
+                *self.tip.lock().unwrap() = Some((height, commitment));
+            }
+        }
     }
 }
diff --git a/consensus/src/marshal/mod.rs b/consensus/src/marshal/mod.rs
@@ -64,7 +64,7 @@ pub mod ingress;
 pub use ingress::mailbox::Mailbox;
 pub mod resolver;
 
-use crate::{simplex::signing_scheme::Scheme, types::Epoch};
+use crate::{simplex::signing_scheme::Scheme, types::Epoch, Block};
 use std::sync::Arc;
 
 /// Supplies the signing scheme the marshal should use for a given epoch.
@@ -76,6 +76,18 @@ pub trait SchemeProvider: Clone + Send + Sync + 'static {
     fn scheme(&self, epoch: Epoch) -> Option<Arc<Self::Scheme>>;
 }
 
+/// An update reported to the application, either a new finalized tip or a finalized block.
+///
+/// Finalized tips are reported as soon as known, whether or not we hold all blocks up to that height.
+/// Finalized blocks are reported to the application in monotonically increasing order (no gaps permitted).
+#[derive(Clone, Debug)]
+pub enum Update<B: Block> {
+    /// A new finalized tip.
+    Tip(u64, B::Commitment),
+    /// A new finalized block.
+    Block(B),
+}
+
 #[cfg(test)]
 pub mod mocks;
 
@@ -417,6 +429,14 @@ mod tests {
                         finished = false;
                         break;
                     }
+                    let Some((height, _)) = app.tip() else {
+                        finished = false;
+                        break;
+                    };
+                    if height < NUM_BLOCKS {
+                        finished = false;
+                        break;
+                    }
                 }
             }
 
@@ -859,7 +879,7 @@ mod tests {
             } = bls12381_threshold::<V, _>(&mut context, NUM_VALIDATORS);
 
             let me = participants[0].clone();
-            let (_application, mut actor) = setup_validator(
+            let (application, mut actor) = setup_validator(
                 context.with_label("validator-0"),
                 &mut oracle,
                 me,
@@ -870,6 +890,7 @@ mod tests {
             // Before any finalization, GetBlock::Latest should be None
             let latest_block = actor.get_block(Identifier::Latest).await;
             assert!(latest_block.is_none());
+            assert!(application.tip().is_none());
 
             // Finalize a block at height 1
             let parent = Sha256::hash(b"");
@@ -889,6 +910,7 @@ mod tests {
             let by_height = actor.get_block(1).await.expect("missing block by height");
             assert_eq!(by_height.height(), 1);
             assert_eq!(by_height.digest(), commitment);
+            assert_eq!(application.tip(), Some((1, commitment)));
 
             // Get by latest
             let by_latest = actor
diff --git a/examples/reshare/src/dkg/ingress.rs b/examples/reshare/src/dkg/ingress.rs
@@ -3,7 +3,7 @@
 //! [Actor]: super::Actor
 
 use crate::{application::Block, dkg::DealOutcome};
-use commonware_consensus::Reporter;
+use commonware_consensus::{marshal::Update, Reporter};
 use commonware_cryptography::{bls12381::primitives::variant::Variant, Hasher, Signer};
 use futures::{
     channel::{mpsc, oneshot},
@@ -78,12 +78,16 @@ where
     C: Signer,
     V: Variant,
 {
-    type Activity = Block<H, C, V>;
+    type Activity = Update<Block<H, C, V>>;
 
-    async fn report(&mut self, block: Self::Activity) {
+    async fn report(&mut self, update: Self::Activity) {
         let (sender, receiver) = oneshot::channel();
 
         // Report the finalized block to the DKG actor on a best-effort basis.
+        let Update::Block(block) = update else {
+            // We ignore any other updates sent by marshal.
+            return;
+        };
         let _ = self
             .sender
             .send(Message::Finalized {
