nits

Author: patrick-ogrady

consensus/src/marshal/core/actor.rs

@@ -160,7 +160,7 @@ where
         finalizations_by_height: FC,
         mut finalized_blocks: FB,
         config: Config<P, ES, T, V::ApplicationBlock, V::Block, V::Commitment>,
-    ) -> (Self, Mailbox<P::Scheme, V>, Height) {
+    ) -> (Self, Mailbox<P::Scheme, V>, Option<Height>) {
         // Initialize cache
         let prunable_config = cache::Config {
             partition_prefix: format!("{}-cache", config.partition_prefix),
@@ -178,7 +178,8 @@ where
         .await;
 
         let stream = Stream::new(context.child("stream"), &config.partition_prefix).await;
-        let last_processed_height = stream.processed_height().unwrap_or_else(Height::zero);
+        let last_processed_height = stream.processed_height();
+        let processed_height_floor = last_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.
@@ -201,11 +202,13 @@ where
         // Create metrics
         let finalized_height = context.gauge("finalized_height", "Finalized height of application");
         let processed_height = context.gauge("processed_height", "Processed height of application");
-        let _ = processed_height.try_set(last_processed_height.get());
+        if let Some(last_processed_height) = last_processed_height {
+            let _ = processed_height.try_set(last_processed_height.get());
+        }
         let floor = pending_floor_anchor.map_or_else(
-            || Floor::resolved(last_processed_height, last_processed_round),
+            || Floor::resolved(processed_height_floor, last_processed_round),
             |finalization| {
-                Floor::awaiting_anchor(last_processed_height, last_processed_round, finalization)
+                Floor::awaiting_anchor(processed_height_floor, last_processed_round, finalization)
             },
         );
 
@@ -241,7 +244,7 @@ where
     async fn ensure_genesis_anchor(
         finalized_blocks: &mut FB,
         anchor: V::Block,
-        last_processed_height: Height,
+        last_processed_height: Option<Height>,
     ) {
         let anchor_height = anchor.height();
         let anchor_commitment = V::commitment(&anchor);
@@ -262,10 +265,12 @@ where
                 );
             }
             Ok(None) => {
-                if anchor_height < last_processed_height {
+                if let Some(existing) =
+                    last_processed_height.filter(|height| anchor_height < *height)
+                {
                     warn!(
                         height = %anchor_height,
-                        existing = %last_processed_height,
+                        %existing,
                         "ignoring stale anchor"
                     );
                     return;
@@ -675,7 +680,7 @@ where
                 response.send_lossy(finalization);
             }
             Message::GetProcessedHeight { response } => {
-                response.send_lossy(self.floor.processed_height());
+                response.send_lossy(self.stream.processed_height());
             }
             Message::HintFinalized { height, targets } => {
                 // Skip if finalization is already available locally.
@@ -2034,7 +2039,13 @@ where
     }
 
     /// Returns the latest known finalization round at or below the processed height.
-    async fn latest_processed_round(finalizations_by_height: &FC, height: Height) -> Round {
+    async fn latest_processed_round(
+        finalizations_by_height: &FC,
+        height: Option<Height>,
+    ) -> Round {
+        let Some(height) = height else {
+            return Round::zero();
+        };
         let Some(finalization_height) = finalizations_by_height
             .ranges_from(Height::zero())
             .filter_map(|(start, end)| (start <= height).then_some(end.min(height)))

consensus/src/marshal/core/mailbox.rs

@@ -51,7 +51,7 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
     /// A request to retrieve the latest processed height.
     GetProcessedHeight {
         /// A channel to send the latest processed height.
-        response: oneshot::Sender<Height>,
+        response: oneshot::Sender<Option<Height>>,
     },
     /// A hint that a finalized block may be available at a given height.
     ///
@@ -590,7 +590,7 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
         let _ = self
             .sender
             .enqueue(Message::GetProcessedHeight { response });
-        receiver.await.ok()
+        receiver.await.ok().flatten()
     }
 
     /// Hints that a finalized block may be available at the given height.

consensus/src/marshal/core/mod.rs

@@ -28,6 +28,7 @@
 //!     blocks_archive,
 //!     config,
 //! ).await;
+//! // `last_height` is `None` until the application acknowledges a block.
 //!
 //! // Start with application and buffer
 //! actor.start(application, Some(buffer), resolver);

consensus/src/marshal/mocks/harness.rs

@@ -168,7 +168,7 @@ pub struct ValidatorSetup<H: TestHarness> {
     pub application: Application<H::ApplicationBlock>,
     pub mailbox: Mailbox<S, H::Variant>,
     pub extra: H::ValidatorExtra,
-    pub height: Height,
+    pub height: Option<Height>,
     pub actor_handle: commonware_runtime::Handle<()>,
 }
 
@@ -839,7 +839,7 @@ pub fn hailstorm<H: TestHarness>(
                 .await;
                 assert_eq!(
                     restarted.height,
-                    Height::new(persisted_height),
+                    Some(Height::new(persisted_height)),
                     "validator {idx} should recover its persisted finalized height before replay"
                 );
 
@@ -3170,7 +3170,7 @@ pub fn ack_pipeline_backlog_persists_on_restart<H: TestHarness>() {
             Application::manual_ack(),
         )
         .await;
-        assert_eq!(restart.height, Height::new(3));
+        assert_eq!(restart.height, Some(Height::new(3)));
     });
 }
 
@@ -3200,8 +3200,14 @@ pub fn genesis_emitted_once<H: TestHarness>() {
             Application::<H::ApplicationBlock>::manual_ack(),
         )
         .await;
-        assert_eq!(setup.height, Height::zero());
+        assert_eq!(setup.height, None);
+        assert_eq!(setup.mailbox.get_processed_height().await, None);
         assert_eq!(setup.application.acknowledged().await, Height::zero());
+        context.sleep(Duration::from_millis(10)).await;
+        assert_eq!(
+            setup.mailbox.get_processed_height().await,
+            Some(Height::zero())
+        );
         assert!(setup.application.blocks().contains_key(&Height::zero()));
 
         // Let marshal persist the height-zero acknowledgement before restart.
@@ -3222,7 +3228,7 @@ pub fn genesis_emitted_once<H: TestHarness>() {
             Application::<H::ApplicationBlock>::manual_ack(),
         )
         .await;
-        assert_eq!(restart.height, Height::zero());
+        assert_eq!(restart.height, Some(Height::zero()));
         context.sleep(Duration::from_millis(100)).await;
         assert!(
             restart.application.blocks().is_empty(),
@@ -5083,8 +5089,8 @@ pub fn init_processed_height<H: TestHarness>() {
         };
         let initial_height = setup.height;
 
-        // Initially should be zero (no blocks processed)
-        assert_eq!(initial_height, Height::zero());
+        // Initially no block has been durably acknowledged.
+        assert_eq!(initial_height, None);
 
         // Finalize blocks 1-5
         let mut parent = Sha256::hash(b"");
@@ -5137,7 +5143,7 @@ pub fn init_processed_height<H: TestHarness>() {
         let recovered_height = setup2.height;
 
         // Should have recovered to height 5
-        assert_eq!(recovered_height, Height::new(5));
+        assert_eq!(recovered_height, Some(Height::new(5)));
     })
 }