refactor: type-guard recently-executed block provenance (leader epoch/round/digest)

crates/consensus/primary/src/consensus/tests/bullshark_tests.rs

@@ -442,7 +442,7 @@ async fn commit_one() {
     Consensus::spawn(config, &cb, bullshark, &task_manager);
     let cb_clone = cb.clone();
     let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-    cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+    cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
     tokio::spawn(async move {
         let mut rx_primary = cb_clone.committed_certificates().subscribe();
         while rx_primary.recv().await.is_some() {}
@@ -501,7 +501,7 @@ async fn dead_node() {
 
     let cb = ConsensusBus::new();
     let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-    cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+    cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
     let mut rx_output = cb.sequence().subscribe();
     let task_manager = TaskManager::default();
     Consensus::spawn(config, &cb, bullshark, &task_manager);
@@ -635,7 +635,7 @@ async fn not_enough_support() {
 
     let cb = ConsensusBus::new();
     let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-    cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+    cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
     let mut rx_output = cb.sequence().subscribe();
     let task_manager = TaskManager::default();
     Consensus::spawn(config, &cb, bullshark, &task_manager);
@@ -735,7 +735,7 @@ async fn missing_leader() {
 
     let cb = ConsensusBus::new();
     let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-    cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+    cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
     let mut rx_output = cb.sequence().subscribe();
     let task_manager = TaskManager::default();
     Consensus::spawn(config, &cb, bullshark, &task_manager);
@@ -803,7 +803,7 @@ async fn committed_round_after_restart() {
 
         let cb = ConsensusBus::new();
         let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-        cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+        cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
         let mut rx_primary = cb.committed_certificates().subscribe();
         let mut rx_output = cb.sequence().subscribe();
         let mut task_manager = TaskManager::default();
@@ -1042,7 +1042,7 @@ async fn restart_with_new_committee() {
 
         let cb = ConsensusBus::new();
         let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-        cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+        cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
         let mut rx_output = cb.sequence().subscribe();
         let mut task_manager = TaskManager::default();
         Consensus::spawn(config.clone(), &cb, bullshark, &task_manager);

crates/consensus/primary/src/consensus/tests/consensus_tests.rs

@@ -63,7 +63,7 @@ async fn test_consensus_recovery_with_bullshark() {
 
     let cb = ConsensusBus::new();
     let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-    cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+    cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
     let mut rx_output = cb.sequence().subscribe();
     let task_manager = TaskManager::default();
     Consensus::spawn(config.clone(), &cb, bullshark, &task_manager);
@@ -148,7 +148,7 @@ async fn test_consensus_recovery_with_bullshark() {
 
     let cb = ConsensusBus::new();
     let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-    cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+    cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
     let mut rx_output = cb.sequence().subscribe();
     let task_manager = TaskManager::default();
     Consensus::spawn(config.clone(), &cb, bullshark, &task_manager);
@@ -206,7 +206,7 @@ async fn test_consensus_recovery_with_bullshark() {
 
     let cb = ConsensusBus::new();
     let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-    cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+    cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
     let mut rx_output = cb.sequence().subscribe();
     let task_manager = TaskManager::default();
     Consensus::spawn(config, &cb, bullshark, &task_manager);
@@ -311,7 +311,7 @@ async fn single_validator_commits_a_subdag() {
 
     let cb = ConsensusBus::new();
     let dummy_parent = SealedHeader::new(ExecHeader::default(), B256::default());
-    cb.recent_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
+    cb.recently_executed_blocks().send_modify(|blocks| blocks.push_latest(dummy_parent));
     let mut rx_output = cb.sequence().subscribe();
     let task_manager = TaskManager::default();
     Consensus::spawn(config.clone(), &cb, bullshark, &task_manager);

crates/consensus/primary/src/consensus_bus.rs

@@ -4,7 +4,7 @@
 
 use crate::{
     certificate_fetcher::CertificateFetcherCommand, consensus::ConsensusRound,
-    proposer::OurDigestMessage, state_sync::CertificateManagerCommand, RecentBlocks,
+    proposer::OurDigestMessage, state_sync::CertificateManagerCommand, RecentlyExecutedBlocks,
 };
 use consensus_metrics::metered_channel::{self, channel_with_total_sender, MeteredMpscChannel};
 use parking_lot::Mutex;
@@ -195,12 +195,13 @@ struct ConsensusBusAppInner {
     /// Signals a new round
     tx_primary_round_updates: watch::Sender<Round>,
 
-    /// Watch tracking most recent blocks
-    tx_recent_blocks: watch::Sender<RecentBlocks>,
+    /// Watch tracking the most recently executed blocks
+    tx_recently_executed_blocks: watch::Sender<RecentlyExecutedBlocks>,
 
     /// The EVM-execution anchor: the consensus header the highest executed block commits to.
     /// Single source of truth for the restart/replay watermark, seeded once at boot from the
-    /// chain and advanced live by the engine, so consumers never re-derive it from recent_blocks.
+    /// chain and advanced live by the engine, so consumers never re-derive it from
+    /// recently_executed_blocks.
     tx_executed_anchor: watch::Sender<ConsensusHeader>,
 
     /// True when the node-scoped engine has executed everything it admitted (its queue is empty
@@ -238,7 +239,8 @@ struct ConsensusBusAppInner {
     executor_metrics: Arc<ExecutorMetrics>,
     /// Signal that the subscriber has finished replaying missed consensus outputs on startup.
     /// The proposer waits for this before creating headers, preventing stale exec_digest
-    /// from recent_blocks that was populated from MDBX before execution replay completed.
+    /// from recently_executed_blocks that was populated from MDBX before execution replay
+    /// completed.
     tx_execution_replay_complete: watch::Sender<bool>,
     /// Batch lifecycle tracker (app-lifetime, survives epoch transitions).
     batch_tracker: Arc<BatchTracker>,
@@ -254,7 +256,7 @@ struct ConsensusBusAppInner {
 }
 
 impl ConsensusBusAppInner {
-    fn new(initial_mode: NodeMode, recent_blocks: u32) -> Self {
+    fn new(initial_mode: NodeMode, recently_executed_blocks: u32) -> Self {
         let network_metrics = Arc::new(NetworkMetrics::default());
         let consensus_metrics = Arc::new(ConsensusMetrics::default());
         let primary_metrics = Arc::new(Metrics::default()); // Initialize the metrics
@@ -266,7 +268,8 @@ impl ConsensusBusAppInner {
         let (tx_primary_round_updates, _) = watch::channel(0u32);
         let (tx_last_consensus_header, _) = watch::channel(ConsensusHeader::default());
         let (tx_last_published_consensus_num_hash, _) = watch::channel((0, BlockHash::default()));
-        let (tx_recent_blocks, _) = watch::channel(RecentBlocks::new(recent_blocks as usize));
+        let (tx_recently_executed_blocks, _) =
+            watch::channel(RecentlyExecutedBlocks::new(recently_executed_blocks as usize));
         let (tx_executed_anchor, _) = watch::channel(ConsensusHeader::default());
         let (tx_engine_idle, _) = watch::channel(false);
         let (tx_execution_replay_complete, _) = watch::channel(false);
@@ -285,7 +288,7 @@ impl ConsensusBusAppInner {
             tx_gc_round_updates,
             tx_requested_missing_epoch,
             tx_primary_round_updates,
-            tx_recent_blocks,
+            tx_recently_executed_blocks,
             tx_executed_anchor,
             tx_engine_idle,
             tx_last_consensus_header,
@@ -319,12 +322,13 @@ impl ConsensusBusAppInner {
         // NOTE: promotion_barrier is intentionally NOT reset here. It is node-lifetime intent
         // that must survive the same-epoch mode-change restart the demotion itself triggers; it
         // self-invalidates via its epoch tag and is cleared lazily in try_rejoin_consensus.
-        let recent_blocks = self.tx_recent_blocks.borrow().block_capacity();
+        let recently_executed_blocks = self.tx_recently_executed_blocks.borrow().block_capacity();
         // Hang onto the last block of the previous epoch, clear the rest.
-        let latest = self.tx_recent_blocks.borrow().latest_block();
-        let mut recent_blocks = RecentBlocks::new(recent_blocks as usize);
-        recent_blocks.push_latest(latest);
-        self.tx_recent_blocks.send_replace(recent_blocks);
+        let latest = self.tx_recently_executed_blocks.borrow().latest_block();
+        let mut recently_executed_blocks =
+            RecentlyExecutedBlocks::new(recently_executed_blocks as usize);
+        recently_executed_blocks.push_latest(latest.into_header());
+        self.tx_recently_executed_blocks.send_replace(recently_executed_blocks);
     }
 }
 
@@ -478,8 +482,8 @@ impl ConsensusBus {
     }
 
     /// Create a bus with an explicit initial mode.
-    pub fn new_with_args(initial_mode: NodeMode, recent_blocks: u32) -> Self {
-        let inner_app = Arc::new(ConsensusBusAppInner::new(initial_mode, recent_blocks));
+    pub fn new_with_args(initial_mode: NodeMode, recently_executed_blocks: u32) -> Self {
+        let inner_app = Arc::new(ConsensusBusAppInner::new(initial_mode, recently_executed_blocks));
         let inner_epoch = Arc::new(ConsensusBusEpochInner::new(&inner_app));
         Self { inner_app, inner_epoch }
     }
@@ -619,17 +623,18 @@ impl ConsensusBus {
     ///
     /// For the frontier epoch/round read the monotonic [`Self::executed_anchor`] instead, or scan
     /// this window for the max-nonce block.
-    pub fn recent_blocks(&self) -> &watch::Sender<RecentBlocks> {
-        &self.inner_app.tx_recent_blocks
+    pub fn recently_executed_blocks(&self) -> &watch::Sender<RecentlyExecutedBlocks> {
+        &self.inner_app.tx_recently_executed_blocks
     }
 
     /// The EVM-execution anchor: the consensus header the highest executed block commits to.
     ///
     /// Single source of truth for the restart/replay watermark. Seeded once at boot from the
     /// chain (highest-nonce recent block) and advanced live by the engine on each executed output;
-    /// replay/catch-up consumers read this instead of re-deriving it from `recent_blocks` (whose
-    /// tip can lag behind after a drained parked batch). Distinct from the peer-derived
-    /// [`Self::last_consensus_header`], which is for header numbering only.
+    /// replay/catch-up consumers read this instead of re-deriving it from
+    /// `recently_executed_blocks` (whose tip can lag behind after a drained parked batch).
+    /// Distinct from the peer-derived [`Self::last_consensus_header`], which is for header
+    /// numbering only.
     pub fn executed_anchor(&self) -> &watch::Sender<ConsensusHeader> {
         &self.inner_app.tx_executed_anchor
     }
@@ -782,7 +787,8 @@ impl ConsensusBus {
     /// Resolves once the target block has executed, or returns a [`WaitForExecutionError`].
     ///
     /// The unbounded form returns `Forked` on a divergent hash at the target number or `Closed` if
-    /// `recent_blocks` closes; never `Stalled` (use [`Self::wait_for_execution_bounded`] for that).
+    /// `recently_executed_blocks` closes; never `Stalled` (use [`Self::wait_for_execution_bounded`]
+    /// for that).
     pub async fn wait_for_execution(
         &self,
         block: BlockNumHash,
@@ -807,34 +813,36 @@ impl ConsensusBus {
         block: BlockNumHash,
         idle_timeout: Option<Duration>,
     ) -> Result<(), WaitForExecutionError> {
-        let mut watch_execution_result = self.recent_blocks().subscribe();
+        let mut watch_execution_result = self.recently_executed_blocks().subscribe();
         let target_number = block.number;
         let target_hash = block.hash;
 
-        // Make sure that our recent blocks is not empty.  If it is we can have a race around block
-        // 0.
-        while self.recent_blocks().borrow().is_empty() {
+        // Make sure that our recently-executed window is not empty.  If it is we can have a race
+        // around block 0.
+        while self.recently_executed_blocks().borrow().is_empty() {
             await_execution_tick(&mut watch_execution_result, idle_timeout).await?;
         }
-        let mut current_number = self.recent_blocks().borrow().latest_block_num_hash().number;
+        let mut current_number =
+            self.recently_executed_blocks().borrow().latest_block_num_hash().number;
         while current_number < target_number {
             await_execution_tick(&mut watch_execution_result, idle_timeout).await?;
-            current_number = self.recent_blocks().borrow().latest_block_num_hash().number;
+            current_number =
+                self.recently_executed_blocks().borrow().latest_block_num_hash().number;
         }
 
-        let recent = self.recent_blocks().borrow();
+        let recent = self.recently_executed_blocks().borrow();
         let oldest_number = recent.oldest_block_number();
         let latest_number = recent.latest_block_num_hash().number;
 
-        // block older than recent_blocks window, assume already executed
+        // block older than recently_executed_blocks window, assume already executed
         if target_number < oldest_number {
             tracing::debug!(
                 target: "consensus-bus",
                 target_number,
                 oldest_number,
                 latest_number,
                 ?target_hash,
-                "wait_for_execution: block older than recent_blocks window, assuming already executed"
+                "wait_for_execution: block older than recently_executed_blocks window, assuming already executed"
             );
             return Ok(());
         }
@@ -854,7 +862,7 @@ impl ConsensusBus {
                 latest_number,
                 block_at_target_hash = ?block_at_target.map(|b| b.hash()),
                 "wait_for_execution: HASH MISMATCH - potential fork detected! \
-                 Expected hash not found in recent_blocks at target number."
+                 Expected hash not found in recently_executed_blocks at target number."
             );
             Err(WaitForExecutionError::Forked)
         }
@@ -865,12 +873,12 @@ impl ConsensusBus {
 /// a progressing chain resets it every block.
 pub const EXECUTION_STALL_TIMEOUT: Duration = Duration::from_secs(30);
 
-/// Await the next `recent_blocks` change, optionally bounded by `idle_timeout`.
+/// Await the next `recently_executed_blocks` change, optionally bounded by `idle_timeout`.
 ///
 /// Returns `Closed` if the receiver closes, or `Stalled` when `idle_timeout` is set and no change
 /// arrives within that window.
 async fn await_execution_tick(
-    rx: &mut watch::Receiver<RecentBlocks>,
+    rx: &mut watch::Receiver<RecentlyExecutedBlocks>,
     idle_timeout: Option<Duration>,
 ) -> Result<(), WaitForExecutionError> {
     match idle_timeout {
@@ -890,10 +898,10 @@ pub enum WaitForExecutionError {
     /// Execution made no progress within the bounded idle window (a stalled chain).
     #[error("execution made no progress within the idle timeout")]
     Stalled,
-    /// `recent_blocks` reported its sender closed. Defensive only: the sole sender is owned for
-    /// the app lifetime by the same `ConsensusBus` the wait borrows, so this is not expected
-    /// to fire.
-    #[error("recent_blocks closed before the target executed")]
+    /// `recently_executed_blocks` reported its sender closed. Defensive only: the sole sender is
+    /// owned for the app lifetime by the same `ConsensusBus` the wait borrows, so this is not
+    /// expected to fire.
+    #[error("recently_executed_blocks closed before the target executed")]
     Closed,
     /// Execution reached the target number but the block hash there differs: a fork.
     #[error("execution reached the target number with a divergent block hash")]

crates/consensus/primary/src/lib.rs

@@ -24,8 +24,8 @@ pub use crate::primary::Primary;
 mod consensus_bus;
 pub use consensus_bus::*;
 
-mod recent_blocks;
-pub use recent_blocks::*;
+mod recently_executed_blocks;
+pub use recently_executed_blocks::*;
 
 #[cfg(feature = "test-utils")]
 pub mod test_utils;

crates/consensus/primary/src/network/handler.rs

@@ -11,7 +11,7 @@ use crate::{
         state::{AuthVoteState, InFlightGuard, IN_FLIGHT_TTL},
     },
     state_sync::{CertificateCollector, StateSynchronizer},
-    ConsensusBus, NodeMode, RecentBlocks,
+    ConsensusBus, NodeMode, RecentlyExecutedBlocks,
 };
 use parking_lot::Mutex;
 use rayls_consensus_network::GossipMessage;
@@ -518,8 +518,10 @@ where
         // one kept check is a non-blocking breaker: refuse to vote if our own executed block at the
         // anchor already diverges, so a quorum cannot certify a forked anchor.
         let target = header.latest_execution_block;
-        let anchor_forked =
-            anchor_diverges_from_executed(&self.consensus_bus.recent_blocks().borrow(), target);
+        let anchor_forked = anchor_diverges_from_executed(
+            &self.consensus_bus.recently_executed_blocks().borrow(),
+            target,
+        );
         if anchor_forked {
             warn!(
                 target: "primary::handler",
@@ -949,7 +951,7 @@ fn collect_missing_certs_blocking<DB: Database>(
 ///
 /// Evidence-only: a target the node has not executed returns false, so a behind node still votes;
 /// only a node whose own executed state contradicts the anchor rejects.
-fn anchor_diverges_from_executed(recent: &RecentBlocks, target: BlockNumHash) -> bool {
+fn anchor_diverges_from_executed(recent: &RecentlyExecutedBlocks, target: BlockNumHash) -> bool {
     recent.block_at_number(target.number).is_some_and(|block| block.hash() != target.hash)
 }
 
@@ -994,7 +996,7 @@ fn evict_oldest_completed(cache: &mut AuthEquivocationMap, max_entries: usize) {
 #[cfg(test)]
 mod tests {
     use super::{anchor_diverges_from_executed, BehindTracker};
-    use crate::RecentBlocks;
+    use crate::RecentlyExecutedBlocks;
     use rayls_infrastructure_types::{BlockNumHash, ExecHeader, SealedHeader, B256};
     use std::time::Duration;
 
@@ -1010,7 +1012,7 @@ mod tests {
     /// votes on data availability and structure rather than waiting for execution to catch up.
     #[test]
     fn anchor_divergence_is_evidence_only() {
-        let mut recent = RecentBlocks::new(8);
+        let mut recent = RecentlyExecutedBlocks::new(8);
         recent.push_latest(sealed_at(5)); // executed block 5, hash repeat_byte(5)
 
         // matching anchor at an executed height: no divergence, vote.

crates/consensus/primary/src/proposer/header_builder.rs

@@ -77,7 +77,7 @@ impl<DB: Database> Proposer<DB> {
             current_epoch,
             payload_vec.into_iter().collect(),
             parents.iter().map(|x| x.digest()).collect(),
-            consensus_bus.recent_blocks().borrow().latest_block_num_hash(),
+            consensus_bus.recently_executed_blocks().borrow().latest_block_num_hash(),
         );
 
         // update metrics before sending/storing header