feat: add missing message handling (#200)

* feat: add missing message handling

* refactor: new block message handling

* feat: add events and commands for transaction handling

* more work in transactions

* chore: silence warnings

crates/interfaces/src/p2p/error.rs

@@ -11,6 +11,8 @@ pub enum RequestError {
     ChannelClosed,
     #[error("Not connected to the peer.")]
     NotConnected,
+    #[error("Connection to a peer dropped while handling the request.")]
+    ConnectionDropped,
     #[error("Capability Message is not supported by remote peer.")]
     UnsupportedCapability,
     #[error("Request timed out while awaiting response.")]

crates/net/network/Cargo.toml

@@ -35,6 +35,7 @@ parking_lot = "0.12"
 async-trait = "0.1"
 bytes = "1.2"
 either = "1.8"
+linked_hash_set = "0.1"
 
 secp256k1 = { version = "0.24", features = [
     "global-context",

crates/net/network/src/cache.rs

@@ -0,0 +1,58 @@
+use linked_hash_set::LinkedHashSet;
+use std::{borrow::Borrow, hash::Hash, num::NonZeroUsize};
+
+/// A minimal LRU cache based on a `LinkedHashSet` with limited capacity.
+///
+/// If the length exceeds the set capacity, the oldest element will be removed
+/// In the limit, for each element inserted the oldest existing element will be removed.
+#[derive(Debug, Clone)]
+pub struct LruCache<T: Hash + Eq> {
+    limit: NonZeroUsize,
+    inner: LinkedHashSet<T>,
+}
+
+impl<T: Hash + Eq> LruCache<T> {
+    /// Creates a new `LruCache` using the given limit
+    pub fn new(limit: NonZeroUsize) -> Self {
+        Self { inner: LinkedHashSet::new(), limit }
+    }
+
+    /// Insert an element into the set.
+    ///
+    /// If the element is new (did not exist before [`LruCache::insert()`]) was called, then the
+    /// given length will be enforced and the oldest element will be removed if the limit was
+    /// exceeded.
+    ///
+    /// If the set did not have this value present, true is returned.
+    /// If the set did have this value present, false is returned.
+    pub fn insert(&mut self, entry: T) -> bool {
+        if self.inner.insert(entry) {
+            if self.limit.get() == self.inner.len() {
+                // remove the oldest element in the set
+                self.inner.pop_front();
+            }
+            return true
+        }
+        false
+    }
+
+    /// Returns `true` if the set contains a value.
+    pub fn contains<Q: ?Sized>(&self, value: &Q) -> bool
+    where
+        T: Borrow<Q>,
+        Q: Hash + Eq,
+    {
+        self.inner.contains(value)
+    }
+}
+
+impl<T> Extend<T> for LruCache<T>
+where
+    T: Eq + Hash,
+{
+    fn extend<I: IntoIterator<Item = T>>(&mut self, iter: I) {
+        for item in iter.into_iter() {
+            self.insert(item);
+        }
+    }
+}

crates/net/network/src/config.rs

@@ -1,4 +1,8 @@
-use crate::{peers::PeersConfig, session::SessionsConfig};
+use crate::{
+    import::{BlockImport, NoopBlockImport},
+    peers::PeersConfig,
+    session::SessionsConfig,
+};
 use reth_discv4::{Discv4Config, Discv4ConfigBuilder, DEFAULT_DISCOVERY_PORT};
 use reth_primitives::{Chain, ForkId, H256};
 use secp256k1::SecretKey;
@@ -30,6 +34,8 @@ pub struct NetworkConfig<C> {
     pub chain: Chain,
     /// Genesis hash of the network
     pub genesis_hash: H256,
+    /// The block importer type.
+    pub block_import: Box<dyn BlockImport>,
 }
 
 // === impl NetworkConfig ===
@@ -82,6 +88,8 @@ pub struct NetworkConfigBuilder<C> {
     chain: Chain,
     /// Network genesis hash
     genesis_hash: H256,
+    /// The block importer type.
+    block_import: Box<dyn BlockImport>,
 }
 
 // === impl NetworkConfigBuilder ===
@@ -100,6 +108,7 @@ impl<C> NetworkConfigBuilder<C> {
             fork_id: None,
             chain: Chain::Named(reth_primitives::rpc::Chain::Mainnet),
             genesis_hash: Default::default(),
+            block_import: Box::<NoopBlockImport>::default(),
         }
     }
 
@@ -109,6 +118,12 @@ impl<C> NetworkConfigBuilder<C> {
         self
     }
 
+    /// Sets the [`BlockImport`] type to configure.
+    pub fn block_import<T: BlockImport + 'static>(mut self, block_import: T) -> Self {
+        self.block_import = Box::new(block_import);
+        self
+    }
+
     /// Consumes the type and creates the actual [`NetworkConfig`]
     pub fn build(self) -> NetworkConfig<C> {
         let Self {
@@ -122,6 +137,7 @@ impl<C> NetworkConfigBuilder<C> {
             fork_id,
             chain,
             genesis_hash,
+            block_import,
         } = self;
         NetworkConfig {
             client,
@@ -138,6 +154,7 @@ impl<C> NetworkConfigBuilder<C> {
             fork_id,
             chain,
             genesis_hash,
+            block_import,
         }
     }
 }