feat(net): support eth/68 (#1361)

Co-authored-by: Georgios Konstantopoulos <me@gakonst.com>

crates/net/eth-wire/src/capability.rs

@@ -64,6 +64,12 @@ impl Capability {
     pub fn is_eth_v67(&self) -> bool {
         self.name == "eth" && self.version == 67
     }
+
+    /// Whether this is eth v68.
+    #[inline]
+    pub fn is_eth_v68(&self) -> bool {
+        self.name == "eth" && self.version == 68
+    }
 }
 
 #[cfg(any(test, feature = "arbitrary"))]
@@ -97,6 +103,7 @@ pub struct Capabilities {
     inner: Vec<Capability>,
     eth_66: bool,
     eth_67: bool,
+    eth_68: bool,
 }
 
 impl Capabilities {
@@ -115,7 +122,7 @@ impl Capabilities {
     /// Whether the peer supports `eth` sub-protocol.
     #[inline]
     pub fn supports_eth(&self) -> bool {
-        self.eth_67 || self.eth_66
+        self.eth_68 || self.eth_67 || self.eth_66
     }
 
     /// Whether this peer supports eth v66 protocol.
@@ -129,13 +136,20 @@ impl Capabilities {
     pub fn supports_eth_v67(&self) -> bool {
         self.eth_67
     }
+
+    /// Whether this peer supports eth v68 protocol.
+    #[inline]
+    pub fn supports_eth_v68(&self) -> bool {
+        self.eth_68
+    }
 }
 
 impl From<Vec<Capability>> for Capabilities {
     fn from(value: Vec<Capability>) -> Self {
         Self {
             eth_66: value.iter().any(Capability::is_eth_v66),
             eth_67: value.iter().any(Capability::is_eth_v67),
+            eth_68: value.iter().any(Capability::is_eth_v68),
             inner: value,
         }
     }
@@ -154,6 +168,7 @@ impl Decodable for Capabilities {
         Ok(Self {
             eth_66: inner.iter().any(Capability::is_eth_v66),
             eth_67: inner.iter().any(Capability::is_eth_v67),
+            eth_68: inner.iter().any(Capability::is_eth_v68),
             inner,
         })
     }
@@ -227,6 +242,15 @@ pub enum SharedCapabilityError {
 mod tests {
     use super::*;
 
+    #[test]
+    fn from_eth_68() {
+        let capability = SharedCapability::new("eth", 68, 0).unwrap();
+
+        assert_eq!(capability.name(), "eth");
+        assert_eq!(capability.version(), 68);
+        assert_eq!(capability, SharedCapability::Eth { version: EthVersion::Eth68, offset: 0 });
+    }
+
     #[test]
     fn from_eth_67() {
         let capability = SharedCapability::new("eth", 67, 0).unwrap();
@@ -244,4 +268,19 @@ mod tests {
         assert_eq!(capability.version(), 66);
         assert_eq!(capability, SharedCapability::Eth { version: EthVersion::Eth66, offset: 0 });
     }
+
+    #[test]
+    fn capabilities_supports_eth() {
+        let capabilities: Capabilities = vec![
+            Capability::new("eth".into(), 66),
+            Capability::new("eth".into(), 67),
+            Capability::new("eth".into(), 68),
+        ]
+        .into();
+
+        assert!(capabilities.supports_eth());
+        assert!(capabilities.supports_eth_v66());
+        assert!(capabilities.supports_eth_v67());
+        assert!(capabilities.supports_eth_v68());
+    }
 }

crates/net/eth-wire/src/errors/eth.rs

@@ -1,5 +1,7 @@
 //! Error handling for (`EthStream`)[crate::EthStream]
-use crate::{errors::P2PStreamError, DisconnectReason};
+use crate::{
+    errors::P2PStreamError, version::ParseVersionError, DisconnectReason, EthMessageID, EthVersion,
+};
 use reth_primitives::{Chain, ValidationError, H256};
 use std::io;
 
@@ -10,9 +12,15 @@ pub enum EthStreamError {
     #[error(transparent)]
     P2PStreamError(#[from] P2PStreamError),
     #[error(transparent)]
+    ParseVersionError(#[from] ParseVersionError),
+    #[error(transparent)]
     EthHandshakeError(#[from] EthHandshakeError),
+    #[error("For {0:?} version, message id({1:?}) is invalid")]
+    EthInvalidMessageError(EthVersion, EthMessageID),
     #[error("message size ({0}) exceeds max length (10MB)")]
     MessageTooBig(usize),
+    #[error("TransactionHashes invalid len of fields: hashes_len={hashes_len} types_len={types_len} sizes_len={sizes_len}")]
+    TransactionHashesInvalidLenOfFields { hashes_len: usize, types_len: usize, sizes_len: usize },
 }
 
 // === impl EthStreamError ===

crates/net/eth-wire/src/ethstream.rs

@@ -2,14 +2,15 @@ use crate::{
     errors::{EthHandshakeError, EthStreamError},
     message::{EthBroadcastMessage, ProtocolBroadcastMessage},
     types::{EthMessage, ProtocolMessage, Status},
+    EthVersion,
 };
 use futures::{ready, Sink, SinkExt, StreamExt};
 use pin_project::pin_project;
 use reth_primitives::{
     bytes::{Bytes, BytesMut},
     ForkFilter,
 };
-use reth_rlp::{Decodable, Encodable};
+use reth_rlp::Encodable;
 use std::{
     pin::Pin,
     task::{Context, Poll},
@@ -76,11 +77,12 @@ where
             return Err(EthStreamError::MessageTooBig(their_msg.len()))
         }
 
-        let msg = match ProtocolMessage::decode(&mut their_msg.as_ref()) {
+        let version = EthVersion::try_from(status.version)?;
+        let msg = match ProtocolMessage::decode_message(version, &mut their_msg.as_ref()) {
             Ok(m) => m,
             Err(err) => {
-                tracing::debug!("rlp decode error in eth handshake: msg={their_msg:x}");
-                return Err(err.into())
+                tracing::debug!("decode error in eth handshake: msg={their_msg:x}");
+                return Err(err)
             }
         };
 
@@ -120,7 +122,7 @@ where
 
                 // now we can create the `EthStream` because the peer has successfully completed
                 // the handshake
-                let stream = EthStream::new(self.inner);
+                let stream = EthStream::new(version, self.inner);
 
                 Ok((stream, resp))
             }
@@ -136,15 +138,21 @@ where
 #[pin_project]
 #[derive(Debug)]
 pub struct EthStream<S> {
+    version: EthVersion,
     #[pin]
     inner: S,
 }
 
 impl<S> EthStream<S> {
     /// Creates a new unauthed [`EthStream`] from a provided stream. You will need
     /// to manually handshake a peer.
-    pub fn new(inner: S) -> Self {
-        Self { inner }
+    pub fn new(version: EthVersion, inner: S) -> Self {
+        Self { version, inner }
+    }
+
+    /// Returns the eth version.
+    pub fn version(&self) -> EthVersion {
+        self.version
     }
 
     /// Returns the underlying stream.
@@ -203,11 +211,11 @@ where
             return Poll::Ready(Some(Err(EthStreamError::MessageTooBig(bytes.len()))))
         }
 
-        let msg = match ProtocolMessage::decode(&mut bytes.as_ref()) {
+        let msg = match ProtocolMessage::decode_message(*this.version, &mut bytes.as_ref()) {
             Ok(m) => m,
             Err(err) => {
-                tracing::debug!("rlp decode error: msg={bytes:x}");
-                return Poll::Ready(Some(Err(err.into())))
+                tracing::debug!("decode error: msg={bytes:x}");
+                return Poll::Ready(Some(Err(err)))
             }
         };
 
@@ -337,7 +345,7 @@ mod tests {
             // roughly based off of the design of tokio::net::TcpListener
             let (incoming, _) = listener.accept().await.unwrap();
             let stream = PassthroughCodec::default().framed(incoming);
-            let mut stream = EthStream::new(stream);
+            let mut stream = EthStream::new(EthVersion::Eth67, stream);
 
             // use the stream to get the next message
             let message = stream.next().await.unwrap().unwrap();
@@ -346,7 +354,7 @@ mod tests {
 
         let outgoing = TcpStream::connect(local_addr).await.unwrap();
         let sink = PassthroughCodec::default().framed(outgoing);
-        let mut client_stream = EthStream::new(sink);
+        let mut client_stream = EthStream::new(EthVersion::Eth67, sink);
 
         client_stream.send(test_msg).await.unwrap();
 
@@ -372,7 +380,7 @@ mod tests {
             // roughly based off of the design of tokio::net::TcpListener
             let (incoming, _) = listener.accept().await.unwrap();
             let stream = ECIESStream::incoming(incoming, server_key).await.unwrap();
-            let mut stream = EthStream::new(stream);
+            let mut stream = EthStream::new(EthVersion::Eth67, stream);
 
             // use the stream to get the next message
             let message = stream.next().await.unwrap().unwrap();
@@ -386,7 +394,7 @@ mod tests {
 
         let outgoing = TcpStream::connect(local_addr).await.unwrap();
         let outgoing = ECIESStream::connect(outgoing, client_key, server_id).await.unwrap();
-        let mut client_stream = EthStream::new(outgoing);
+        let mut client_stream = EthStream::new(EthVersion::Eth67, outgoing);
 
         client_stream.send(test_msg).await.unwrap();
 

crates/net/eth-wire/src/hello.rs

@@ -100,7 +100,7 @@ impl HelloMessageBuilder {
             protocol_version: protocol_version.unwrap_or_default(),
             client_version: client_version.unwrap_or_else(|| DEFAULT_CLIENT_VERSION.to_string()),
             capabilities: capabilities
-                .unwrap_or_else(|| vec![EthVersion::Eth66.into(), EthVersion::Eth67.into()]),
+                .unwrap_or_else(|| vec![EthVersion::Eth67.into(), EthVersion::Eth66.into()]),
             port: port.unwrap_or(30303),
             id,
         }

crates/net/eth-wire/src/p2pstream.rs

@@ -948,7 +948,7 @@ mod tests {
     #[test]
     fn test_peer_lower_capability_version() {
         let local_capabilities: Vec<Capability> =
-            vec![EthVersion::Eth66.into(), EthVersion::Eth67.into()];
+            vec![EthVersion::Eth66.into(), EthVersion::Eth67.into(), EthVersion::Eth68.into()];
         let peer_capabilities: Vec<Capability> = vec![EthVersion::Eth66.into()];
 
         let shared_capability =

crates/net/eth-wire/src/types/broadcast.rs

@@ -104,19 +104,33 @@ pub struct SharedTransactions(
 #[derive_arbitrary(rlp)]
 #[derive(Clone, Debug, PartialEq, Eq, RlpEncodableWrapper, RlpDecodableWrapper, Default)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
-pub struct NewPooledTransactionHashes(
+pub struct NewPooledTransactionHashes66(
     /// Transaction hashes for new transactions that have appeared on the network.
     /// Clients should request the transactions with the given hashes using a
     /// [`GetPooledTransactions`](crate::GetPooledTransactions) message.
     pub Vec<H256>,
 );
 
-impl From<Vec<H256>> for NewPooledTransactionHashes {
+impl From<Vec<H256>> for NewPooledTransactionHashes66 {
     fn from(v: Vec<H256>) -> Self {
-        NewPooledTransactionHashes(v)
+        NewPooledTransactionHashes66(v)
     }
 }
 
+/// Same as [`NewPooledTransactionHashes66`] but extends that that beside the transaction hashes,
+/// the node sends the transaction types and their sizes (as defined in EIP-2718) as well.
+#[derive_arbitrary(rlp)]
+#[derive(Clone, Debug, PartialEq, Eq, RlpEncodable, RlpDecodable, Default)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub struct NewPooledTransactionHashes68 {
+    /// Transaction types for new transactions that have appeared on the network.
+    pub types: Vec<u8>,
+    /// Transaction sizes for new transactions that have appeared on the network.
+    pub sizes: Vec<usize>,
+    /// Transaction hashes for new transactions that have appeared on the network.
+    pub hashes: Vec<H256>,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;