nits

patrick-ogrady · patrick-ogrady · commit cb882d227201 · 2026-05-22T00:01:17.000-07:00
diff --git a/consensus/src/marshal/coding/mod.rs b/consensus/src/marshal/coding/mod.rs
@@ -779,6 +779,11 @@ mod tests {
         harness::ack_pipeline_backlog_persists_on_restart::<CodingHarness>();
     }
 
+    #[test_traced("WARN")]
+    fn test_coding_genesis_emitted_once() {
+        harness::genesis_emitted_once::<CodingHarness>();
+    }
+
     #[test_traced("WARN")]
     fn test_coding_proposed_success_implies_recoverable_after_restart() {
         harness::proposed_success_implies_recoverable_after_restart::<CodingHarness>(0..16);
diff --git a/consensus/src/marshal/config.rs b/consensus/src/marshal/config.rs
@@ -17,6 +17,9 @@ 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.
diff --git a/consensus/src/marshal/core/acks.rs b/consensus/src/marshal/core/acks.rs
@@ -62,12 +62,12 @@ impl<V: Variant, A: Acknowledgement> PendingAcks<V, A> {
     }
 
     /// Returns the next height to dispatch while preserving sequential order.
-    pub(super) fn next_dispatch_height(&self, last_processed_height: Height) -> Height {
+    pub(super) fn next_dispatch_height(&self, start_height: Height) -> Height {
         self.queue
             .back()
             .map(|ack| ack.height.next())
             .or_else(|| self.current.as_ref().map(|ack| ack.height.next()))
-            .unwrap_or_else(|| last_processed_height.next())
+            .unwrap_or(start_height)
     }
 
     /// Enqueues a newly dispatched ack, arming it immediately when idle.
@@ -136,18 +136,18 @@ mod tests {
     fn enqueue_tracks_capacity_and_fifo_ready_order() {
         let mut pending = PendingAcks::<TestVariant, Exact>::new(2);
         assert!(pending.has_capacity());
-        assert_eq!(pending.next_dispatch_height(Height::new(7)), Height::new(8));
+        assert_eq!(pending.next_dispatch_height(Height::new(8)), Height::new(8));
 
         let (first, first_ack) = pending_ack(8, 1);
         pending.enqueue(first);
         assert!(pending.has_capacity());
-        assert_eq!(pending.next_dispatch_height(Height::new(7)), Height::new(9));
+        assert_eq!(pending.next_dispatch_height(Height::new(8)), Height::new(9));
 
         let (second, second_ack) = pending_ack(9, 2);
         pending.enqueue(second);
         assert!(!pending.has_capacity());
         assert_eq!(
-            pending.next_dispatch_height(Height::new(7)),
+            pending.next_dispatch_height(Height::new(8)),
             Height::new(10)
         );
 
@@ -185,7 +185,7 @@ mod tests {
         assert!(pending.pop_ready().is_none());
         assert!(pending.has_capacity());
         assert_eq!(
-            pending.next_dispatch_height(Height::new(9)),
+            pending.next_dispatch_height(Height::new(10)),
             Height::new(10)
         );
     }
diff --git a/consensus/src/marshal/core/actor.rs b/consensus/src/marshal/core/actor.rs
@@ -68,6 +68,40 @@ 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,
+        }
+    }
+
+    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.
@@ -121,6 +155,8 @@ where
     last_proposed_block: Option<(Round, V::Commitment, V::Block)>,
     // Current processed floor and any pending floor update
     floor: Floor<P::Scheme, V::Commitment>,
+    // Application delivery cursor
+    application_floor: ApplicationFloor,
     // Pending application acknowledgements
     pending_acks: PendingAcks<V, A>,
     // Highest known finalized height
@@ -193,10 +229,9 @@ where
         )
         .await
         .expect("failed to initialize application metadata");
-        let last_processed_height = application_metadata
-            .get(&LATEST_KEY)
-            .copied()
-            .unwrap_or(Height::zero());
+        let application_floor =
+            ApplicationFloor::from_metadata(application_metadata.get(&LATEST_KEY).copied());
+        let last_processed_height = application_floor.processed_height();
 
         // Genesis is a local anchor. A floor finalization is verified and
         // resolved after `run` receives the resolver and buffer.
@@ -245,6 +280,7 @@ where
                 strategy: config.strategy,
                 last_proposed_block: None,
                 floor,
+                application_floor,
                 pending_acks: PendingAcks::new(config.max_pending_acks.get()),
                 tip: Height::zero(),
                 block_subscriptions: Subscriptions::new(),
@@ -1586,7 +1622,7 @@ where
         while self.pending_acks.has_capacity() {
             let next_height = self
                 .pending_acks
-                .next_dispatch_height(self.floor.processed_height());
+                .next_dispatch_height(self.application_floor.next_dispatch_height());
             let Some(block) = self.get_finalized_block(next_height).await else {
                 return;
             };
@@ -2042,6 +2078,7 @@ where
             PublicKey = <P::Scheme as CertificateScheme>::PublicKey,
         >,
     ) {
+        self.application_floor.acknowledge(height);
         self.application_metadata.put(LATEST_KEY, height);
         self.floor.set_processed_height(height);
         let _ = self
diff --git a/consensus/src/marshal/mocks/harness.rs b/consensus/src/marshal/mocks/harness.rs
@@ -1323,7 +1323,7 @@ pub fn certify_at_later_view_survives_earlier_view_pruning<H: TestHarness>() {
             parent = digest;
             parent_commitment = commitment;
         }
-        while (application.blocks().len() as u64) < CHAIN_LEN {
+        while application.tip().map(|(height, _)| height) != Some(Height::new(CHAIN_LEN)) {
             context.sleep(Duration::from_millis(10)).await;
         }
         context.sleep(Duration::from_millis(100)).await;
@@ -1491,13 +1491,18 @@ pub fn delivery_visibility_implies_recoverable_after_restart<H: TestHarness>(
                 H::verify(&mut handle, round, &block, &mut peers).await;
                 H::report_finalization(&mut mailbox, finalization.clone()).await;
 
-                let height = application.acknowledged().await;
-                assert_eq!(
-                    height,
-                    Height::new(1),
-                    "expected the first delivered finalized block to become visible at height 1 \
-                     before restart (seed={seed})"
-                );
+                loop {
+                    let height = application.acknowledged().await;
+                    if height == Height::new(1) {
+                        break;
+                    }
+                    assert_eq!(
+                        height,
+                        Height::zero(),
+                        "only genesis may precede the first finalized block before restart \
+                         (seed={seed})"
+                    );
+                }
             }
         });
 
@@ -2925,7 +2930,7 @@ pub fn finalize<H: TestHarness>(seed: u64, link: Link, quorum_sees_finalization:
             }
             finished = true;
             for app in applications.values() {
-                if app.blocks().len() != NUM_BLOCKS as usize {
+                if app.blocks().len() != (NUM_BLOCKS + 1) as usize {
                     finished = false;
                     break;
                 }
@@ -2981,6 +2986,7 @@ pub fn ack_pipeline_backlog<H: TestHarness>() {
             extra: setup.extra,
         }];
         let mut handle = handles[0].clone();
+        assert_eq!(application.acknowledged().await, Height::zero());
 
         let epocher = FixedEpocher::new(BLOCKS_PER_EPOCH);
         let mut parent = Sha256::hash(b"");
@@ -3077,6 +3083,7 @@ pub fn ack_pipeline_backlog_persists_on_restart<H: TestHarness>() {
             extra: setup.extra,
         }];
         let mut handle = handles[0].clone();
+        assert_eq!(application.acknowledged().await, Height::zero());
 
         let epocher = FixedEpocher::new(BLOCKS_PER_EPOCH);
         let mut parent = Sha256::hash(b"");
@@ -3145,6 +3152,67 @@ pub fn ack_pipeline_backlog_persists_on_restart<H: TestHarness>() {
     });
 }
 
