feat(net): support eth/68

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,5 @@
 //! Error handling for (`EthStream`)[crate::EthStream]
-use crate::{errors::P2PStreamError, DisconnectReason};
+use crate::{errors::P2PStreamError, version::ParseVersionError, DisconnectReason};
 use reth_primitives::{Chain, ValidationError, H256};
 use std::io;
 
@@ -10,9 +10,13 @@ pub enum EthStreamError {
     #[error(transparent)]
     P2PStreamError(#[from] P2PStreamError),
     #[error(transparent)]
+    ParseVersionError(#[from] ParseVersionError),
+    #[error(transparent)]
     EthHandshakeError(#[from] EthHandshakeError),
     #[error("message size ({0}) exceeds max length (10MB)")]
     MessageTooBig(usize),
+    #[error("TransactionHashes invalid len of fields: {0} {1} {2}")]
+    TransactionHashesInvalidLenOfFields(usize, usize, 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,7 +77,8 @@ 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}");
@@ -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,7 +211,7 @@ 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}");
@@ -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

@@ -949,7 +949,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

@@ -1,6 +1,6 @@
 //! Types for broadcasting new data.
 use reth_codecs::derive_arbitrary;
-use reth_primitives::{Block, TransactionSigned, H256, U128};
+use reth_primitives::{Block, TransactionSigned, TxType, H256, U128};
 use reth_rlp::{RlpDecodable, RlpDecodableWrapper, RlpEncodable, RlpEncodableWrapper};
 use std::sync::Arc;
 
@@ -104,16 +104,83 @@ 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>,
+}
+
+impl TryFrom<NewPooledTransactionHashes> for NewPooledTransactionHashes68 {
+    type Error = &'static str;
+
+    fn try_from(msg: NewPooledTransactionHashes) -> Result<Self, Self::Error> {
+        if msg.types.is_none() {
+            return Err("`types` is none")
+        }
+        if msg.sizes.is_none() {
+            return Err("`sizes` is none")
+        }
+        Ok(Self { types: msg.types.unwrap(), sizes: msg.sizes.unwrap(), hashes: msg.hashes })
+    }
+}
+
+/// Same as [`NewPooledTransactionHashes68`] but types and sizes are Option. It's used internally
+/// instead of [`NewPooledTransactionHashes66`] and [`NewPooledTransactionHashes68`].
+#[derive(Clone, Debug, PartialEq, Eq, Default)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub struct NewPooledTransactionHashes {
+    /// Transaction types for new transactions that have appeared on the network.
+    pub types: Option<Vec<u8>>,
+    /// Transaction sizes for new transactions that have appeared on the network.
+    pub sizes: Option<Vec<usize>>,
+    /// Transaction hashes for new transactions that have appeared on the network.
+    pub hashes: Vec<H256>,
+}
+
+impl From<NewPooledTransactionHashes66> for NewPooledTransactionHashes {
+    fn from(msg: NewPooledTransactionHashes66) -> Self {
+        NewPooledTransactionHashes { types: None, sizes: None, hashes: msg.0 }
+    }
+}
+
+impl From<NewPooledTransactionHashes68> for NewPooledTransactionHashes {
+    fn from(msg: NewPooledTransactionHashes68) -> Self {
+        NewPooledTransactionHashes {
+            types: Some(msg.types),
+            sizes: Some(msg.sizes),
+            hashes: msg.hashes,
+        }
+    }
+}
+
+impl From<(Vec<TxType>, Vec<usize>, Vec<H256>)> for NewPooledTransactionHashes {
+    fn from((types, sizes, hashes): (Vec<TxType>, Vec<usize>, Vec<H256>)) -> Self {
+        NewPooledTransactionHashes {
+            types: Some(types.into_iter().map(|t| t as u8).collect()),
+            sizes: Some(sizes),
+            hashes,
+        }
     }
 }