feat: Add info logs for beginning of newPayload requests (paradigmxyz#16463)

Author: frankudoags

Cargo.lock

@@ -26,6 +26,7 @@ reth-trie-common.workspace = true
 alloy-primitives.workspace = true
 alloy-consensus.workspace = true
 alloy-rpc-types-engine.workspace = true
+alloy-eips.workspace = true
 
 # async
 tokio = { workspace = true, features = ["sync"] }
@@ -46,6 +47,7 @@ std = [
     "alloy-primitives/std",
     "alloy-consensus/std",
     "alloy-rpc-types-engine/std",
+    "alloy-eips/std",
     "futures/std",
     "serde/std",
     "thiserror/std",

crates/engine/primitives/Cargo.toml

@@ -3,6 +3,7 @@
 use crate::ForkchoiceStatus;
 use alloc::boxed::Box;
 use alloy_consensus::BlockHeader;
+use alloy_eips::BlockNumHash;
 use alloy_primitives::B256;
 use alloy_rpc_types_engine::ForkchoiceState;
 use core::{
@@ -20,6 +21,8 @@ pub enum BeaconConsensusEngineEvent<N: NodePrimitives = EthPrimitives> {
     ForkchoiceUpdated(ForkchoiceState, ForkchoiceStatus),
     /// A block was added to the fork chain.
     ForkBlockAdded(ExecutedBlockWithTrieUpdates<N>, Duration),
+    /// A new block was received from the consensus engine
+    BlockReceived(BlockNumHash),
     /// A block was added to the canonical chain, and the elapsed time validating the block
     CanonicalBlockAdded(ExecutedBlockWithTrieUpdates<N>, Duration),
     /// A canonical chain was committed, and the elapsed time committing the data
@@ -69,6 +72,9 @@ where
             Self::LiveSyncProgress(progress) => {
                 write!(f, "LiveSyncProgress({progress:?})")
             }
+            Self::BlockReceived(num_hash) => {
+                write!(f, "BlockReceived({num_hash:?})")
+            }
         }
     }
 }

crates/engine/primitives/src/event.rs

@@ -571,6 +571,10 @@ where
             }
         };
 
+        let num_hash = block.num_hash();
+        let engine_event = BeaconConsensusEngineEvent::BlockReceived(num_hash);
+        self.emit_event(EngineApiEvent::BeaconConsensus(engine_event));
+
         let block_hash = block.hash();
         let mut lowest_buffered_ancestor = self.lowest_buffered_ancestor_or(block_hash);
         if lowest_buffered_ancestor == block_hash {

crates/engine/tree/src/tree/mod.rs

@@ -349,6 +349,18 @@ impl TestHarness {
         }
     }
 
+    async fn check_block_received(&mut self, hash: B256) {
+        let event = self.from_tree_rx.recv().await.unwrap();
+        match event {
+            EngineApiEvent::BeaconConsensus(BeaconConsensusEngineEvent::BlockReceived(
+                num_hash,
+            )) => {
+                assert_eq!(num_hash.hash, hash);
+            }
+            _ => panic!("Unexpected event: {event:#?}"),
+        }
+    }
+
     fn persist_blocks(&self, blocks: Vec<RecoveredBlock<reth_ethereum_primitives::Block>>) {
         let mut block_data: Vec<(B256, Block)> = Vec::with_capacity(blocks.len());
         let mut headers_data: Vec<(B256, Header)> = Vec::with_capacity(blocks.len());
@@ -1137,6 +1149,9 @@ async fn test_engine_tree_buffered_blocks_are_eventually_connected() {
     assert!(test_harness.tree.state.buffer.block(&buffered_block_hash).is_none());
 
     // both blocks are added to the canon chain in order
+    // note that the buffered block is received first, but added last
+    test_harness.check_block_received(buffered_block_hash).await;
+    test_harness.check_block_received(non_buffered_block_hash).await;
     test_harness.check_canon_block_added(non_buffered_block_hash).await;
     test_harness.check_canon_block_added(buffered_block_hash).await;
 }

crates/engine/tree/src/tree/tests.rs

@@ -280,6 +280,9 @@ impl NodeState {
             BeaconConsensusEngineEvent::InvalidBlock(block) => {
                 warn!(number=block.number(), hash=?block.hash(), "Encountered invalid block");
             }
+            BeaconConsensusEngineEvent::BlockReceived(num_hash) => {
+                info!(number=num_hash.number, hash=?num_hash.hash, "Received block from consensus engine");
+            }
         }
     }
 

crates/node/events/src/node.rs

@@ -123,6 +123,7 @@ pub async fn maintain_pending_state<P>(
             }
             // ignore
             BeaconConsensusEngineEvent::CanonicalChainCommitted(_, _) |
+            BeaconConsensusEngineEvent::BlockReceived(_) |
             BeaconConsensusEngineEvent::LiveSyncProgress(_) => (),
         }
     }