+/// Test that genesis is delivered once and then suppressed after its ack is durable.
+pub fn genesis_emitted_once<H: TestHarness>() {
+    let runner = deterministic::Runner::timed(Duration::from_secs(60));
+    runner.start(|mut context| async move {
+        let Fixture {
+            participants,
+            schemes,
+            ..
+        } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
+        let mut oracle = setup_network_with_participants(
+            context.child("network"),
+            NZUsize!(1),
+            participants.clone(),
+        )
+        .await;
+
+        let validator = participants[0].clone();
+        let setup = H::setup_validator_with(
+            context.child("validator").with_attribute("index", 0),
+            &mut oracle,
+            validator.clone(),
+            ConstantProvider::new(schemes[0].clone()),
+            NZUsize!(1),
+            Application::<H::ApplicationBlock>::manual_ack(),
+        )
+        .await;
+        assert_eq!(setup.height, Height::zero());
+        assert_eq!(setup.application.acknowledged().await, Height::zero());
+        assert!(setup.application.blocks().contains_key(&Height::zero()));
+
+        // Let marshal persist the height-zero acknowledgement before restart.
+        context.sleep(Duration::from_millis(10)).await;
+        setup.actor_handle.abort();
+        drop(setup.mailbox);
+        drop(setup.extra);
+        context.sleep(Duration::from_millis(10)).await;
+
+        let restart = H::setup_validator_with(
+            context
+                .child("validator_restart")
+                .with_attribute("index", 0),
+            &mut oracle,
+            validator,
+            ConstantProvider::new(schemes[0].clone()),
+            NZUsize!(1),
+            Application::<H::ApplicationBlock>::manual_ack(),
+        )
+        .await;
+        assert_eq!(restart.height, Height::zero());
+        context.sleep(Duration::from_millis(100)).await;
+        assert!(
+            restart.application.blocks().is_empty(),
+            "genesis should not be emitted again after height zero is acknowledged"
+        );
+        assert!(
+            restart.application.pending_ack_heights().is_empty(),
+            "restart should not leave a duplicate genesis ack pending"
+        );
+    });
+}
+
 /// Test sync height floor.
 pub fn sync_height_floor<H: TestHarness>() {
     let runner = deterministic::Runner::new(
@@ -3240,7 +3308,7 @@ pub fn sync_height_floor<H: TestHarness>() {
             context.sleep(Duration::from_secs(1)).await;
             finished = true;
             for app in applications.values().skip(1) {
-                if app.blocks().len() != NUM_BLOCKS as usize {
+                if app.blocks().len() != (NUM_BLOCKS + 1) as usize {
                     finished = false;
                     break;
                 }
@@ -3289,7 +3357,7 @@ pub fn sync_height_floor<H: TestHarness>() {
         while !finished {
             context.sleep(Duration::from_secs(1)).await;
             finished = true;
-            if app.blocks().len() != (NUM_BLOCKS - NEW_SYNC_FLOOR + 1) as usize {
+            if app.blocks().len() != (NUM_BLOCKS - NEW_SYNC_FLOOR + 2) as usize {
                 finished = false;
                 continue;
             }
@@ -3393,7 +3461,7 @@ pub fn prune_finalized_archives<H: TestHarness>() {
             H::report_finalization(&mut mailbox, finalization).await;
         }
 
-        while application.blocks().len() < 20 {
+        while application.tip().map(|(height, _)| height) != Some(Height::new(20)) {
             context.sleep(Duration::from_millis(10)).await;
         }
 
@@ -4712,7 +4780,7 @@ pub fn hint_finalized_triggers_fetch<H: TestHarness>() {
         }
 
         // Wait for validator 0 to process all blocks
-        while app0.blocks().len() < 5 {
+        while app0.tip().map(|(height, _)| height) != Some(Height::new(5)) {
             context.sleep(Duration::from_millis(10)).await;
         }
 
@@ -5027,7 +5095,7 @@ pub fn init_processed_height<H: TestHarness>() {
         }
 
         // Wait for application to process all blocks
-        while app.blocks().len() < 5 {
+        while app.tip().map(|(height, _)| height) != Some(Height::new(5)) {
             context.sleep(Duration::from_millis(10)).await;
         }
 
diff --git a/consensus/src/marshal/standard/mod.rs b/consensus/src/marshal/standard/mod.rs
