diff --git a/Cargo.lock b/Cargo.lock index 0ad5c1acd..5ce7b0c0e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7349,6 +7349,7 @@ dependencies = [ "reth-chainspec", "reth-codecs-derive", "reth-primitives", + "reth-primitives-traits", "serde", "thiserror", ] @@ -7784,6 +7785,7 @@ dependencies = [ "reth-network-peers", "reth-network-types", "reth-primitives", + "reth-primitives-traits", "reth-provider", "reth-storage-api", "reth-tasks", diff --git a/crates/net/eth-wire-types/Cargo.toml b/crates/net/eth-wire-types/Cargo.toml index 582ab7557..9ce712bf8 100644 --- a/crates/net/eth-wire-types/Cargo.toml +++ b/crates/net/eth-wire-types/Cargo.toml @@ -16,9 +16,9 @@ workspace = true reth-chainspec.workspace = true reth-codecs-derive.workspace = true reth-primitives.workspace = true +reth-primitives-traits.workspace = true # ethereum -alloy-consensus.workspace = true alloy-chains = { workspace = true, features = ["rlp"] } alloy-eips.workspace = true alloy-primitives.workspace = true @@ -54,7 +54,8 @@ arbitrary = [ "reth-chainspec/arbitrary", "alloy-consensus/arbitrary", "alloy-eips/arbitrary", - "alloy-primitives/arbitrary" + "alloy-primitives/arbitrary", + "reth-primitives-traits/arbitrary", ] serde = [ "dep:serde", diff --git a/crates/net/eth-wire-types/src/primitives.rs b/crates/net/eth-wire-types/src/primitives.rs index 04b8b429e..15cfaaff0 100644 --- a/crates/net/eth-wire-types/src/primitives.rs +++ b/crates/net/eth-wire-types/src/primitives.rs @@ -2,8 +2,8 @@ use std::fmt::Debug; -use alloy_consensus::BlockHeader; use alloy_rlp::{Decodable, Encodable}; +use reth_primitives_traits::{Block, BlockHeader}; /// Abstraction over primitive types which might appear in network messages. See /// [`crate::EthMessage`] for more context. @@ -34,7 +34,8 @@ pub trait NetworkPrimitives: + Eq + 'static; /// Full block type. - type Block: Encodable + type Block: Block
+ + Encodable + Decodable + Send + Sync diff --git a/crates/net/network/Cargo.toml b/crates/net/network/Cargo.toml index 148eef34b..09f81e63e 100644 --- a/crates/net/network/Cargo.toml +++ b/crates/net/network/Cargo.toml @@ -16,6 +16,7 @@ workspace = true reth-chainspec.workspace = true reth-fs-util.workspace = true reth-primitives = { workspace = true, features = ["secp256k1"] } +reth-primitives-traits.workspace = true reth-net-banlist.workspace = true reth-network-api.workspace = true reth-network-p2p.workspace = true @@ -130,7 +131,8 @@ test-utils = [ "reth-discv4/test-utils", "reth-network/test-utils", "reth-network-p2p/test-utils", - "reth-primitives/test-utils" + "reth-primitives/test-utils", + "reth-primitives-traits/test-utils", ] [[bench]] diff --git a/crates/net/network/src/import.rs b/crates/net/network/src/import.rs index 201dc3e4f..749b3c347 100644 --- a/crates/net/network/src/import.rs +++ b/crates/net/network/src/import.rs @@ -7,7 +7,7 @@ use reth_network_peers::PeerId; use crate::message::NewBlockMessage; /// Abstraction over block import. -pub trait BlockImport: std::fmt::Debug + Send + Sync { +pub trait BlockImport: std::fmt::Debug + Send + Sync { /// Invoked for a received `NewBlock` broadcast message from the peer. /// /// > When a `NewBlock` announcement message is received from a peer, the client first verifies @@ -15,35 +15,35 @@ pub trait BlockImport: std::fmt::Debug + Send + Sync { /// /// This is supposed to start verification. The results are then expected to be returned via /// [`BlockImport::poll`]. - fn on_new_block(&mut self, peer_id: PeerId, incoming_block: NewBlockMessage); + fn on_new_block(&mut self, peer_id: PeerId, incoming_block: NewBlockMessage); /// Returns the results of a [`BlockImport::on_new_block`] - fn poll(&mut self, cx: &mut Context<'_>) -> Poll; + fn poll(&mut self, cx: &mut Context<'_>) -> Poll>; } /// Outcome of the [`BlockImport`]'s block handling. #[derive(Debug)] -pub struct BlockImportOutcome { +pub struct BlockImportOutcome { /// Sender of the `NewBlock` message. pub peer: PeerId, /// The result after validating the block - pub result: Result, + pub result: Result, BlockImportError>, } /// Represents the successful validation of a received `NewBlock` message. #[derive(Debug)] -pub enum BlockValidation { +pub enum BlockValidation { /// Basic Header validity check, after which the block should be relayed to peers via a /// `NewBlock` message ValidHeader { /// received block - block: NewBlockMessage, + block: NewBlockMessage, }, /// Successfully imported: state-root matches after execution. The block should be relayed via /// `NewBlockHashes` ValidBlock { /// validated block. - block: NewBlockMessage, + block: NewBlockMessage, }, } @@ -62,10 +62,10 @@ pub enum BlockImportError { #[non_exhaustive] pub struct ProofOfStakeBlockImport; -impl BlockImport for ProofOfStakeBlockImport { - fn on_new_block(&mut self, _peer_id: PeerId, _incoming_block: NewBlockMessage) {} +impl BlockImport for ProofOfStakeBlockImport { + fn on_new_block(&mut self, _peer_id: PeerId, _incoming_block: NewBlockMessage) {} - fn poll(&mut self, _cx: &mut Context<'_>) -> Poll { + fn poll(&mut self, _cx: &mut Context<'_>) -> Poll> { Poll::Pending } } diff --git a/crates/net/network/src/message.rs b/crates/net/network/src/message.rs index bdb13875f..304057741 100644 --- a/crates/net/network/src/message.rs +++ b/crates/net/network/src/message.rs @@ -8,6 +8,7 @@ use std::{ task::{ready, Context, Poll}, }; +use alloy_consensus::BlockHeader; use alloy_primitives::{Bytes, B256}; use futures::FutureExt; use reth_eth_wire::{ @@ -23,30 +24,30 @@ use tokio::sync::oneshot; /// Internal form of a `NewBlock` message #[derive(Debug, Clone)] -pub struct NewBlockMessage { +pub struct NewBlockMessage { /// Hash of the block pub hash: B256, /// Raw received message - pub block: Arc, + pub block: Arc>, } // === impl NewBlockMessage === -impl NewBlockMessage { +impl NewBlockMessage { /// Returns the block number of the block pub fn number(&self) -> u64 { - self.block.block.header.number + self.block.block.header().number() } } /// All Bi-directional eth-message variants that can be sent to a session or received from a /// session. #[derive(Debug)] -pub enum PeerMessage { +pub enum PeerMessage { /// Announce new block hashes NewBlockHashes(NewBlockHashes), /// Broadcast new block. - NewBlock(NewBlockMessage), + NewBlock(NewBlockMessage), /// Received transactions _from_ the peer ReceivedTransaction(Transactions), /// Broadcast transactions _from_ local _to_ a peer. @@ -54,7 +55,7 @@ pub enum PeerMessage { /// Send new pooled transactions PooledTransactions(NewPooledTransactionHashes), /// All `eth` request variants. - EthRequest(PeerRequest), + EthRequest(PeerRequest), /// Other than eth namespace message Other(RawCapabilityMessage), } diff --git a/crates/net/network/src/session/active.rs b/crates/net/network/src/session/active.rs index 10048823c..f979a912c 100644 --- a/crates/net/network/src/session/active.rs +++ b/crates/net/network/src/session/active.rs @@ -11,18 +11,20 @@ use std::{ time::{Duration, Instant}, }; +use alloy_primitives::Sealable; use futures::{stream::Fuse, SinkExt, StreamExt}; use metrics::Gauge; use reth_eth_wire::{ errors::{EthHandshakeError, EthStreamError, P2PStreamError}, message::{EthBroadcastMessage, RequestPair}, - Capabilities, DisconnectP2P, DisconnectReason, EthMessage, + Capabilities, DisconnectP2P, DisconnectReason, EthMessage, NetworkPrimitives, }; use reth_metrics::common::mpsc::MeteredPollSender; use reth_network_api::PeerRequest; use reth_network_p2p::error::RequestError; use reth_network_peers::PeerId; use reth_network_types::session::config::INITIAL_REQUEST_TIMEOUT; +use reth_primitives_traits::Block; use rustc_hash::FxHashMap; use tokio::{ sync::{mpsc::error::TrySendError, oneshot}, @@ -62,11 +64,11 @@ const TIMEOUT_SCALING: u32 = 3; /// - incoming requests/broadcasts _from remote_ via the connection /// - responses for handled ETH requests received from the remote peer. #[allow(dead_code)] -pub(crate) struct ActiveSession { +pub(crate) struct ActiveSession { /// Keeps track of request ids. pub(crate) next_id: u64, /// The underlying connection. - pub(crate) conn: EthRlpxConnection, + pub(crate) conn: EthRlpxConnection, /// Identifier of the node we're connected to. pub(crate) remote_peer_id: PeerId, /// The address we're connected to. @@ -76,19 +78,19 @@ pub(crate) struct ActiveSession { /// Internal identifier of this session pub(crate) session_id: SessionId, /// Incoming commands from the manager - pub(crate) commands_rx: ReceiverStream, + pub(crate) commands_rx: ReceiverStream>, /// Sink to send messages to the [`SessionManager`](super::SessionManager). - pub(crate) to_session_manager: MeteredPollSender, + pub(crate) to_session_manager: MeteredPollSender>, /// A message that needs to be delivered to the session manager - pub(crate) pending_message_to_session: Option, + pub(crate) pending_message_to_session: Option>, /// Incoming internal requests which are delegated to the remote peer. - pub(crate) internal_request_tx: Fuse>, + pub(crate) internal_request_tx: Fuse>>, /// All requests sent to the remote peer we're waiting on a response - pub(crate) inflight_requests: FxHashMap, + pub(crate) inflight_requests: FxHashMap>>, /// All requests that were sent by the remote peer and we're waiting on an internal response - pub(crate) received_requests_from_remote: Vec, + pub(crate) received_requests_from_remote: Vec>, /// Buffered messages that should be handled and sent to the peer. - pub(crate) queued_outgoing: QueuedOutgoingMessages, + pub(crate) queued_outgoing: QueuedOutgoingMessages, /// The maximum time we wait for a response from a peer. pub(crate) internal_request_timeout: Arc, /// Interval when to check for timed out requests. @@ -97,10 +99,11 @@ pub(crate) struct ActiveSession { /// considered a protocol violation and the session will initiate a drop. pub(crate) protocol_breach_request_timeout: Duration, /// Used to reserve a slot to guarantee that the termination message is delivered - pub(crate) terminate_message: Option<(PollSender, ActiveSessionMessage)>, + pub(crate) terminate_message: + Option<(PollSender>, ActiveSessionMessage)>, } -impl ActiveSession { +impl ActiveSession { /// Returns `true` if the session is currently in the process of disconnecting fn is_disconnecting(&self) -> bool { self.conn.inner().is_disconnecting() @@ -122,7 +125,7 @@ impl ActiveSession { /// Handle a message read from the connection. /// /// Returns an error if the message is considered to be in violation of the protocol. - fn on_incoming_message(&mut self, msg: EthMessage) -> OnIncomingMessageOutcome { + fn on_incoming_message(&mut self, msg: EthMessage) -> OnIncomingMessageOutcome { /// A macro that handles an incoming request /// This creates a new channel and tries to send the sender half to the session while /// storing the receiver half internally so the pending response can be polled. @@ -182,7 +185,7 @@ impl ActiveSession { } EthMessage::NewBlock(msg) => { let block = - NewBlockMessage { hash: msg.block.header.hash_slow(), block: Arc::new(*msg) }; + NewBlockMessage { hash: msg.block.header().hash_slow(), block: Arc::new(*msg) }; self.try_emit_broadcast(PeerMessage::NewBlock(block)).into() } EthMessage::Transactions(msg) => { @@ -238,7 +241,7 @@ impl ActiveSession { } /// Handle an internal peer request that will be sent to the remote. - fn on_internal_peer_request(&mut self, request: PeerRequest, deadline: Instant) { + fn on_internal_peer_request(&mut self, request: PeerRequest, deadline: Instant) { let request_id = self.next_id(); let msg = request.create_request_message(request_id); self.queued_outgoing.push_back(msg.into()); @@ -251,7 +254,7 @@ impl ActiveSession { } /// Handle a message received from the internal network - fn on_internal_peer_message(&mut self, msg: PeerMessage) { + fn on_internal_peer_message(&mut self, msg: PeerMessage) { match msg { PeerMessage::NewBlockHashes(msg) => { self.queued_outgoing.push_back(EthMessage::NewBlockHashes(msg).into()); @@ -289,7 +292,7 @@ impl ActiveSession { /// Handle a Response to the peer /// /// This will queue the response to be sent to the peer - fn handle_outgoing_response(&mut self, id: u64, resp: PeerResponseResult) { + fn handle_outgoing_response(&mut self, id: u64, resp: PeerResponseResult) { match resp.try_into_message(id) { Ok(msg) => { self.queued_outgoing.push_back(msg.into()); @@ -304,7 +307,7 @@ impl ActiveSession { /// /// Returns the message if the bounded channel is currently unable to handle this message. #[allow(clippy::result_large_err)] - fn try_emit_broadcast(&self, message: PeerMessage) -> Result<(), ActiveSessionMessage> { + fn try_emit_broadcast(&self, message: PeerMessage) -> Result<(), ActiveSessionMessage> { let Some(sender) = self.to_session_manager.inner().get_ref() else { return Ok(()) }; match sender @@ -330,7 +333,7 @@ impl ActiveSession { /// /// Returns the message if the bounded channel is currently unable to handle this message. #[allow(clippy::result_large_err)] - fn try_emit_request(&self, message: PeerMessage) -> Result<(), ActiveSessionMessage> { + fn try_emit_request(&self, message: PeerMessage) -> Result<(), ActiveSessionMessage> { let Some(sender) = self.to_session_manager.inner().get_ref() else { return Ok(()) }; match sender @@ -470,7 +473,7 @@ impl ActiveSession { } } -impl Future for ActiveSession { +impl Future for ActiveSession { type Output = (); fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { @@ -656,20 +659,20 @@ impl Future for ActiveSession { } /// Tracks a request received from the peer -pub(crate) struct ReceivedRequest { +pub(crate) struct ReceivedRequest { /// Protocol Identifier request_id: u64, /// Receiver half of the channel that's supposed to receive the proper response. - rx: PeerResponse, + rx: PeerResponse, /// Timestamp when we read this msg from the wire. #[allow(dead_code)] received: Instant, } /// A request that waits for a response from the peer -pub(crate) struct InflightRequest { +pub(crate) struct InflightRequest { /// Request we sent to peer and the internal response channel - request: RequestState, + request: RequestState, /// Instant when the request was sent timestamp: Instant, /// Time limit for the response @@ -678,7 +681,7 @@ pub(crate) struct InflightRequest { // === impl InflightRequest === -impl InflightRequest { +impl InflightRequest> { /// Returns true if the request is timedout #[inline] fn is_timed_out(&self, now: Instant) -> bool { @@ -703,17 +706,19 @@ impl InflightRequest { } /// All outcome variants when handling an incoming message -enum OnIncomingMessageOutcome { +enum OnIncomingMessageOutcome { /// Message successfully handled. Ok, /// Message is considered to be in violation of the protocol - BadMessage { error: EthStreamError, message: EthMessage }, + BadMessage { error: EthStreamError, message: EthMessage }, /// Currently no capacity to handle the message - NoCapacity(ActiveSessionMessage), + NoCapacity(ActiveSessionMessage), } -impl From> for OnIncomingMessageOutcome { - fn from(res: Result<(), ActiveSessionMessage>) -> Self { +impl From>> + for OnIncomingMessageOutcome +{ + fn from(res: Result<(), ActiveSessionMessage>) -> Self { match res { Ok(_) => Self::Ok, Err(msg) => Self::NoCapacity(msg), @@ -721,29 +726,29 @@ impl From> for OnIncomingMessageOutcome { } } -enum RequestState { +enum RequestState { /// Waiting for the response - Waiting(PeerRequest), + Waiting(R), /// Request already timed out TimedOut, } /// Outgoing messages that can be sent over the wire. -pub(crate) enum OutgoingMessage { +pub(crate) enum OutgoingMessage { /// A message that is owned. - Eth(EthMessage), + Eth(EthMessage), /// A message that may be shared by multiple sessions. - Broadcast(EthBroadcastMessage), + Broadcast(EthBroadcastMessage), } -impl From for OutgoingMessage { - fn from(value: EthMessage) -> Self { +impl From> for OutgoingMessage { + fn from(value: EthMessage) -> Self { Self::Eth(value) } } -impl From for OutgoingMessage { - fn from(value: EthBroadcastMessage) -> Self { +impl From> for OutgoingMessage { + fn from(value: EthBroadcastMessage) -> Self { Self::Broadcast(value) } } @@ -760,22 +765,22 @@ fn calculate_new_timeout(current_timeout: Duration, estimated_rtt: Duration) -> } /// A helper struct that wraps the queue of outgoing messages and a metric to track their count -pub(crate) struct QueuedOutgoingMessages { - messages: VecDeque, +pub(crate) struct QueuedOutgoingMessages { + messages: VecDeque>, count: Gauge, } -impl QueuedOutgoingMessages { +impl QueuedOutgoingMessages { pub(crate) const fn new(metric: Gauge) -> Self { Self { messages: VecDeque::new(), count: metric } } - pub(crate) fn push_back(&mut self, message: OutgoingMessage) { + pub(crate) fn push_back(&mut self, message: OutgoingMessage) { self.messages.push_back(message); self.count.increment(1); } - pub(crate) fn pop_front(&mut self) -> Option { + pub(crate) fn pop_front(&mut self) -> Option> { self.messages.pop_front().inspect(|_| self.count.decrement(1)) } @@ -791,8 +796,8 @@ mod tests { use reth_chainspec::MAINNET; use reth_ecies::stream::ECIESStream; use reth_eth_wire::{ - EthStream, GetBlockBodies, HelloMessageWithProtocols, P2PStream, Status, StatusBuilder, - UnauthedEthStream, UnauthedP2PStream, + EthNetworkPrimitives, EthStream, GetBlockBodies, HelloMessageWithProtocols, P2PStream, + Status, StatusBuilder, UnauthedEthStream, UnauthedP2PStream, }; use reth_network_peers::pk2id; use reth_network_types::session::config::PROTOCOL_BREACH_REQUEST_TIMEOUT; @@ -808,11 +813,11 @@ mod tests { HelloMessageWithProtocols::builder(pk2id(&server_key.public_key(SECP256K1))).build() } - struct SessionBuilder { + struct SessionBuilder { _remote_capabilities: Arc, - active_session_tx: mpsc::Sender, - active_session_rx: ReceiverStream, - to_sessions: Vec>, + active_session_tx: mpsc::Sender>, + active_session_rx: ReceiverStream>, + to_sessions: Vec>>, secret_key: SecretKey, local_peer_id: PeerId, hello: HelloMessageWithProtocols, @@ -821,7 +826,7 @@ mod tests { next_id: usize, } - impl SessionBuilder { + impl SessionBuilder { fn next_id(&mut self) -> SessionId { let id = self.next_id; self.next_id += 1; @@ -858,7 +863,7 @@ mod tests { }) } - async fn connect_incoming(&mut self, stream: TcpStream) -> ActiveSession { + async fn connect_incoming(&mut self, stream: TcpStream) -> ActiveSession { let remote_addr = stream.local_addr().unwrap(); let session_id = self.next_id(); let (_disconnect_tx, disconnect_rx) = oneshot::channel(); diff --git a/crates/net/network/src/session/conn.rs b/crates/net/network/src/session/conn.rs index 628c880c8..5329f0102 100644 --- a/crates/net/network/src/session/conn.rs +++ b/crates/net/network/src/session/conn.rs @@ -11,16 +11,16 @@ use reth_eth_wire::{ errors::EthStreamError, message::EthBroadcastMessage, multiplex::{ProtocolProxy, RlpxSatelliteStream}, - EthMessage, EthStream, EthVersion, P2PStream, + EthMessage, EthNetworkPrimitives, EthStream, EthVersion, NetworkPrimitives, P2PStream, }; use tokio::net::TcpStream; /// The type of the underlying peer network connection. -pub type EthPeerConnection = EthStream>>; +pub type EthPeerConnection = EthStream>, N>; /// Various connection types that at least support the ETH protocol. -pub type EthSatelliteConnection = - RlpxSatelliteStream, EthStream>; +pub type EthSatelliteConnection = + RlpxSatelliteStream, EthStream>; /// Connection types that support the ETH protocol. /// @@ -30,14 +30,14 @@ pub type EthSatelliteConnection = // This type is boxed because the underlying stream is ~6KB, // mostly coming from `P2PStream`'s `snap::Encoder` (2072), and `ECIESStream` (3600). #[derive(Debug)] -pub enum EthRlpxConnection { +pub enum EthRlpxConnection { /// A connection that only supports the ETH protocol. - EthOnly(Box), + EthOnly(Box>), /// A connection that supports the ETH protocol and __at least one other__ `RLPx` protocol. - Satellite(Box), + Satellite(Box>), } -impl EthRlpxConnection { +impl EthRlpxConnection { /// Returns the negotiated ETH version. #[inline] pub(crate) const fn version(&self) -> EthVersion { @@ -78,7 +78,7 @@ impl EthRlpxConnection { #[inline] pub fn start_send_broadcast( &mut self, - item: EthBroadcastMessage, + item: EthBroadcastMessage, ) -> Result<(), EthStreamError> { match self { Self::EthOnly(conn) => conn.start_send_broadcast(item), @@ -87,16 +87,16 @@ impl EthRlpxConnection { } } -impl From for EthRlpxConnection { +impl From> for EthRlpxConnection { #[inline] - fn from(conn: EthPeerConnection) -> Self { + fn from(conn: EthPeerConnection) -> Self { Self::EthOnly(Box::new(conn)) } } -impl From for EthRlpxConnection { +impl From> for EthRlpxConnection { #[inline] - fn from(conn: EthSatelliteConnection) -> Self { + fn from(conn: EthSatelliteConnection) -> Self { Self::Satellite(Box::new(conn)) } } @@ -112,22 +112,22 @@ macro_rules! delegate_call { } } -impl Stream for EthRlpxConnection { - type Item = Result; +impl Stream for EthRlpxConnection { + type Item = Result, EthStreamError>; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { delegate_call!(self.poll_next(cx)) } } -impl Sink for EthRlpxConnection { +impl Sink> for EthRlpxConnection { type Error = EthStreamError; fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { delegate_call!(self.poll_ready(cx)) } - fn start_send(self: Pin<&mut Self>, item: EthMessage) -> Result<(), Self::Error> { + fn start_send(self: Pin<&mut Self>, item: EthMessage) -> Result<(), Self::Error> { delegate_call!(self.start_send(item)) } diff --git a/crates/net/network/src/session/handle.rs b/crates/net/network/src/session/handle.rs index a022e6704..f80428630 100644 --- a/crates/net/network/src/session/handle.rs +++ b/crates/net/network/src/session/handle.rs @@ -5,7 +5,7 @@ use std::{io, net::SocketAddr, sync::Arc, time::Instant}; use reth_ecies::ECIESError; use reth_eth_wire::{ capability::CapabilityMessage, errors::EthStreamError, Capabilities, DisconnectReason, - EthVersion, Status, + EthVersion, NetworkPrimitives, Status, }; use reth_network_api::PeerInfo; use reth_network_peers::{NodeRecord, PeerId}; @@ -54,7 +54,7 @@ impl PendingSessionHandle { /// Within an active session that supports the `Ethereum Wire Protocol `, three high-level tasks can /// be performed: chain synchronization, block propagation and transaction exchange. #[derive(Debug)] -pub struct ActiveSessionHandle { +pub struct ActiveSessionHandle { /// The direction of the session pub(crate) direction: Direction, /// The assigned id for this session @@ -68,7 +68,7 @@ pub struct ActiveSessionHandle { /// Announced capabilities of the peer. pub(crate) capabilities: Arc, /// Sender half of the command channel used send commands _to_ the spawned session - pub(crate) commands_to_session: mpsc::Sender, + pub(crate) commands_to_session: mpsc::Sender>, /// The client's name and version pub(crate) client_version: Arc, /// The address we're connected to @@ -81,7 +81,7 @@ pub struct ActiveSessionHandle { // === impl ActiveSessionHandle === -impl ActiveSessionHandle { +impl ActiveSessionHandle { /// Sends a disconnect command to the session. pub fn disconnect(&self, reason: Option) { // Note: we clone the sender which ensures the channel has capacity to send the message @@ -93,7 +93,7 @@ impl ActiveSessionHandle { pub async fn try_disconnect( &self, reason: Option, - ) -> Result<(), SendError> { + ) -> Result<(), SendError>> { self.commands_to_session.clone().send(SessionCommand::Disconnect { reason }).await } @@ -162,7 +162,7 @@ impl ActiveSessionHandle { /// /// A session starts with a `Handshake`, followed by a `Hello` message which #[derive(Debug)] -pub enum PendingSessionEvent { +pub enum PendingSessionEvent { /// Represents a successful `Hello` and `Status` exchange: Established { /// An internal identifier for the established session @@ -179,7 +179,7 @@ pub enum PendingSessionEvent { status: Arc, /// The actual connection stream which can be used to send and receive `eth` protocol /// messages - conn: EthRlpxConnection, + conn: EthRlpxConnection, /// The direction of the session, either `Inbound` or `Outgoing` direction: Direction, /// The remote node's user agent, usually containing the client name and version @@ -222,20 +222,20 @@ pub enum PendingSessionEvent { /// Commands that can be sent to the spawned session. #[derive(Debug)] -pub enum SessionCommand { +pub enum SessionCommand { /// Disconnect the connection Disconnect { /// Why the disconnect was initiated reason: Option, }, /// Sends a message to the peer - Message(PeerMessage), + Message(PeerMessage), } /// Message variants an active session can produce and send back to the /// [`SessionManager`](crate::session::SessionManager) #[derive(Debug)] -pub enum ActiveSessionMessage { +pub enum ActiveSessionMessage { /// Session was gracefully disconnected. Disconnected { /// The remote node's public key @@ -257,7 +257,7 @@ pub enum ActiveSessionMessage { /// Identifier of the remote peer. peer_id: PeerId, /// Message received from the peer. - message: PeerMessage, + message: PeerMessage, }, /// Received a message that does not match the announced capabilities of the peer. InvalidMessage { diff --git a/crates/net/network/src/session/mod.rs b/crates/net/network/src/session/mod.rs index 30b1cda9d..a95f0e889 100644 --- a/crates/net/network/src/session/mod.rs +++ b/crates/net/network/src/session/mod.rs @@ -28,11 +28,11 @@ use futures::{future::Either, io, FutureExt, StreamExt}; use reth_ecies::{stream::ECIESStream, ECIESError}; use reth_eth_wire::{ capability::CapabilityMessage, errors::EthStreamError, multiplex::RlpxProtocolMultiplexer, - Capabilities, DisconnectReason, EthVersion, HelloMessageWithProtocols, Status, - UnauthedEthStream, UnauthedP2PStream, + Capabilities, DisconnectReason, EthVersion, HelloMessageWithProtocols, NetworkPrimitives, + Status, UnauthedEthStream, UnauthedP2PStream, }; use reth_metrics::common::mpsc::MeteredPollSender; -use reth_network_api::PeerRequestSender; +use reth_network_api::{PeerRequest, PeerRequestSender}; use reth_network_peers::PeerId; use reth_network_types::SessionsConfig; use reth_primitives::{ForkFilter, ForkId, ForkTransition, Head}; @@ -62,7 +62,7 @@ pub struct SessionId(usize); /// Manages a set of sessions. #[must_use = "Session Manager must be polled to process session events."] #[derive(Debug)] -pub struct SessionManager { +pub struct SessionManager { /// Tracks the identifier for the next session. next_id: usize, /// Keeps track of all sessions @@ -93,21 +93,21 @@ pub struct SessionManager { /// session is authenticated, it can be moved to the `active_session` set. pending_sessions: FxHashMap, /// All active sessions that are ready to exchange messages. - active_sessions: HashMap, + active_sessions: HashMap>, /// The original Sender half of the [`PendingSessionEvent`] channel. /// /// When a new (pending) session is created, the corresponding [`PendingSessionHandle`] will /// get a clone of this sender half. - pending_sessions_tx: mpsc::Sender, + pending_sessions_tx: mpsc::Sender>, /// Receiver half that listens for [`PendingSessionEvent`] produced by pending sessions. - pending_session_rx: ReceiverStream, + pending_session_rx: ReceiverStream>, /// The original Sender half of the [`ActiveSessionMessage`] channel. /// /// When active session state is reached, the corresponding [`ActiveSessionHandle`] will get a /// clone of this sender half. - active_session_tx: MeteredPollSender, + active_session_tx: MeteredPollSender>, /// Receiver half that listens for [`ActiveSessionMessage`] produced by pending sessions. - active_session_rx: ReceiverStream, + active_session_rx: ReceiverStream>, /// Additional `RLPx` sub-protocols to be used by the session manager. extra_protocols: RlpxSubProtocols, /// Tracks the ongoing graceful disconnections attempts for incoming connections. @@ -118,7 +118,7 @@ pub struct SessionManager { // === impl SessionManager === -impl SessionManager { +impl SessionManager { /// Creates a new empty [`SessionManager`]. #[allow(clippy::too_many_arguments)] pub fn new( @@ -182,7 +182,7 @@ impl SessionManager { } /// Returns a borrowed reference to the active sessions. - pub const fn active_sessions(&self) -> &HashMap { + pub const fn active_sessions(&self) -> &HashMap> { &self.active_sessions } @@ -348,7 +348,7 @@ impl SessionManager { } /// Sends a message to the peer's session - pub fn send_message(&mut self, peer_id: &PeerId, msg: PeerMessage) { + pub fn send_message(&mut self, peer_id: &PeerId, msg: PeerMessage) { if let Some(session) = self.active_sessions.get_mut(peer_id) { let _ = session.commands_to_session.try_send(SessionCommand::Message(msg)).inspect_err( |e| { @@ -373,7 +373,7 @@ impl SessionManager { } /// Removes the [`PendingSessionHandle`] if it exists. - fn remove_active_session(&mut self, id: &PeerId) -> Option { + fn remove_active_session(&mut self, id: &PeerId) -> Option> { let session = self.active_sessions.remove(id)?; self.counter.dec_active(&session.direction); Some(session) @@ -411,7 +411,7 @@ impl SessionManager { /// This polls all the session handles and returns [`SessionEvent`]. /// /// Active sessions are prioritized. - pub(crate) fn poll(&mut self, cx: &mut Context<'_>) -> Poll { + pub(crate) fn poll(&mut self, cx: &mut Context<'_>) -> Poll> { // Poll events from active sessions match self.active_session_rx.poll_next_unpin(cx) { Poll::Pending => {} @@ -663,7 +663,7 @@ impl DisconnectionsCounter { /// Events produced by the [`SessionManager`] #[derive(Debug)] -pub enum SessionEvent { +pub enum SessionEvent { /// A new session was successfully authenticated. /// /// This session is now able to exchange data. @@ -681,7 +681,7 @@ pub enum SessionEvent { /// The Status message the peer sent during the `eth` handshake status: Arc, /// The channel for sending messages to the peer with the session - messages: PeerRequestSender, + messages: PeerRequestSender>, /// The direction of the session, either `Inbound` or `Outgoing` direction: Direction, /// The maximum time that the session waits for a response from the peer before timing out @@ -702,7 +702,7 @@ pub enum SessionEvent { /// The remote node's public key peer_id: PeerId, /// Message received from the peer. - message: PeerMessage, + message: PeerMessage, }, /// Received a message that does not match the announced capabilities of the peer. InvalidMessage { @@ -797,12 +797,12 @@ impl PendingSessionHandshakeError { pub struct ExceedsSessionLimit(pub(crate) u32); /// Starts a pending session authentication with a timeout. -pub(crate) async fn pending_session_with_timeout( +pub(crate) async fn pending_session_with_timeout( timeout: Duration, session_id: SessionId, remote_addr: SocketAddr, direction: Direction, - events: mpsc::Sender, + events: mpsc::Sender>, f: F, ) where F: Future, @@ -823,11 +823,11 @@ pub(crate) async fn pending_session_with_timeout( /// /// This will wait for the _incoming_ handshake request and answer it. #[allow(clippy::too_many_arguments)] -pub(crate) async fn start_pending_incoming_session( +pub(crate) async fn start_pending_incoming_session( disconnect_rx: oneshot::Receiver<()>, session_id: SessionId, stream: TcpStream, - events: mpsc::Sender, + events: mpsc::Sender>, remote_addr: SocketAddr, secret_key: SecretKey, hello: HelloMessageWithProtocols, @@ -854,9 +854,9 @@ pub(crate) async fn start_pending_incoming_session( /// Starts the authentication process for a connection initiated by a remote peer. #[instrument(skip_all, fields(%remote_addr, peer_id), target = "net")] #[allow(clippy::too_many_arguments)] -async fn start_pending_outbound_session( +async fn start_pending_outbound_session( disconnect_rx: oneshot::Receiver<()>, - events: mpsc::Sender, + events: mpsc::Sender>, session_id: SessionId, remote_addr: SocketAddr, remote_peer_id: PeerId, @@ -903,9 +903,9 @@ async fn start_pending_outbound_session( /// Authenticates a session #[allow(clippy::too_many_arguments)] -async fn authenticate( +async fn authenticate( disconnect_rx: oneshot::Receiver<()>, - events: mpsc::Sender, + events: mpsc::Sender>, stream: TcpStream, session_id: SessionId, remote_addr: SocketAddr, @@ -986,7 +986,7 @@ async fn get_ecies_stream( /// If additional [`RlpxSubProtocolHandlers`] are provided, the hello message will be updated to /// also negotiate the additional protocols. #[allow(clippy::too_many_arguments)] -async fn authenticate_stream( +async fn authenticate_stream( stream: UnauthedP2PStream>, session_id: SessionId, remote_addr: SocketAddr, @@ -996,7 +996,7 @@ async fn authenticate_stream( mut status: Status, fork_filter: ForkFilter, mut extra_handlers: RlpxSubProtocolHandlers, -) -> PendingSessionEvent { +) -> PendingSessionEvent { // Add extra protocols to the hello message extra_handlers.retain(|handler| hello.try_add_protocol(handler.protocol()).is_ok()); diff --git a/crates/net/network/src/state.rs b/crates/net/network/src/state.rs index 9ad7b5351..3bafbf258 100644 --- a/crates/net/network/src/state.rs +++ b/crates/net/network/src/state.rs @@ -12,6 +12,7 @@ use std::{ task::{Context, Poll}, }; +use alloy_consensus::BlockHeader; use alloy_primitives::B256; use rand::seq::SliceRandom; use reth_eth_wire::{ @@ -22,6 +23,7 @@ use reth_network_api::{DiscoveredEvent, DiscoveryEvent, PeerRequest, PeerRequest use reth_network_peers::PeerId; use reth_network_types::{PeerAddr, PeerKind}; use reth_primitives::ForkId; +use reth_primitives_traits::Block; use tokio::sync::oneshot; use tracing::{debug, trace}; @@ -78,7 +80,7 @@ pub struct NetworkState { /// Manages connections to peers. peers_manager: PeersManager, /// Buffered messages until polled. - queued_messages: VecDeque, + queued_messages: VecDeque>, /// The client type that can interact with the chain. /// /// This type is used to fetch the block number after we established a session and received the @@ -185,12 +187,12 @@ impl NetworkState { /// > the total number of peers) using the `NewBlock` message. /// /// See also - pub(crate) fn announce_new_block(&mut self, msg: NewBlockMessage) { + pub(crate) fn announce_new_block(&mut self, msg: NewBlockMessage) { // send a `NewBlock` message to a fraction of the connected peers (square root of the total // number of peers) let num_propagate = (self.active_peers.len() as f64).sqrt() as u64 + 1; - let number = msg.block.block.header.number; + let number = msg.block.block.header().number(); let mut count = 0; // Shuffle to propagate to a random sample of peers on every block announcement @@ -227,8 +229,8 @@ impl NetworkState { /// Completes the block propagation process started in [`NetworkState::announce_new_block()`] /// but sending `NewBlockHash` broadcast to all peers that haven't seen it yet. - pub(crate) fn announce_new_block_hash(&mut self, msg: NewBlockMessage) { - let number = msg.block.block.header.number; + pub(crate) fn announce_new_block_hash(&mut self, msg: NewBlockMessage) { + let number = msg.block.block.header().number(); let hashes = NewBlockHashes(vec![BlockHashNumber { hash: msg.hash, number }]); for (peer_id, peer) in &mut self.active_peers { if peer.blocks.contains(&msg.hash) { @@ -385,7 +387,10 @@ impl NetworkState { } /// Handle the outcome of processed response, for example directly queue another request. - fn on_block_response_outcome(&mut self, outcome: BlockResponseOutcome) -> Option { + fn on_block_response_outcome( + &mut self, + outcome: BlockResponseOutcome, + ) -> Option> { match outcome { BlockResponseOutcome::Request(peer, request) => { self.handle_block_request(peer, request); @@ -406,7 +411,7 @@ impl NetworkState { &mut self, peer: PeerId, resp: PeerResponseResult, - ) -> Option { + ) -> Option> { match resp { PeerResponseResult::BlockHeaders(res) => { let outcome = self.state_fetcher.on_block_headers_response(peer, res)?; @@ -421,7 +426,7 @@ impl NetworkState { } /// Advances the state - pub(crate) fn poll(&mut self, cx: &mut Context<'_>) -> Poll { + pub(crate) fn poll(&mut self, cx: &mut Context<'_>) -> Poll> { loop { // drain buffered messages if let Some(message) = self.queued_messages.pop_front() { @@ -515,13 +520,13 @@ pub(crate) struct ActivePeer { /// Message variants triggered by the [`NetworkState`] #[derive(Debug)] -pub(crate) enum StateAction { +pub(crate) enum StateAction { /// Dispatch a `NewBlock` message to the peer NewBlock { /// Target of the message peer_id: PeerId, /// The `NewBlock` message - block: NewBlockMessage, + block: NewBlockMessage, }, NewBlockHashes { /// Target of the message diff --git a/crates/net/network/src/swarm.rs b/crates/net/network/src/swarm.rs index c1fe9f9e2..655934f20 100644 --- a/crates/net/network/src/swarm.rs +++ b/crates/net/network/src/swarm.rs @@ -9,9 +9,9 @@ use std::{ use futures::Stream; use reth_eth_wire::{ capability::CapabilityMessage, errors::EthStreamError, Capabilities, DisconnectReason, - EthVersion, Status, + EthNetworkPrimitives, EthVersion, NetworkPrimitives, Status, }; -use reth_network_api::PeerRequestSender; +use reth_network_api::{PeerRequest, PeerRequestSender}; use reth_network_peers::PeerId; use tracing::trace; @@ -50,23 +50,23 @@ use crate::{ /// `include_mmd!("docs/mermaid/swarm.mmd`") #[derive(Debug)] #[must_use = "Swarm does nothing unless polled"] -pub(crate) struct Swarm { +pub(crate) struct Swarm { /// Listens for new incoming connections. incoming: ConnectionListener, /// All sessions. - sessions: SessionManager, + sessions: SessionManager, /// Tracks the entire state of the network and handles events received from the sessions. - state: NetworkState, + state: NetworkState, } // === impl Swarm === -impl Swarm { +impl Swarm { /// Configures a new swarm instance. pub(crate) const fn new( incoming: ConnectionListener, - sessions: SessionManager, - state: NetworkState, + sessions: SessionManager, + state: NetworkState, ) -> Self { Self { incoming, sessions, state } } @@ -77,12 +77,12 @@ impl Swarm { } /// Access to the state. - pub(crate) const fn state(&self) -> &NetworkState { + pub(crate) const fn state(&self) -> &NetworkState { &self.state } /// Mutable access to the state. - pub(crate) fn state_mut(&mut self) -> &mut NetworkState { + pub(crate) fn state_mut(&mut self) -> &mut NetworkState { &mut self.state } @@ -92,17 +92,17 @@ impl Swarm { } /// Access to the [`SessionManager`]. - pub(crate) const fn sessions(&self) -> &SessionManager { + pub(crate) const fn sessions(&self) -> &SessionManager { &self.sessions } /// Mutable access to the [`SessionManager`]. - pub(crate) fn sessions_mut(&mut self) -> &mut SessionManager { + pub(crate) fn sessions_mut(&mut self) -> &mut SessionManager { &mut self.sessions } } -impl Swarm { +impl Swarm { /// Triggers a new outgoing connection to the given node pub(crate) fn dial_outbound(&mut self, remote_addr: SocketAddr, remote_id: PeerId) { self.sessions.dial_outbound(remote_addr, remote_id) @@ -112,7 +112,7 @@ impl Swarm { /// /// This either updates the state or produces a new [`SwarmEvent`] that is bubbled up to the /// manager. - fn on_session_event(&mut self, event: SessionEvent) -> Option { + fn on_session_event(&mut self, event: SessionEvent) -> Option> { match event { SessionEvent::SessionEstablished { peer_id, @@ -181,7 +181,7 @@ impl Swarm { /// Callback for events produced by [`ConnectionListener`]. /// /// Depending on the event, this will produce a new [`SwarmEvent`]. - fn on_connection(&mut self, event: ListenerEvent) -> Option { + fn on_connection(&mut self, event: ListenerEvent) -> Option> { match event { ListenerEvent::Error(err) => return Some(SwarmEvent::TcpListenerError(err)), ListenerEvent::ListenerClosed { local_address: address } => { @@ -229,7 +229,7 @@ impl Swarm { } /// Hook for actions pulled from the state - fn on_state_action(&mut self, event: StateAction) -> Option { + fn on_state_action(&mut self, event: StateAction) -> Option> { match event { StateAction::Connect { remote_addr, peer_id } => { self.dial_outbound(remote_addr, peer_id); @@ -286,8 +286,8 @@ impl Swarm { } } -impl Stream for Swarm { - type Item = SwarmEvent; +impl Stream for Swarm { + type Item = SwarmEvent; /// This advances all components. /// @@ -338,13 +338,13 @@ impl Stream for Swarm { /// All events created or delegated by the [`Swarm`] that represents changes to the state of the /// network. -pub(crate) enum SwarmEvent { +pub(crate) enum SwarmEvent { /// Events related to the actual network protocol. ValidMessage { /// The peer that sent the message peer_id: PeerId, /// Message received from the peer - message: PeerMessage, + message: PeerMessage, }, /// Received a message that does not match the announced capabilities of the peer. InvalidCapabilityMessage { @@ -394,7 +394,7 @@ pub(crate) enum SwarmEvent { capabilities: Arc, /// negotiated eth version version: EthVersion, - messages: PeerRequestSender, + messages: PeerRequestSender>, status: Arc, direction: Direction, }, diff --git a/crates/primitives-traits/src/block/body.rs b/crates/primitives-traits/src/block/body.rs index 14941ffed..6ec184a21 100644 --- a/crates/primitives-traits/src/block/body.rs +++ b/crates/primitives-traits/src/block/body.rs @@ -6,7 +6,7 @@ use alloy_consensus::{BlockHeader, Transaction, TxType}; use alloy_eips::{eip4895::Withdrawal, eip7685::Requests}; use alloy_primitives::{Address, B256}; -use crate::{Block, InMemorySize}; +use crate::InMemorySize; /// Abstraction for block's body. pub trait BlockBody: @@ -47,11 +47,6 @@ pub trait BlockBody: /// Returns [`Requests`] in block, if any. fn requests(&self) -> Option<&Requests>; - /// Create a [`Block`] from the body and its header. - fn into_block>(self, header: Self::Header) -> T { - T::from((header, self)) - } - /// Calculate the transaction root for the block body. fn calculate_tx_root(&self) -> B256; diff --git a/crates/primitives-traits/src/block/mod.rs b/crates/primitives-traits/src/block/mod.rs index cfc9e9a55..33008c438 100644 --- a/crates/primitives-traits/src/block/mod.rs +++ b/crates/primitives-traits/src/block/mod.rs @@ -3,12 +3,11 @@ pub mod body; pub mod header; -use alloc::{fmt, vec::Vec}; +use alloc::fmt; -use alloy_primitives::{Address, B256}; use reth_codecs::Compact; -use crate::{BlockBody, BlockHeader, FullBlockHeader, InMemorySize}; +use crate::{BlockHeader, FullBlockHeader, InMemorySize}; /// Helper trait that unifies all behaviour required by block to support full node operations. pub trait FullBlock: Block + Compact {} @@ -30,79 +29,17 @@ pub trait Block: + Eq + serde::Serialize + for<'a> serde::Deserialize<'a> - + From<(Self::Header, Self::Body)> - + Into<(Self::Header, Self::Body)> + InMemorySize { /// Header part of the block. type Header: BlockHeader; /// The block's body contains the transactions in the block. - type Body: BlockBody; + type Body: Send + Sync + Unpin + 'static; - /// A block and block hash. - type SealedBlock; - - /// A block and addresses of senders of transactions in it. - type BlockWithSenders; - - /// Returns reference to [`BlockHeader`] type. + /// Returns reference to block header. fn header(&self) -> &Self::Header; - /// Returns reference to [`BlockBody`] type. + /// Returns reference to block body. fn body(&self) -> &Self::Body; - - /// Calculate the header hash and seal the block so that it can't be changed. - // todo: can be default impl if sealed block type is made generic over header and body and - // migrated to alloy - fn seal_slow(self) -> Self::SealedBlock; - - /// Seal the block with a known hash. - /// - /// WARNING: This method does not perform validation whether the hash is correct. - // todo: can be default impl if sealed block type is made generic over header and body and - // migrated to alloy - fn seal(self, hash: B256) -> Self::SealedBlock; - - /// Expensive operation that recovers transaction signer. See - /// `SealedBlockWithSenders`. - fn senders(&self) -> Option> { - self.body().recover_signers() - } - - /// Transform into a `BlockWithSenders`. - /// - /// # Panics - /// - /// If the number of senders does not match the number of transactions in the block - /// and the signer recovery for one of the transactions fails. - /// - /// Note: this is expected to be called with blocks read from disk. - #[track_caller] - fn with_senders_unchecked(self, senders: Vec
) -> Self::BlockWithSenders { - self.try_with_senders_unchecked(senders).expect("stored block is valid") - } - - /// Transform into a `BlockWithSenders` using the given senders. - /// - /// If the number of senders does not match the number of transactions in the block, this falls - /// back to manually recovery, but _without ensuring that the signature has a low `s` value_. - /// See also `SignedTransaction::recover_signer_unchecked`. - /// - /// Returns an error if a signature is invalid. - // todo: can be default impl if block with senders type is made generic over block and migrated - // to alloy - #[track_caller] - fn try_with_senders_unchecked( - self, - senders: Vec
, - ) -> Result, Self>; - - /// **Expensive**. Transform into a `BlockWithSenders` by recovering senders in the contained - /// transactions. - /// - /// Returns `None` if a transaction is invalid. - // todo: can be default impl if sealed block type is made generic over header and body and - // migrated to alloy - fn with_recovered_senders(self) -> Option>; } diff --git a/crates/primitives/src/block.rs b/crates/primitives/src/block.rs index c0586ed6a..6743cab3d 100644 --- a/crates/primitives/src/block.rs +++ b/crates/primitives/src/block.rs @@ -87,6 +87,19 @@ impl Block { } } +impl reth_primitives_traits::Block for Block { + type Header = Header; + type Body = BlockBody; + + fn body(&self) -> &Self::Body { + &self.body + } + + fn header(&self) -> &Self::Header { + &self.header + } +} + impl InMemorySize for Block { /// Calculates a heuristic for the in-memory size of the [`Block`]. #[inline]