chore: rm beacon consensus dep from engine-tree (#13720)

crates/engine/tree/Cargo.toml

@@ -12,7 +12,6 @@ workspace = true
 
 [dependencies]
 # reth
-reth-beacon-consensus.workspace = true
 reth-blockchain-tree-api.workspace = true
 reth-chain-state.workspace = true
 reth-chainspec = { workspace = true, optional = true }
@@ -57,6 +56,7 @@ metrics.workspace = true
 reth-metrics = { workspace = true, features = ["common"] }
 
 # misc
+schnellru.workspace = true
 rayon.workspace = true
 tracing.workspace = true
 derive_more.workspace = true

crates/engine/tree/src/download.rs

@@ -323,8 +323,8 @@ mod tests {
     use alloy_consensus::Header;
     use alloy_eips::eip1559::ETHEREUM_BLOCK_GAS_LIMIT;
     use assert_matches::assert_matches;
-    use reth_beacon_consensus::EthBeaconConsensus;
     use reth_chainspec::{ChainSpecBuilder, MAINNET};
+    use reth_ethereum_consensus::EthBeaconConsensus;
     use reth_network_p2p::test_utils::TestFullBlockClient;
     use reth_primitives::SealedHeader;
     use std::{future::poll_fn, sync::Arc};

crates/engine/tree/src/engine.rs

@@ -7,9 +7,8 @@ use crate::{
 };
 use alloy_primitives::B256;
 use futures::{Stream, StreamExt};
-use reth_beacon_consensus::BeaconConsensusEngineEvent;
 use reth_chain_state::ExecutedBlock;
-use reth_engine_primitives::{BeaconEngineMessage, EngineTypes};
+use reth_engine_primitives::{BeaconConsensusEngineEvent, BeaconEngineMessage, EngineTypes};
 use reth_primitives::{NodePrimitives, SealedBlockWithSenders};
 use reth_primitives_traits::Block;
 use std::{

crates/engine/tree/src/tree/config.rs

@@ -1,5 +1,14 @@
 //! Engine tree configuration.
 
+use alloy_eips::merge::EPOCH_SLOTS;
+
+/// The largest gap for which the tree will be used for sync. See docs for `pipeline_run_threshold`
+/// for more information.
+///
+/// This is the default threshold, the distance to the head that the tree will be used for sync.
+/// If the distance exceeds this threshold, the pipeline will be used for sync.
+pub(crate) const MIN_BLOCKS_FOR_PIPELINE_RUN: u64 = EPOCH_SLOTS;
+
 /// Triggers persistence when the number of canonical blocks in memory exceeds this threshold.
 pub const DEFAULT_PERSISTENCE_THRESHOLD: u64 = 2;
 

crates/engine/tree/src/tree/invalid_headers.rs

@@ -0,0 +1,125 @@
+use alloy_eips::eip1898::BlockWithParent;
+use alloy_primitives::B256;
+use reth_metrics::{
+    metrics::{Counter, Gauge},
+    Metrics,
+};
+use schnellru::{ByLength, LruMap};
+use std::fmt::Debug;
+use tracing::warn;
+
+/// The max hit counter for invalid headers in the cache before it is forcefully evicted.
+///
+/// In other words, if a header is referenced more than this number of times, it will be evicted to
+/// allow for reprocessing.
+const INVALID_HEADER_HIT_EVICTION_THRESHOLD: u8 = 128;
+
+/// Keeps track of invalid headers.
+#[derive(Debug)]
+pub(super) struct InvalidHeaderCache {
+    /// This maps a header hash to a reference to its invalid ancestor.
+    headers: LruMap<B256, HeaderEntry>,
+    /// Metrics for the cache.
+    metrics: InvalidHeaderCacheMetrics,
+}
+
+impl InvalidHeaderCache {
+    /// Invalid header cache constructor.
+    pub(super) fn new(max_length: u32) -> Self {
+        Self { headers: LruMap::new(ByLength::new(max_length)), metrics: Default::default() }
+    }
+
+    fn insert_entry(&mut self, hash: B256, header: BlockWithParent) {
+        self.headers.insert(hash, HeaderEntry { header, hit_count: 0 });
+    }
+
+    /// Returns the invalid ancestor's header if it exists in the cache.
+    ///
+    /// If this is called, the hit count for the entry is incremented.
+    /// If the hit count exceeds the threshold, the entry is evicted and `None` is returned.
+    pub(super) fn get(&mut self, hash: &B256) -> Option<BlockWithParent> {
+        {
+            let entry = self.headers.get(hash)?;
+            entry.hit_count += 1;
+            if entry.hit_count < INVALID_HEADER_HIT_EVICTION_THRESHOLD {
+                return Some(entry.header)
+            }
+        }
+        // if we get here, the entry has been hit too many times, so we evict it
+        self.headers.remove(hash);
+        self.metrics.hit_evictions.increment(1);
+        None
+    }
+
+    /// Inserts an invalid block into the cache, with a given invalid ancestor.
+    pub(super) fn insert_with_invalid_ancestor(
+        &mut self,
+        header_hash: B256,
+        invalid_ancestor: BlockWithParent,
+    ) {
+        if self.get(&header_hash).is_none() {
+            warn!(target: "consensus::engine", hash=?header_hash, ?invalid_ancestor, "Bad block with existing invalid ancestor");
+            self.insert_entry(header_hash, invalid_ancestor);
+
+            // update metrics
+            self.metrics.known_ancestor_inserts.increment(1);
+            self.metrics.count.set(self.headers.len() as f64);
+        }
+    }
+
+    /// Inserts an invalid ancestor into the map.
+    pub(super) fn insert(&mut self, invalid_ancestor: BlockWithParent) {
+        if self.get(&invalid_ancestor.block.hash).is_none() {
+            warn!(target: "consensus::engine", ?invalid_ancestor, "Bad block with hash");
+            self.insert_entry(invalid_ancestor.block.hash, invalid_ancestor);
+
+            // update metrics
+            self.metrics.unique_inserts.increment(1);
+            self.metrics.count.set(self.headers.len() as f64);
+        }
+    }
+}
+
+struct HeaderEntry {
+    /// Keeps track how many times this header has been hit.
+    hit_count: u8,
+    /// The actual header entry
+    header: BlockWithParent,
+}
+
+/// Metrics for the invalid headers cache.
+#[derive(Metrics)]
+#[metrics(scope = "consensus.engine.beacon.invalid_headers")]
+struct InvalidHeaderCacheMetrics {
+    /// The total number of invalid headers in the cache.
+    count: Gauge,
+    /// The number of inserts with a known ancestor.
+    known_ancestor_inserts: Counter,
+    /// The number of unique invalid header inserts (i.e. without a known ancestor).
+    unique_inserts: Counter,
+    /// The number of times a header was evicted from the cache because it was hit too many times.
+    hit_evictions: Counter,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy_consensus::Header;
+    use reth_primitives::SealedHeader;
+
+    #[test]
+    fn test_hit_eviction() {
+        let mut cache = InvalidHeaderCache::new(10);
+        let header = Header::default();
+        let header = SealedHeader::seal(header);
+        cache.insert(header.block_with_parent());
+        assert_eq!(cache.headers.get(&header.hash()).unwrap().hit_count, 0);
+
+        for hit in 1..INVALID_HEADER_HIT_EVICTION_THRESHOLD {
+            assert!(cache.get(&header.hash()).is_some());
+            assert_eq!(cache.headers.get(&header.hash()).unwrap().hit_count, hit);
+        }
+
+        assert!(cache.get(&header.hash()).is_none());
+    }
+}

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

@@ -16,9 +16,6 @@ use alloy_rpc_types_engine::{
     PayloadValidationError,
 };
 use block_buffer::BlockBuffer;
-use reth_beacon_consensus::{
-    BeaconConsensusEngineEvent, InvalidHeaderCache, MIN_BLOCKS_FOR_PIPELINE_RUN,
-};
 use reth_blockchain_tree_api::{
     error::{InsertBlockErrorKindTwo, InsertBlockErrorTwo, InsertBlockFatalError},
     BlockStatus2, InsertPayloadOk2,
@@ -29,8 +26,9 @@ use reth_chain_state::{
 use reth_consensus::{Consensus, FullConsensus, PostExecutionInput};
 pub use reth_engine_primitives::InvalidBlockHook;
 use reth_engine_primitives::{
-    BeaconEngineMessage, BeaconOnNewPayloadError, EngineApiMessageVersion, EngineTypes,
-    EngineValidator, ForkchoiceStateTracker, OnForkChoiceUpdated,
+    BeaconConsensusEngineEvent, BeaconEngineMessage, BeaconOnNewPayloadError,
+    EngineApiMessageVersion, EngineTypes, EngineValidator, ForkchoiceStateTracker,
+    OnForkChoiceUpdated,
 };
 use reth_errors::{ConsensusError, ProviderResult};
 use reth_evm::{execute::BlockExecutorProvider, system_calls::OnStateHook};
@@ -81,11 +79,13 @@ use tracing::*;
 mod block_buffer;
 pub mod config;
 mod invalid_block_hook;
+mod invalid_headers;
 mod metrics;
 mod persistence_state;
 pub mod root;
 mod trie_updates;
 
+use crate::tree::{config::MIN_BLOCKS_FOR_PIPELINE_RUN, invalid_headers::InvalidHeaderCache};
 pub use config::TreeConfig;
 pub use invalid_block_hook::{InvalidBlockHooks, NoopInvalidBlockHook};
 pub use persistence_state::PersistenceState;