gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-05-30 04:37:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

add a unique integer id to Rpc requests (#6444)

Browse Source 
* add id to rpc requests

* rename rpc request and response types for more accurate meaning

* remove unrequired build_request function

* remove unirequired Request wrapper types and unify Outbound and Inbound Request

* add RequestId to NetworkMessage::SendResponse

,NetworkMessage::SendErrorResponse to be passed to Rpc::send_response
This commit is contained in:
João Oliveira
2024-10-01 02:36:17 +01:00
committed by GitHub
parent 5d1ff7c6f8
commit 82098e1ef7
20 changed files with 1327 additions and 1046 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
beacon_node/lighthouse_network/src/lib.rs
View File

@@ -122,6 +122,6 @@ pub use peer_manager::{
ConnectionDirection, PeerConnectionStatus, PeerInfo, PeerManager, SyncInfo, SyncStatus,
};
// pub use service::{load_private_key, Context, Libp2pEvent, Service, NETWORK_KEY_FILENAME};
pub use service::api_types::{PeerRequestId, Request, Response};
pub use service::api_types::{PeerRequestId, Response};
pub use service::utils::*;
pub use service::{Gossipsub, NetworkEvent};

14
beacon_node/lighthouse_network/src/peer_manager/mod.rs
View File

@@ -2,7 +2,7 @@
use crate::discovery::enr_ext::EnrExt;
use crate::discovery::peer_id_to_node_id;
use crate::rpc::{GoodbyeReason, MetaData, Protocol, RPCError, RPCResponseErrorCode};
use crate::rpc::{GoodbyeReason, MetaData, Protocol, RPCError, RpcErrorResponse};
use crate::service::TARGET_SUBNET_PEERS;
use crate::{error, metrics, Gossipsub, NetworkGlobals, PeerId, Subnet, SubnetDiscovery};
use delay_map::HashSetDelay;
@@ -526,8 +526,8 @@ impl<E: EthSpec> PeerManager<E> {
PeerAction::HighToleranceError
}
RPCError::ErrorResponse(code, _) => match code {
RPCResponseErrorCode::Unknown => PeerAction::HighToleranceError,
RPCResponseErrorCode::ResourceUnavailable => {
RpcErrorResponse::Unknown => PeerAction::HighToleranceError,
RpcErrorResponse::ResourceUnavailable => {
// Don't ban on this because we want to retry with a block by root request.
if matches!(
protocol,
@@ -558,9 +558,9 @@ impl<E: EthSpec> PeerManager<E> {
ConnectionDirection::Incoming => return,
}
}
RPCResponseErrorCode::ServerError => PeerAction::MidToleranceError,
RPCResponseErrorCode::InvalidRequest => PeerAction::LowToleranceError,
RPCResponseErrorCode::RateLimited => match protocol {
RpcErrorResponse::ServerError => PeerAction::MidToleranceError,
RpcErrorResponse::InvalidRequest => PeerAction::LowToleranceError,
RpcErrorResponse::RateLimited => match protocol {
Protocol::Ping => PeerAction::MidToleranceError,
Protocol::BlocksByRange => PeerAction::MidToleranceError,
Protocol::BlocksByRoot => PeerAction::MidToleranceError,
@@ -577,7 +577,7 @@ impl<E: EthSpec> PeerManager<E> {
Protocol::MetaData => PeerAction::LowToleranceError,
Protocol::Status => PeerAction::LowToleranceError,
},
RPCResponseErrorCode::BlobsNotFoundForBlock => PeerAction::LowToleranceError,
RpcErrorResponse::BlobsNotFoundForBlock => PeerAction::LowToleranceError,
},
RPCError::SSZDecodeError(_) => PeerAction::Fatal,
RPCError::UnsupportedProtocol => {

494
beacon_node/lighthouse_network/src/rpc/codec.rs
View File

File diff suppressed because it is too large Load Diff

59
beacon_node/lighthouse_network/src/rpc/handler.rs
View File

@@ -1,11 +1,12 @@
#![allow(clippy::type_complexity)]
#![allow(clippy::cognitive_complexity)]
use super::methods::{GoodbyeReason, RPCCodedResponse, RPCResponseErrorCode};
use super::methods::{GoodbyeReason, RpcErrorResponse, RpcResponse};
use super::outbound::OutboundRequestContainer;
use super::protocol::{InboundOutput, InboundRequest, Protocol, RPCError, RPCProtocol};
use super::{RPCReceived, RPCSend, ReqId};
use crate::rpc::outbound::{OutboundFramed, OutboundRequest};
use super::protocol::{InboundOutput, Protocol, RPCError, RPCProtocol, RequestType};
use super::RequestId;
use super::{RPCReceived, RPCSend, ReqId, Request};
use crate::rpc::outbound::OutboundFramed;
use crate::rpc::protocol::InboundFramed;
use fnv::FnvHashMap;
use futures::prelude::*;
@@ -95,7 +96,7 @@ where
events_out: SmallVec<[HandlerEvent<Id, E>; 4]>,
/// Queue of outbound substreams to open.
dial_queue: SmallVec<[(Id, OutboundRequest<E>); 4]>,
dial_queue: SmallVec<[(Id, RequestType<E>); 4]>,
/// Current number of concurrent outbound substreams being opened.
dial_negotiated: u32,
@@ -159,7 +160,7 @@ struct InboundInfo<E: EthSpec> {
/// State of the substream.
state: InboundState<E>,
/// Responses queued for sending.
pending_items: VecDeque<RPCCodedResponse<E>>,
pending_items: VecDeque<RpcResponse<E>>,
/// Protocol of the original request we received from the peer.
protocol: Protocol,
/// Responses that the peer is still expecting from us.
@@ -205,7 +206,7 @@ pub enum OutboundSubstreamState<E: EthSpec> {
/// The framed negotiated substream.
substream: Box<OutboundFramed<Stream, E>>,
/// Keeps track of the actual request sent.
request: OutboundRequest<E>,
request: RequestType<E>,
},
/// Closing an outbound substream>
Closing(Box<OutboundFramed<Stream, E>>),
@@ -263,7 +264,7 @@ where
// Queue our goodbye message.
if let Some((id, reason)) = goodbye_reason {
self.dial_queue.push((id, OutboundRequest::Goodbye(reason)));
self.dial_queue.push((id, RequestType::Goodbye(reason)));
}
self.state = HandlerState::ShuttingDown(Box::pin(sleep(Duration::from_secs(
@@ -273,7 +274,7 @@ where
}
/// Opens an outbound substream with a request.
fn send_request(&mut self, id: Id, req: OutboundRequest<E>) {
fn send_request(&mut self, id: Id, req: RequestType<E>) {
match self.state {
HandlerState::Active => {
self.dial_queue.push((id, req));
@@ -291,10 +292,10 @@ where
/// Sends a response to a peer's request.
// NOTE: If the substream has closed due to inactivity, or the substream is in the
// wrong state a response will fail silently.
fn send_response(&mut self, inbound_id: SubstreamId, response: RPCCodedResponse<E>) {
fn send_response(&mut self, inbound_id: SubstreamId, response: RpcResponse<E>) {
// check if the stream matching the response still exists
let Some(inbound_info) = self.inbound_substreams.get_mut(&inbound_id) else {
if !matches!(response, RPCCodedResponse::StreamTermination(..)) {
if !matches!(response, RpcResponse::StreamTermination(..)) {
// the stream is closed after sending the expected number of responses
trace!(self.log, "Inbound stream has expired. Response not sent";
"response" => %response, "id" => inbound_id);
@@ -302,7 +303,7 @@ where
return;
};
// If the response we are sending is an error, report back for handling
if let RPCCodedResponse::Error(ref code, ref reason) = response {
if let RpcResponse::Error(ref code, ref reason) = response {
self.events_out.push(HandlerEvent::Err(HandlerErr::Inbound {
error: RPCError::ErrorResponse(*code, reason.to_string()),
proto: inbound_info.protocol,
@@ -329,7 +330,7 @@ where
type ToBehaviour = HandlerEvent<Id, E>;
type InboundProtocol = RPCProtocol<E>;
type OutboundProtocol = OutboundRequestContainer<E>;
type OutboundOpenInfo = (Id, OutboundRequest<E>); // Keep track of the id and the request
type OutboundOpenInfo = (Id, RequestType<E>); // Keep track of the id and the request
type InboundOpenInfo = ();
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, ()> {
@@ -403,8 +404,8 @@ where
if info.pending_items.back().map(|l| l.close_after()) == Some(false) {
// if the last chunk does not close the stream, append an error
info.pending_items.push_back(RPCCodedResponse::Error(
RPCResponseErrorCode::ServerError,
info.pending_items.push_back(RpcResponse::Error(
RpcErrorResponse::ServerError,
"Request timed out".into(),
));
}
@@ -672,13 +673,13 @@ where
let proto = entry.get().proto;
let received = match response {
RPCCodedResponse::StreamTermination(t) => {
RpcResponse::StreamTermination(t) => {
HandlerEvent::Ok(RPCReceived::EndOfStream(id, t))
}
RPCCodedResponse::Success(resp) => {
RpcResponse::Success(resp) => {
HandlerEvent::Ok(RPCReceived::Response(id, resp))
}
RPCCodedResponse::Error(ref code, ref r) => {
RpcResponse::Error(ref code, ref r) => {
HandlerEvent::Err(HandlerErr::Outbound {
id,
proto,
@@ -888,21 +889,23 @@ where
}
// If we received a goodbye, shutdown the connection.
if let InboundRequest::Goodbye(_) = req {
if let RequestType::Goodbye(_) = req {
self.shutdown(None);
}
self.events_out.push(HandlerEvent::Ok(RPCReceived::Request(
self.current_inbound_substream_id,
req,
)));
self.events_out
.push(HandlerEvent::Ok(RPCReceived::Request(Request {
id: RequestId::next(),
substream_id: self.current_inbound_substream_id,
r#type: req,
})));
self.current_inbound_substream_id.0 += 1;
}
fn on_fully_negotiated_outbound(
&mut self,
substream: OutboundFramed<Stream, E>,
(id, request): (Id, OutboundRequest<E>),
(id, request): (Id, RequestType<E>),
) {
self.dial_negotiated -= 1;
// Reset any io-retries counter.
@@ -958,7 +961,7 @@ where
}
fn on_dial_upgrade_error(
&mut self,
request_info: (Id, OutboundRequest<E>),
request_info: (Id, RequestType<E>),
error: StreamUpgradeError<RPCError>,
) {
let (id, req) = request_info;
@@ -1016,15 +1019,15 @@ impl slog::Value for SubstreamId {
/// error that occurred with sending a message is reported also.
async fn send_message_to_inbound_substream<E: EthSpec>(
mut substream: InboundSubstream<E>,
message: RPCCodedResponse<E>,
message: RpcResponse<E>,
last_chunk: bool,
) -> Result<(InboundSubstream<E>, bool), RPCError> {
if matches!(message, RPCCodedResponse::StreamTermination(_)) {
if matches!(message, RpcResponse::StreamTermination(_)) {
substream.close().await.map(|_| (substream, true))
} else {
// chunks that are not stream terminations get sent, and the stream is closed if
// the response is an error
let is_error = matches!(message, RPCCodedResponse::Error(..));
let is_error = matches!(message, RpcResponse::Error(..));
let send_result = substream.send(message).await;

126
beacon_node/lighthouse_network/src/rpc/methods.rs
View File

@@ -481,7 +481,7 @@ impl DataColumnsByRootRequest {
// Collection of enums and structs used by the Codecs to encode/decode RPC messages
#[derive(Debug, Clone, PartialEq)]
pub enum RPCResponse<E: EthSpec> {
pub enum RpcSuccessResponse<E: EthSpec> {
/// A HELLO message.
Status(StatusMessage),
@@ -545,11 +545,11 @@ pub enum ResponseTermination {
/// The structured response containing a result/code indicating success or failure
/// and the contents of the response
#[derive(Debug, Clone)]
pub enum RPCCodedResponse<E: EthSpec> {
pub enum RpcResponse<E: EthSpec> {
/// The response is a successful.
Success(RPCResponse<E>),
Success(RpcSuccessResponse<E>),
Error(RPCResponseErrorCode, ErrorType),
Error(RpcErrorResponse, ErrorType),
/// Received a stream termination indicating which response is being terminated.
StreamTermination(ResponseTermination),
@@ -564,7 +564,7 @@ pub struct LightClientBootstrapRequest {
/// The code assigned to an erroneous `RPCResponse`.
#[derive(Debug, Clone, Copy, PartialEq, IntoStaticStr)]
#[strum(serialize_all = "snake_case")]
pub enum RPCResponseErrorCode {
pub enum RpcErrorResponse {
RateLimited,
BlobsNotFoundForBlock,
InvalidRequest,
@@ -574,13 +574,13 @@ pub enum RPCResponseErrorCode {
Unknown,
}
impl<E: EthSpec> RPCCodedResponse<E> {
impl<E: EthSpec> RpcResponse<E> {
/// Used to encode the response in the codec.
pub fn as_u8(&self) -> Option<u8> {
match self {
RPCCodedResponse::Success(_) => Some(0),
RPCCodedResponse::Error(code, _) => Some(code.as_u8()),
RPCCodedResponse::StreamTermination(_) => None,
RpcResponse::Success(_) => Some(0),
RpcResponse::Error(code, _) => Some(code.as_u8()),
RpcResponse::StreamTermination(_) => None,
}
}
@@ -592,64 +592,66 @@ impl<E: EthSpec> RPCCodedResponse<E> {
/// Builds an RPCCodedResponse from a response code and an ErrorMessage
pub fn from_error(response_code: u8, err: ErrorType) -> Self {
let code = match response_code {
1 => RPCResponseErrorCode::InvalidRequest,
2 => RPCResponseErrorCode::ServerError,
3 => RPCResponseErrorCode::ResourceUnavailable,
139 => RPCResponseErrorCode::RateLimited,
140 => RPCResponseErrorCode::BlobsNotFoundForBlock,
_ => RPCResponseErrorCode::Unknown,
1 => RpcErrorResponse::InvalidRequest,
2 => RpcErrorResponse::ServerError,
3 => RpcErrorResponse::ResourceUnavailable,
139 => RpcErrorResponse::RateLimited,
140 => RpcErrorResponse::BlobsNotFoundForBlock,
_ => RpcErrorResponse::Unknown,
};
RPCCodedResponse::Error(code, err)
RpcResponse::Error(code, err)
}
/// Returns true if this response always terminates the stream.
pub fn close_after(&self) -> bool {
!matches!(self, RPCCodedResponse::Success(_))
!matches!(self, RpcResponse::Success(_))
}
}
impl RPCResponseErrorCode {
impl RpcErrorResponse {
fn as_u8(&self) -> u8 {
match self {
RPCResponseErrorCode::InvalidRequest => 1,
RPCResponseErrorCode::ServerError => 2,
RPCResponseErrorCode::ResourceUnavailable => 3,
RPCResponseErrorCode::Unknown => 255,
RPCResponseErrorCode::RateLimited => 139,
RPCResponseErrorCode::BlobsNotFoundForBlock => 140,
RpcErrorResponse::InvalidRequest => 1,
RpcErrorResponse::ServerError => 2,
RpcErrorResponse::ResourceUnavailable => 3,
RpcErrorResponse::Unknown => 255,
RpcErrorResponse::RateLimited => 139,
RpcErrorResponse::BlobsNotFoundForBlock => 140,
}
}
}
use super::Protocol;
impl<E: EthSpec> RPCResponse<E> {
impl<E: EthSpec> RpcSuccessResponse<E> {
pub fn protocol(&self) -> Protocol {
match self {
RPCResponse::Status(_) => Protocol::Status,
RPCResponse::BlocksByRange(_) => Protocol::BlocksByRange,
RPCResponse::BlocksByRoot(_) => Protocol::BlocksByRoot,
RPCResponse::BlobsByRange(_) => Protocol::BlobsByRange,
RPCResponse::BlobsByRoot(_) => Protocol::BlobsByRoot,
RPCResponse::DataColumnsByRoot(_) => Protocol::DataColumnsByRoot,
RPCResponse::DataColumnsByRange(_) => Protocol::DataColumnsByRange,
RPCResponse::Pong(_) => Protocol::Ping,
RPCResponse::MetaData(_) => Protocol::MetaData,
RPCResponse::LightClientBootstrap(_) => Protocol::LightClientBootstrap,
RPCResponse::LightClientOptimisticUpdate(_) => Protocol::LightClientOptimisticUpdate,
RPCResponse::LightClientFinalityUpdate(_) => Protocol::LightClientFinalityUpdate,
RpcSuccessResponse::Status(_) => Protocol::Status,
RpcSuccessResponse::BlocksByRange(_) => Protocol::BlocksByRange,
RpcSuccessResponse::BlocksByRoot(_) => Protocol::BlocksByRoot,
RpcSuccessResponse::BlobsByRange(_) => Protocol::BlobsByRange,
RpcSuccessResponse::BlobsByRoot(_) => Protocol::BlobsByRoot,
RpcSuccessResponse::DataColumnsByRoot(_) => Protocol::DataColumnsByRoot,
RpcSuccessResponse::DataColumnsByRange(_) => Protocol::DataColumnsByRange,
RpcSuccessResponse::Pong(_) => Protocol::Ping,
RpcSuccessResponse::MetaData(_) => Protocol::MetaData,
RpcSuccessResponse::LightClientBootstrap(_) => Protocol::LightClientBootstrap,
RpcSuccessResponse::LightClientOptimisticUpdate(_) => {
Protocol::LightClientOptimisticUpdate
}
RpcSuccessResponse::LightClientFinalityUpdate(_) => Protocol::LightClientFinalityUpdate,
}
}
}
impl std::fmt::Display for RPCResponseErrorCode {
impl std::fmt::Display for RpcErrorResponse {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let repr = match self {
RPCResponseErrorCode::InvalidRequest => "The request was invalid",
RPCResponseErrorCode::ResourceUnavailable => "Resource unavailable",
RPCResponseErrorCode::ServerError => "Server error occurred",
RPCResponseErrorCode::Unknown => "Unknown error occurred",
RPCResponseErrorCode::RateLimited => "Rate limited",
RPCResponseErrorCode::BlobsNotFoundForBlock => "No blobs for the given root",
RpcErrorResponse::InvalidRequest => "The request was invalid",
RpcErrorResponse::ResourceUnavailable => "Resource unavailable",
RpcErrorResponse::ServerError => "Server error occurred",
RpcErrorResponse::Unknown => "Unknown error occurred",
RpcErrorResponse::RateLimited => "Rate limited",
RpcErrorResponse::BlobsNotFoundForBlock => "No blobs for the given root",
};
f.write_str(repr)
}
@@ -661,45 +663,47 @@ impl std::fmt::Display for StatusMessage {
}
}
impl<E: EthSpec> std::fmt::Display for RPCResponse<E> {
impl<E: EthSpec> std::fmt::Display for RpcSuccessResponse<E> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RPCResponse::Status(status) => write!(f, "{}", status),
RPCResponse::BlocksByRange(block) => {
RpcSuccessResponse::Status(status) => write!(f, "{}", status),
RpcSuccessResponse::BlocksByRange(block) => {
write!(f, "BlocksByRange: Block slot: {}", block.slot())
}
RPCResponse::BlocksByRoot(block) => {
RpcSuccessResponse::BlocksByRoot(block) => {
write!(f, "BlocksByRoot: Block slot: {}", block.slot())
}
RPCResponse::BlobsByRange(blob) => {
RpcSuccessResponse::BlobsByRange(blob) => {
write!(f, "BlobsByRange: Blob slot: {}", blob.slot())
}
RPCResponse::BlobsByRoot(sidecar) => {
RpcSuccessResponse::BlobsByRoot(sidecar) => {
write!(f, "BlobsByRoot: Blob slot: {}", sidecar.slot())
}
RPCResponse::DataColumnsByRoot(sidecar) => {
RpcSuccessResponse::DataColumnsByRoot(sidecar) => {
write!(f, "DataColumnsByRoot: Data column slot: {}", sidecar.slot())
}
RPCResponse::DataColumnsByRange(sidecar) => {
RpcSuccessResponse::DataColumnsByRange(sidecar) => {
write!(
f,
"DataColumnsByRange: Data column slot: {}",
sidecar.slot()
)
}
RPCResponse::Pong(ping) => write!(f, "Pong: {}", ping.data),
RPCResponse::MetaData(metadata) => write!(f, "Metadata: {}", metadata.seq_number()),
RPCResponse::LightClientBootstrap(bootstrap) => {
RpcSuccessResponse::Pong(ping) => write!(f, "Pong: {}", ping.data),
RpcSuccessResponse::MetaData(metadata) => {
write!(f, "Metadata: {}", metadata.seq_number())
}
RpcSuccessResponse::LightClientBootstrap(bootstrap) => {
write!(f, "LightClientBootstrap Slot: {}", bootstrap.get_slot())
}
RPCResponse::LightClientOptimisticUpdate(update) => {
RpcSuccessResponse::LightClientOptimisticUpdate(update) => {
write!(
f,
"LightClientOptimisticUpdate Slot: {}",
update.signature_slot()
)
}
RPCResponse::LightClientFinalityUpdate(update) => {
RpcSuccessResponse::LightClientFinalityUpdate(update) => {
write!(
f,
"LightClientFinalityUpdate Slot: {}",
@@ -710,12 +714,12 @@ impl<E: EthSpec> std::fmt::Display for RPCResponse<E> {
}
}
impl<E: EthSpec> std::fmt::Display for RPCCodedResponse<E> {
impl<E: EthSpec> std::fmt::Display for RpcResponse<E> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RPCCodedResponse::Success(res) => write!(f, "{}", res),
RPCCodedResponse::Error(code, err) => write!(f, "{}: {}", code, err),
RPCCodedResponse::StreamTermination(_) => write!(f, "Stream Termination"),
RpcResponse::Success(res) => write!(f, "{}", res),
RpcResponse::Error(code, err) => write!(f, "{}: {}", code, err),
RpcResponse::StreamTermination(_) => write!(f, "Stream Termination"),
}
}
}

102
beacon_node/lighthouse_network/src/rpc/mod.rs
View File

@@ -16,6 +16,7 @@ use libp2p::PeerId;
use rate_limiter::{RPCRateLimiter as RateLimiter, RateLimitedErr};
use slog::{crit, debug, o, trace};
use std::marker::PhantomData;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
@@ -23,16 +24,15 @@ use types::{EthSpec, ForkContext};
pub(crate) use handler::{HandlerErr, HandlerEvent};
pub(crate) use methods::{
MetaData, MetaDataV1, MetaDataV2, MetaDataV3, Ping, RPCCodedResponse, RPCResponse,
MetaData, MetaDataV1, MetaDataV2, MetaDataV3, Ping, RpcResponse, RpcSuccessResponse,
};
pub(crate) use protocol::InboundRequest;
pub use protocol::RequestType;
pub use handler::SubstreamId;
pub use methods::{
BlocksByRangeRequest, BlocksByRootRequest, GoodbyeReason, LightClientBootstrapRequest,
RPCResponseErrorCode, ResponseTermination, StatusMessage,
ResponseTermination, RpcErrorResponse, StatusMessage,
};
pub(crate) use outbound::OutboundRequest;
pub use protocol::{max_rpc_size, Protocol, RPCError};
use self::config::{InboundRateLimiterConfig, OutboundRateLimiterConfig};
@@ -48,6 +48,8 @@ mod protocol;
mod rate_limiter;
mod self_limiter;
static NEXT_REQUEST_ID: AtomicUsize = AtomicUsize::new(1);
/// Composite trait for a request id.
pub trait ReqId: Send + 'static + std::fmt::Debug + Copy + Clone {}
impl<T> ReqId for T where T: Send + 'static + std::fmt::Debug + Copy + Clone {}
@@ -59,13 +61,13 @@ pub enum RPCSend<Id, E: EthSpec> {
///
/// The `Id` is given by the application making the request. These
/// go over *outbound* connections.
Request(Id, OutboundRequest<E>),
Request(Id, RequestType<E>),
/// A response sent from Lighthouse.
///
/// The `SubstreamId` must correspond to the RPC-given ID of the original request received from the
/// peer. The second parameter is a single chunk of a response. These go over *inbound*
/// connections.
Response(SubstreamId, RPCCodedResponse<E>),
Response(SubstreamId, RpcResponse<E>),
/// Lighthouse has requested to terminate the connection with a goodbye message.
Shutdown(Id, GoodbyeReason),
}
@@ -77,17 +79,46 @@ pub enum RPCReceived<Id, E: EthSpec> {
///
/// The `SubstreamId` is given by the `RPCHandler` as it identifies this request with the
/// *inbound* substream over which it is managed.
Request(SubstreamId, InboundRequest<E>),
Request(Request<E>),
/// A response received from the outside.
///
/// The `Id` corresponds to the application given ID of the original request sent to the
/// peer. The second parameter is a single chunk of a response. These go over *outbound*
/// connections.
Response(Id, RPCResponse<E>),
Response(Id, RpcSuccessResponse<E>),
/// Marks a request as completed
EndOfStream(Id, ResponseTermination),
}
/// Rpc `Request` identifier.
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
pub struct RequestId(usize);
impl RequestId {
/// Returns the next available [`RequestId`].
pub fn next() -> Self {
Self(NEXT_REQUEST_ID.fetch_add(1, Ordering::SeqCst))
}
/// Creates an _unchecked_ [`RequestId`].
///
/// [`Rpc`] enforces that [`RequestId`]s are unique and not reused.
/// This constructor does not, hence the _unchecked_.
///
/// It is primarily meant for allowing manual tests.
pub fn new_unchecked(id: usize) -> Self {
Self(id)
}
}
/// An Rpc Request.
#[derive(Debug, Clone)]
pub struct Request<E: EthSpec> {
pub id: RequestId,
pub substream_id: SubstreamId,
pub r#type: RequestType<E>,
}
impl<E: EthSpec, Id: std::fmt::Debug> std::fmt::Display for RPCSend<Id, E> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
@@ -177,7 +208,8 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
&mut self,
peer_id: PeerId,
id: (ConnectionId, SubstreamId),
event: RPCCodedResponse<E>,
_request_id: RequestId,
event: RpcResponse<E>,
) {
self.events.push(ToSwarm::NotifyHandler {
peer_id,
@@ -189,7 +221,7 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
/// Submits an RPC request.
///
/// The peer must be connected for this to succeed.
pub fn send_request(&mut self, peer_id: PeerId, request_id: Id, req: OutboundRequest<E>) {
pub fn send_request(&mut self, peer_id: PeerId, request_id: Id, req: RequestType<E>) {
let event = if let Some(self_limiter) = self.self_limiter.as_mut() {
match self_limiter.allows(peer_id, request_id, req) {
Ok(event) => event,
@@ -229,7 +261,7 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
data: self.seq_number,
};
trace!(self.log, "Sending Ping"; "peer_id" => %peer_id);
self.send_request(peer_id, id, OutboundRequest::Ping(ping));
self.send_request(peer_id, id, RequestType::Ping(ping));
}
}
@@ -368,13 +400,17 @@ where
event: <Self::ConnectionHandler as ConnectionHandler>::ToBehaviour,
) {
match event {
HandlerEvent::Ok(RPCReceived::Request(id, req)) => {
HandlerEvent::Ok(RPCReceived::Request(Request {
id,
substream_id,
r#type,
})) => {
if let Some(limiter) = self.limiter.as_mut() {
// check if the request is conformant to the quota
match limiter.allows(&peer_id, &req) {
match limiter.allows(&peer_id, &r#type) {
Err(RateLimitedErr::TooLarge) => {
// we set the batch sizes, so this is a coding/config err for most protocols
let protocol = req.versioned_protocol().protocol();
let protocol = r#type.versioned_protocol().protocol();
if matches!(
protocol,
Protocol::BlocksByRange
@@ -384,7 +420,7 @@ where
| Protocol::BlobsByRoot
| Protocol::DataColumnsByRoot
) {
debug!(self.log, "Request too large to process"; "request" => %req, "protocol" => %protocol);
debug!(self.log, "Request too large to process"; "request" => %r#type, "protocol" => %protocol);
} else {
// Other protocols shouldn't be sending large messages, we should flag the peer kind
crit!(self.log, "Request size too large to ever be processed"; "protocol" => %protocol);
@@ -393,9 +429,10 @@ where
// the handler upon receiving the error code will send it back to the behaviour
self.send_response(
peer_id,
(conn_id, id),
RPCCodedResponse::Error(
RPCResponseErrorCode::RateLimited,
(conn_id, substream_id),
id,
RpcResponse::Error(
RpcErrorResponse::RateLimited,
"Rate limited. Request too large".into(),
),
);
@@ -403,30 +440,33 @@ where
}
Err(RateLimitedErr::TooSoon(wait_time)) => {
debug!(self.log, "Request exceeds the rate limit";
"request" => %req, "peer_id" => %peer_id, "wait_time_ms" => wait_time.as_millis());
"request" => %r#type, "peer_id" => %peer_id, "wait_time_ms" => wait_time.as_millis());
// send an error code to the peer.
// the handler upon receiving the error code will send it back to the behaviour
self.send_response(
peer_id,
(conn_id, id),
RPCCodedResponse::Error(
RPCResponseErrorCode::RateLimited,
(conn_id, substream_id),
id,
RpcResponse::Error(
RpcErrorResponse::RateLimited,
format!("Wait {:?}", wait_time).into(),
),
);
return;
}
// No rate limiting, continue.
Ok(_) => {}
Ok(()) => {}
}
}
// If we received a Ping, we queue a Pong response.
if let InboundRequest::Ping(_) = req {
if let RequestType::Ping(_) = r#type {
trace!(self.log, "Received Ping, queueing Pong";"connection_id" => %conn_id, "peer_id" => %peer_id);
self.send_response(
peer_id,
(conn_id, id),
RPCCodedResponse::Success(RPCResponse::Pong(Ping {
(conn_id, substream_id),
id,
RpcResponse::Success(RpcSuccessResponse::Pong(Ping {
data: self.seq_number,
})),
);
@@ -435,7 +475,11 @@ where
self.events.push(ToSwarm::GenerateEvent(RPCMessage {
peer_id,
conn_id,
message: Ok(RPCReceived::Request(id, req)),
message: Ok(RPCReceived::Request(Request {
id,
substream_id,
r#type,
})),
}));
}
HandlerEvent::Ok(rpc) => {
@@ -496,8 +540,8 @@ where
match &self.message {
Ok(received) => {
let (msg_kind, protocol) = match received {
RPCReceived::Request(_, req) => {
("request", req.versioned_protocol().protocol())
RPCReceived::Request(Request { r#type, .. }) => {
("request", r#type.versioned_protocol().protocol())
}
RPCReceived::Response(_, res) => ("response", res.protocol()),
RPCReceived::EndOfStream(_, end) => (

165
beacon_node/lighthouse_network/src/rpc/outbound.rs
View File

@@ -1,7 +1,6 @@
use super::methods::*;
use super::protocol::ProtocolId;
use super::protocol::SupportedProtocol;
use super::RPCError;
use super::RequestType;
use crate::rpc::codec::SSZSnappyOutboundCodec;
use crate::rpc::protocol::Encoding;
use futures::future::BoxFuture;
@@ -21,25 +20,11 @@ use types::{EthSpec, ForkContext};
#[derive(Debug, Clone)]
pub struct OutboundRequestContainer<E: EthSpec> {
pub req: OutboundRequest<E>,
pub req: RequestType<E>,
pub fork_context: Arc<ForkContext>,
pub max_rpc_size: usize,
}
#[derive(Debug, Clone, PartialEq)]
pub enum OutboundRequest<E: EthSpec> {
Status(StatusMessage),
Goodbye(GoodbyeReason),
BlocksByRange(OldBlocksByRangeRequest),
BlocksByRoot(BlocksByRootRequest),
BlobsByRange(BlobsByRangeRequest),
BlobsByRoot(BlobsByRootRequest),
DataColumnsByRoot(DataColumnsByRootRequest),
DataColumnsByRange(DataColumnsByRangeRequest),
Ping(Ping),
MetaData(MetadataRequest<E>),
}
impl<E: EthSpec> UpgradeInfo for OutboundRequestContainer<E> {
type Info = ProtocolId;
type InfoIter = Vec<Self::Info>;
@@ -50,133 +35,6 @@ impl<E: EthSpec> UpgradeInfo for OutboundRequestContainer<E> {
}
}
/// Implements the encoding per supported protocol for `RPCRequest`.
impl<E: EthSpec> OutboundRequest<E> {
pub fn supported_protocols(&self) -> Vec<ProtocolId> {
match self {
// add more protocols when versions/encodings are supported
OutboundRequest::Status(_) => vec![ProtocolId::new(
SupportedProtocol::StatusV1,
Encoding::SSZSnappy,
)],
OutboundRequest::Goodbye(_) => vec![ProtocolId::new(
SupportedProtocol::GoodbyeV1,
Encoding::SSZSnappy,
)],
OutboundRequest::BlocksByRange(_) => vec![
ProtocolId::new(SupportedProtocol::BlocksByRangeV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlocksByRangeV1, Encoding::SSZSnappy),
],
OutboundRequest::BlocksByRoot(_) => vec![
ProtocolId::new(SupportedProtocol::BlocksByRootV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlocksByRootV1, Encoding::SSZSnappy),
],
OutboundRequest::BlobsByRange(_) => vec![ProtocolId::new(
SupportedProtocol::BlobsByRangeV1,
Encoding::SSZSnappy,
)],
OutboundRequest::BlobsByRoot(_) => vec![ProtocolId::new(
SupportedProtocol::BlobsByRootV1,
Encoding::SSZSnappy,
)],
OutboundRequest::DataColumnsByRoot(_) => vec![ProtocolId::new(
SupportedProtocol::DataColumnsByRootV1,
Encoding::SSZSnappy,
)],
OutboundRequest::DataColumnsByRange(_) => vec![ProtocolId::new(
SupportedProtocol::DataColumnsByRangeV1,
Encoding::SSZSnappy,
)],
OutboundRequest::Ping(_) => vec![ProtocolId::new(
SupportedProtocol::PingV1,
Encoding::SSZSnappy,
)],
OutboundRequest::MetaData(_) => vec![
ProtocolId::new(SupportedProtocol::MetaDataV3, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::MetaDataV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::MetaDataV1, Encoding::SSZSnappy),
],
}
}
/* These functions are used in the handler for stream management */
/// Maximum number of responses expected for this request.
pub fn max_responses(&self) -> u64 {
match self {
OutboundRequest::Status(_) => 1,
OutboundRequest::Goodbye(_) => 0,
OutboundRequest::BlocksByRange(req) => *req.count(),
OutboundRequest::BlocksByRoot(req) => req.block_roots().len() as u64,
OutboundRequest::BlobsByRange(req) => req.max_blobs_requested::<E>(),
OutboundRequest::BlobsByRoot(req) => req.blob_ids.len() as u64,
OutboundRequest::DataColumnsByRoot(req) => req.data_column_ids.len() as u64,
OutboundRequest::DataColumnsByRange(req) => req.max_requested::<E>(),
OutboundRequest::Ping(_) => 1,
OutboundRequest::MetaData(_) => 1,
}
}
pub fn expect_exactly_one_response(&self) -> bool {
match self {
OutboundRequest::Status(_) => true,
OutboundRequest::Goodbye(_) => false,
OutboundRequest::BlocksByRange(_) => false,
OutboundRequest::BlocksByRoot(_) => false,
OutboundRequest::BlobsByRange(_) => false,
OutboundRequest::BlobsByRoot(_) => false,
OutboundRequest::DataColumnsByRoot(_) => false,
OutboundRequest::DataColumnsByRange(_) => false,
OutboundRequest::Ping(_) => true,
OutboundRequest::MetaData(_) => true,
}
}
/// Gives the corresponding `SupportedProtocol` to this request.
pub fn versioned_protocol(&self) -> SupportedProtocol {
match self {
OutboundRequest::Status(_) => SupportedProtocol::StatusV1,
OutboundRequest::Goodbye(_) => SupportedProtocol::GoodbyeV1,
OutboundRequest::BlocksByRange(req) => match req {
OldBlocksByRangeRequest::V1(_) => SupportedProtocol::BlocksByRangeV1,
OldBlocksByRangeRequest::V2(_) => SupportedProtocol::BlocksByRangeV2,
},
OutboundRequest::BlocksByRoot(req) => match req {
BlocksByRootRequest::V1(_) => SupportedProtocol::BlocksByRootV1,
BlocksByRootRequest::V2(_) => SupportedProtocol::BlocksByRootV2,
},
OutboundRequest::BlobsByRange(_) => SupportedProtocol::BlobsByRangeV1,
OutboundRequest::BlobsByRoot(_) => SupportedProtocol::BlobsByRootV1,
OutboundRequest::DataColumnsByRoot(_) => SupportedProtocol::DataColumnsByRootV1,
OutboundRequest::DataColumnsByRange(_) => SupportedProtocol::DataColumnsByRangeV1,
OutboundRequest::Ping(_) => SupportedProtocol::PingV1,
OutboundRequest::MetaData(req) => match req {
MetadataRequest::V1(_) => SupportedProtocol::MetaDataV1,
MetadataRequest::V2(_) => SupportedProtocol::MetaDataV2,
MetadataRequest::V3(_) => SupportedProtocol::MetaDataV3,
},
}
}
/// Returns the `ResponseTermination` type associated with the request if a stream gets
/// terminated.
pub fn stream_termination(&self) -> ResponseTermination {
match self {
// this only gets called after `multiple_responses()` returns true. Therefore, only
// variants that have `multiple_responses()` can have values.
OutboundRequest::BlocksByRange(_) => ResponseTermination::BlocksByRange,
OutboundRequest::BlocksByRoot(_) => ResponseTermination::BlocksByRoot,
OutboundRequest::BlobsByRange(_) => ResponseTermination::BlobsByRange,
OutboundRequest::BlobsByRoot(_) => ResponseTermination::BlobsByRoot,
OutboundRequest::DataColumnsByRoot(_) => ResponseTermination::DataColumnsByRoot,
OutboundRequest::DataColumnsByRange(_) => ResponseTermination::DataColumnsByRange,
OutboundRequest::Status(_) => unreachable!(),
OutboundRequest::Goodbye(_) => unreachable!(),
OutboundRequest::Ping(_) => unreachable!(),
OutboundRequest::MetaData(_) => unreachable!(),
}
}
}
/* RPC Response type - used for outbound upgrades */
/* Outbound upgrades */
@@ -211,22 +69,3 @@ where
.boxed()
}
}
impl<E: EthSpec> std::fmt::Display for OutboundRequest<E> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
OutboundRequest::Status(status) => write!(f, "Status Message: {}", status),
OutboundRequest::Goodbye(reason) => write!(f, "Goodbye: {}", reason),
OutboundRequest::BlocksByRange(req) => write!(f, "Blocks by range: {}", req),
OutboundRequest::BlocksByRoot(req) => write!(f, "Blocks by root: {:?}", req),
OutboundRequest::BlobsByRange(req) => write!(f, "Blobs by range: {:?}", req),
OutboundRequest::BlobsByRoot(req) => write!(f, "Blobs by root: {:?}", req),
OutboundRequest::DataColumnsByRoot(req) => write!(f, "Data columns by root: {:?}", req),
OutboundRequest::DataColumnsByRange(req) => {
write!(f, "Data columns by range: {:?}", req)
}
OutboundRequest::Ping(ping) => write!(f, "Ping: {}", ping.data),
OutboundRequest::MetaData(_) => write!(f, "MetaData request"),
}
}
}

201
beacon_node/lighthouse_network/src/rpc/protocol.rs
View File

@@ -645,7 +645,7 @@ pub fn rpc_data_column_limits<E: EthSpec>() -> RpcLimits {
// The inbound protocol reads the request, decodes it and returns the stream to the protocol
// handler to respond to once ready.
pub type InboundOutput<TSocket, E> = (InboundRequest<E>, InboundFramed<TSocket, E>);
pub type InboundOutput<TSocket, E> = (RequestType<E>, InboundFramed<TSocket, E>);
pub type InboundFramed<TSocket, E> =
Framed<std::pin::Pin<Box<TimeoutStream<Compat<TSocket>>>>, SSZSnappyInboundCodec<E>>;
@@ -679,19 +679,19 @@ where
// MetaData requests should be empty, return the stream
match versioned_protocol {
SupportedProtocol::MetaDataV1 => {
Ok((InboundRequest::MetaData(MetadataRequest::new_v1()), socket))
Ok((RequestType::MetaData(MetadataRequest::new_v1()), socket))
}
SupportedProtocol::MetaDataV2 => {
Ok((InboundRequest::MetaData(MetadataRequest::new_v2()), socket))
Ok((RequestType::MetaData(MetadataRequest::new_v2()), socket))
}
SupportedProtocol::MetaDataV3 => {
Ok((InboundRequest::MetaData(MetadataRequest::new_v3()), socket))
Ok((RequestType::MetaData(MetadataRequest::new_v3()), socket))
}
SupportedProtocol::LightClientOptimisticUpdateV1 => {
Ok((InboundRequest::LightClientOptimisticUpdate, socket))
Ok((RequestType::LightClientOptimisticUpdate, socket))
}
SupportedProtocol::LightClientFinalityUpdateV1 => {
Ok((InboundRequest::LightClientFinalityUpdate, socket))
Ok((RequestType::LightClientFinalityUpdate, socket))
}
_ => {
match tokio::time::timeout(
@@ -713,7 +713,7 @@ where
}
#[derive(Debug, Clone, PartialEq)]
pub enum InboundRequest<E: EthSpec> {
pub enum RequestType<E: EthSpec> {
Status(StatusMessage),
Goodbye(GoodbyeReason),
BlocksByRange(OldBlocksByRangeRequest),
@@ -730,56 +730,56 @@ pub enum InboundRequest<E: EthSpec> {
}
/// Implements the encoding per supported protocol for `RPCRequest`.
impl<E: EthSpec> InboundRequest<E> {
impl<E: EthSpec> RequestType<E> {
/* These functions are used in the handler for stream management */
/// Maximum number of responses expected for this request.
pub fn max_responses(&self) -> u64 {
match self {
InboundRequest::Status(_) => 1,
InboundRequest::Goodbye(_) => 0,
InboundRequest::BlocksByRange(req) => *req.count(),
InboundRequest::BlocksByRoot(req) => req.block_roots().len() as u64,
InboundRequest::BlobsByRange(req) => req.max_blobs_requested::<E>(),
InboundRequest::BlobsByRoot(req) => req.blob_ids.len() as u64,
InboundRequest::DataColumnsByRoot(req) => req.data_column_ids.len() as u64,
InboundRequest::DataColumnsByRange(req) => req.max_requested::<E>(),
InboundRequest::Ping(_) => 1,
InboundRequest::MetaData(_) => 1,
InboundRequest::LightClientBootstrap(_) => 1,
InboundRequest::LightClientOptimisticUpdate => 1,
InboundRequest::LightClientFinalityUpdate => 1,
RequestType::Status(_) => 1,
RequestType::Goodbye(_) => 0,
RequestType::BlocksByRange(req) => *req.count(),
RequestType::BlocksByRoot(req) => req.block_roots().len() as u64,
RequestType::BlobsByRange(req) => req.max_blobs_requested::<E>(),
RequestType::BlobsByRoot(req) => req.blob_ids.len() as u64,
RequestType::DataColumnsByRoot(req) => req.data_column_ids.len() as u64,
RequestType::DataColumnsByRange(req) => req.max_requested::<E>(),
RequestType::Ping(_) => 1,
RequestType::MetaData(_) => 1,
RequestType::LightClientBootstrap(_) => 1,
RequestType::LightClientOptimisticUpdate => 1,
RequestType::LightClientFinalityUpdate => 1,
}
}
/// Gives the corresponding `SupportedProtocol` to this request.
pub fn versioned_protocol(&self) -> SupportedProtocol {
match self {
InboundRequest::Status(_) => SupportedProtocol::StatusV1,
InboundRequest::Goodbye(_) => SupportedProtocol::GoodbyeV1,
InboundRequest::BlocksByRange(req) => match req {
RequestType::Status(_) => SupportedProtocol::StatusV1,
RequestType::Goodbye(_) => SupportedProtocol::GoodbyeV1,
RequestType::BlocksByRange(req) => match req {
OldBlocksByRangeRequest::V1(_) => SupportedProtocol::BlocksByRangeV1,
OldBlocksByRangeRequest::V2(_) => SupportedProtocol::BlocksByRangeV2,
},
InboundRequest::BlocksByRoot(req) => match req {
RequestType::BlocksByRoot(req) => match req {
BlocksByRootRequest::V1(_) => SupportedProtocol::BlocksByRootV1,
BlocksByRootRequest::V2(_) => SupportedProtocol::BlocksByRootV2,
},
InboundRequest::BlobsByRange(_) => SupportedProtocol::BlobsByRangeV1,
InboundRequest::BlobsByRoot(_) => SupportedProtocol::BlobsByRootV1,
InboundRequest::DataColumnsByRoot(_) => SupportedProtocol::DataColumnsByRootV1,
InboundRequest::DataColumnsByRange(_) => SupportedProtocol::DataColumnsByRangeV1,
InboundRequest::Ping(_) => SupportedProtocol::PingV1,
InboundRequest::MetaData(req) => match req {
RequestType::BlobsByRange(_) => SupportedProtocol::BlobsByRangeV1,
RequestType::BlobsByRoot(_) => SupportedProtocol::BlobsByRootV1,
RequestType::DataColumnsByRoot(_) => SupportedProtocol::DataColumnsByRootV1,
RequestType::DataColumnsByRange(_) => SupportedProtocol::DataColumnsByRangeV1,
RequestType::Ping(_) => SupportedProtocol::PingV1,
RequestType::MetaData(req) => match req {
MetadataRequest::V1(_) => SupportedProtocol::MetaDataV1,
MetadataRequest::V2(_) => SupportedProtocol::MetaDataV2,
MetadataRequest::V3(_) => SupportedProtocol::MetaDataV3,
},
InboundRequest::LightClientBootstrap(_) => SupportedProtocol::LightClientBootstrapV1,
InboundRequest::LightClientOptimisticUpdate => {
RequestType::LightClientBootstrap(_) => SupportedProtocol::LightClientBootstrapV1,
RequestType::LightClientOptimisticUpdate => {
SupportedProtocol::LightClientOptimisticUpdateV1
}
InboundRequest::LightClientFinalityUpdate => {
RequestType::LightClientFinalityUpdate => {
SupportedProtocol::LightClientFinalityUpdateV1
}
}
@@ -791,19 +791,96 @@ impl<E: EthSpec> InboundRequest<E> {
match self {
// this only gets called after `multiple_responses()` returns true. Therefore, only
// variants that have `multiple_responses()` can have values.
InboundRequest::BlocksByRange(_) => ResponseTermination::BlocksByRange,
InboundRequest::BlocksByRoot(_) => ResponseTermination::BlocksByRoot,
InboundRequest::BlobsByRange(_) => ResponseTermination::BlobsByRange,
InboundRequest::BlobsByRoot(_) => ResponseTermination::BlobsByRoot,
InboundRequest::DataColumnsByRoot(_) => ResponseTermination::DataColumnsByRoot,
InboundRequest::DataColumnsByRange(_) => ResponseTermination::DataColumnsByRange,
InboundRequest::Status(_) => unreachable!(),
InboundRequest::Goodbye(_) => unreachable!(),
InboundRequest::Ping(_) => unreachable!(),
InboundRequest::MetaData(_) => unreachable!(),
InboundRequest::LightClientBootstrap(_) => unreachable!(),
InboundRequest::LightClientFinalityUpdate => unreachable!(),
InboundRequest::LightClientOptimisticUpdate => unreachable!(),
RequestType::BlocksByRange(_) => ResponseTermination::BlocksByRange,
RequestType::BlocksByRoot(_) => ResponseTermination::BlocksByRoot,
RequestType::BlobsByRange(_) => ResponseTermination::BlobsByRange,
RequestType::BlobsByRoot(_) => ResponseTermination::BlobsByRoot,
RequestType::DataColumnsByRoot(_) => ResponseTermination::DataColumnsByRoot,
RequestType::DataColumnsByRange(_) => ResponseTermination::DataColumnsByRange,
RequestType::Status(_) => unreachable!(),
RequestType::Goodbye(_) => unreachable!(),
RequestType::Ping(_) => unreachable!(),
RequestType::MetaData(_) => unreachable!(),
RequestType::LightClientBootstrap(_) => unreachable!(),
RequestType::LightClientFinalityUpdate => unreachable!(),
RequestType::LightClientOptimisticUpdate => unreachable!(),
}
}
pub fn supported_protocols(&self) -> Vec<ProtocolId> {
match self {
// add more protocols when versions/encodings are supported
RequestType::Status(_) => vec![ProtocolId::new(
SupportedProtocol::StatusV1,
Encoding::SSZSnappy,
)],
RequestType::Goodbye(_) => vec![ProtocolId::new(
SupportedProtocol::GoodbyeV1,
Encoding::SSZSnappy,
)],
RequestType::BlocksByRange(_) => vec![
ProtocolId::new(SupportedProtocol::BlocksByRangeV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlocksByRangeV1, Encoding::SSZSnappy),
],
RequestType::BlocksByRoot(_) => vec![
ProtocolId::new(SupportedProtocol::BlocksByRootV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlocksByRootV1, Encoding::SSZSnappy),
],
RequestType::BlobsByRange(_) => vec![ProtocolId::new(
SupportedProtocol::BlobsByRangeV1,
Encoding::SSZSnappy,
)],
RequestType::BlobsByRoot(_) => vec![ProtocolId::new(
SupportedProtocol::BlobsByRootV1,
Encoding::SSZSnappy,
)],
RequestType::DataColumnsByRoot(_) => vec![ProtocolId::new(
SupportedProtocol::DataColumnsByRootV1,
Encoding::SSZSnappy,
)],
RequestType::DataColumnsByRange(_) => vec![ProtocolId::new(
SupportedProtocol::DataColumnsByRangeV1,
Encoding::SSZSnappy,
)],
RequestType::Ping(_) => vec![ProtocolId::new(
SupportedProtocol::PingV1,
Encoding::SSZSnappy,
)],
RequestType::MetaData(_) => vec![
ProtocolId::new(SupportedProtocol::MetaDataV3, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::MetaDataV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::MetaDataV1, Encoding::SSZSnappy),
],
RequestType::LightClientBootstrap(_) => vec![ProtocolId::new(
SupportedProtocol::LightClientBootstrapV1,
Encoding::SSZSnappy,
)],
RequestType::LightClientOptimisticUpdate => vec![ProtocolId::new(
SupportedProtocol::LightClientOptimisticUpdateV1,
Encoding::SSZSnappy,
)],
RequestType::LightClientFinalityUpdate => vec![ProtocolId::new(
SupportedProtocol::LightClientFinalityUpdateV1,
Encoding::SSZSnappy,
)],
}
}
pub fn expect_exactly_one_response(&self) -> bool {
match self {
RequestType::Status(_) => true,
RequestType::Goodbye(_) => false,
RequestType::BlocksByRange(_) => false,
RequestType::BlocksByRoot(_) => false,
RequestType::BlobsByRange(_) => false,
RequestType::BlobsByRoot(_) => false,
RequestType::DataColumnsByRoot(_) => false,
RequestType::DataColumnsByRange(_) => false,
RequestType::Ping(_) => true,
RequestType::MetaData(_) => true,
RequestType::LightClientBootstrap(_) => true,
RequestType::LightClientOptimisticUpdate => true,
RequestType::LightClientFinalityUpdate => true,
}
}
}
@@ -819,7 +896,7 @@ pub enum RPCError {
/// IO Error.
IoError(String),
/// The peer returned a valid response but the response indicated an error.
ErrorResponse(RPCResponseErrorCode, String),
ErrorResponse(RpcErrorResponse, String),
/// Timed out waiting for a response.
StreamTimeout,
/// Peer does not support the protocol.
@@ -898,28 +975,28 @@ impl std::error::Error for RPCError {
}
}
impl<E: EthSpec> std::fmt::Display for InboundRequest<E> {
impl<E: EthSpec> std::fmt::Display for RequestType<E> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
InboundRequest::Status(status) => write!(f, "Status Message: {}", status),
InboundRequest::Goodbye(reason) => write!(f, "Goodbye: {}", reason),
InboundRequest::BlocksByRange(req) => write!(f, "Blocks by range: {}", req),
InboundRequest::BlocksByRoot(req) => write!(f, "Blocks by root: {:?}", req),
InboundRequest::BlobsByRange(req) => write!(f, "Blobs by range: {:?}", req),
InboundRequest::BlobsByRoot(req) => write!(f, "Blobs by root: {:?}", req),
InboundRequest::DataColumnsByRoot(req) => write!(f, "Data columns by root: {:?}", req),
InboundRequest::DataColumnsByRange(req) => {
RequestType::Status(status) => write!(f, "Status Message: {}", status),
RequestType::Goodbye(reason) => write!(f, "Goodbye: {}", reason),
RequestType::BlocksByRange(req) => write!(f, "Blocks by range: {}", req),
RequestType::BlocksByRoot(req) => write!(f, "Blocks by root: {:?}", req),
RequestType::BlobsByRange(req) => write!(f, "Blobs by range: {:?}", req),
RequestType::BlobsByRoot(req) => write!(f, "Blobs by root: {:?}", req),
RequestType::DataColumnsByRoot(req) => write!(f, "Data columns by root: {:?}", req),
RequestType::DataColumnsByRange(req) => {
write!(f, "Data columns by range: {:?}", req)
}
InboundRequest::Ping(ping) => write!(f, "Ping: {}", ping.data),
InboundRequest::MetaData(_) => write!(f, "MetaData request"),
InboundRequest::LightClientBootstrap(bootstrap) => {
RequestType::Ping(ping) => write!(f, "Ping: {}", ping.data),
RequestType::MetaData(_) => write!(f, "MetaData request"),
RequestType::LightClientBootstrap(bootstrap) => {
write!(f, "Light client boostrap: {}", bootstrap.root)
}
InboundRequest::LightClientOptimisticUpdate => {
RequestType::LightClientOptimisticUpdate => {
write!(f, "Light client optimistic update request")
}
InboundRequest::LightClientFinalityUpdate => {
RequestType::LightClientFinalityUpdate => {
write!(f, "Light client finality update request")
}
}

11
beacon_node/lighthouse_network/src/rpc/rate_limiter.rs
View File

@@ -252,7 +252,7 @@ pub trait RateLimiterItem {
fn max_responses(&self) -> u64;
}
impl<E: EthSpec> RateLimiterItem for super::InboundRequest<E> {
impl<E: EthSpec> RateLimiterItem for super::RequestType<E> {
fn protocol(&self) -> Protocol {
self.versioned_protocol().protocol()
}
@@ -262,15 +262,6 @@ impl<E: EthSpec> RateLimiterItem for super::InboundRequest<E> {
}
}
impl<E: EthSpec> RateLimiterItem for super::OutboundRequest<E> {
fn protocol(&self) -> Protocol {
self.versioned_protocol().protocol()
}
fn max_responses(&self) -> u64 {
self.max_responses()
}
}
impl RPCRateLimiter {
pub fn new_with_config(config: RateLimiterConfig) -> Result<Self, &'static str> {
// Destructure to make sure every configuration value is used.

12
beacon_node/lighthouse_network/src/rpc/self_limiter.rs
View File

@@ -14,13 +14,13 @@ use types::EthSpec;
use super::{
config::OutboundRateLimiterConfig,
rate_limiter::{RPCRateLimiter as RateLimiter, RateLimitedErr},
BehaviourAction, OutboundRequest, Protocol, RPCSend, ReqId,
BehaviourAction, Protocol, RPCSend, ReqId, RequestType,
};
/// A request that was rate limited or waiting on rate limited requests for the same peer and
/// protocol.
struct QueuedRequest<Id: ReqId, E: EthSpec> {
req: OutboundRequest<E>,
req: RequestType<E>,
request_id: Id,
}
@@ -70,7 +70,7 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
&mut self,
peer_id: PeerId,
request_id: Id,
req: OutboundRequest<E>,
req: RequestType<E>,
) -> Result<BehaviourAction<Id, E>, Error> {
let protocol = req.versioned_protocol().protocol();
// First check that there are not already other requests waiting to be sent.
@@ -101,7 +101,7 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
limiter: &mut RateLimiter,
peer_id: PeerId,
request_id: Id,
req: OutboundRequest<E>,
req: RequestType<E>,
log: &Logger,
) -> Result<BehaviourAction<Id, E>, (QueuedRequest<Id, E>, Duration)> {
match limiter.allows(&peer_id, &req) {
@@ -211,7 +211,7 @@ mod tests {
use crate::rpc::config::{OutboundRateLimiterConfig, RateLimiterConfig};
use crate::rpc::rate_limiter::Quota;
use crate::rpc::self_limiter::SelfRateLimiter;
use crate::rpc::{OutboundRequest, Ping, Protocol};
use crate::rpc::{Ping, Protocol, RequestType};
use crate::service::api_types::{AppRequestId, RequestId, SyncRequestId};
use libp2p::PeerId;
use std::time::Duration;
@@ -235,7 +235,7 @@ mod tests {
RequestId::Application(AppRequestId::Sync(SyncRequestId::RangeBlockAndBlobs {
id: i,
})),
OutboundRequest::Ping(Ping { data: i as u64 }),
RequestType::Ping(Ping { data: i as u64 }),
);
}

112
beacon_node/lighthouse_network/src/service/api_types.rs
View File

@@ -6,16 +6,9 @@ use types::{
LightClientFinalityUpdate, LightClientOptimisticUpdate, SignedBeaconBlock,
};
use crate::rpc::methods::{
BlobsByRangeRequest, BlobsByRootRequest, DataColumnsByRangeRequest, DataColumnsByRootRequest,
};
use crate::rpc::{
methods::{
BlocksByRangeRequest, BlocksByRootRequest, LightClientBootstrapRequest,
OldBlocksByRangeRequest, OldBlocksByRangeRequestV1, OldBlocksByRangeRequestV2,
RPCCodedResponse, RPCResponse, ResponseTermination, StatusMessage,
},
OutboundRequest, SubstreamId,
methods::{ResponseTermination, RpcResponse, RpcSuccessResponse, StatusMessage},
SubstreamId,
};
/// Identifier of requests sent by a peer.
@@ -93,69 +86,6 @@ pub enum RequestId {
Internal,
}
/// The type of RPC requests the Behaviour informs it has received and allows for sending.
///
// NOTE: This is an application-level wrapper over the lower network level requests that can be
// sent. The main difference is the absence of the Ping, Metadata and Goodbye protocols, which don't
// leave the Behaviour. For all protocols managed by RPC see `RPCRequest`.
#[derive(Debug, Clone, PartialEq)]
pub enum Request {
/// A Status message.
Status(StatusMessage),
/// A blocks by range request.
BlocksByRange(BlocksByRangeRequest),
/// A blobs by range request.
BlobsByRange(BlobsByRangeRequest),
/// A request blocks root request.
BlocksByRoot(BlocksByRootRequest),
// light client bootstrap request
LightClientBootstrap(LightClientBootstrapRequest),
// light client optimistic update request
LightClientOptimisticUpdate,
// light client finality update request
LightClientFinalityUpdate,
/// A request blobs root request.
BlobsByRoot(BlobsByRootRequest),
/// A request data columns root request.
DataColumnsByRoot(DataColumnsByRootRequest),
/// A request data columns by range request.
DataColumnsByRange(DataColumnsByRangeRequest),
}
impl<E: EthSpec> std::convert::From<Request> for OutboundRequest<E> {
fn from(req: Request) -> OutboundRequest<E> {
match req {
Request::BlocksByRoot(r) => OutboundRequest::BlocksByRoot(r),
Request::BlocksByRange(r) => match r {
BlocksByRangeRequest::V1(req) => OutboundRequest::BlocksByRange(
OldBlocksByRangeRequest::V1(OldBlocksByRangeRequestV1 {
start_slot: req.start_slot,
count: req.count,
step: 1,
}),
),
BlocksByRangeRequest::V2(req) => OutboundRequest::BlocksByRange(
OldBlocksByRangeRequest::V2(OldBlocksByRangeRequestV2 {
start_slot: req.start_slot,
count: req.count,
step: 1,
}),
),
},
Request::LightClientBootstrap(_)
| Request::LightClientOptimisticUpdate
| Request::LightClientFinalityUpdate => {
unreachable!("Lighthouse never makes an outbound light client request")
}
Request::BlobsByRange(r) => OutboundRequest::BlobsByRange(r),
Request::BlobsByRoot(r) => OutboundRequest::BlobsByRoot(r),
Request::DataColumnsByRoot(r) => OutboundRequest::DataColumnsByRoot(r),
Request::DataColumnsByRange(r) => OutboundRequest::DataColumnsByRange(r),
Request::Status(s) => OutboundRequest::Status(s),
}
}
}
/// The type of RPC responses the Behaviour informs it has received, and allows for sending.
///
// NOTE: This is an application-level wrapper over the lower network level responses that can be
@@ -186,44 +116,42 @@ pub enum Response<E: EthSpec> {
LightClientFinalityUpdate(Arc<LightClientFinalityUpdate<E>>),
}
impl<E: EthSpec> std::convert::From<Response<E>> for RPCCodedResponse<E> {
fn from(resp: Response<E>) -> RPCCodedResponse<E> {
impl<E: EthSpec> std::convert::From<Response<E>> for RpcResponse<E> {
fn from(resp: Response<E>) -> RpcResponse<E> {
match resp {
Response::BlocksByRoot(r) => match r {
Some(b) => RPCCodedResponse::Success(RPCResponse::BlocksByRoot(b)),
None => RPCCodedResponse::StreamTermination(ResponseTermination::BlocksByRoot),
Some(b) => RpcResponse::Success(RpcSuccessResponse::BlocksByRoot(b)),
None => RpcResponse::StreamTermination(ResponseTermination::BlocksByRoot),
},
Response::BlocksByRange(r) => match r {
Some(b) => RPCCodedResponse::Success(RPCResponse::BlocksByRange(b)),
None => RPCCodedResponse::StreamTermination(ResponseTermination::BlocksByRange),
Some(b) => RpcResponse::Success(RpcSuccessResponse::BlocksByRange(b)),
None => RpcResponse::StreamTermination(ResponseTermination::BlocksByRange),
},
Response::BlobsByRoot(r) => match r {
Some(b) => RPCCodedResponse::Success(RPCResponse::BlobsByRoot(b)),
None => RPCCodedResponse::StreamTermination(ResponseTermination::BlobsByRoot),
Some(b) => RpcResponse::Success(RpcSuccessResponse::BlobsByRoot(b)),
None => RpcResponse::StreamTermination(ResponseTermination::BlobsByRoot),
},
Response::BlobsByRange(r) => match r {
Some(b) => RPCCodedResponse::Success(RPCResponse::BlobsByRange(b)),
None => RPCCodedResponse::StreamTermination(ResponseTermination::BlobsByRange),
Some(b) => RpcResponse::Success(RpcSuccessResponse::BlobsByRange(b)),
None => RpcResponse::StreamTermination(ResponseTermination::BlobsByRange),
},
Response::DataColumnsByRoot(r) => match r {
Some(d) => RPCCodedResponse::Success(RPCResponse::DataColumnsByRoot(d)),
None => RPCCodedResponse::StreamTermination(ResponseTermination::DataColumnsByRoot),
Some(d) => RpcResponse::Success(RpcSuccessResponse::DataColumnsByRoot(d)),
None => RpcResponse::StreamTermination(ResponseTermination::DataColumnsByRoot),
},
Response::DataColumnsByRange(r) => match r {
Some(d) => RPCCodedResponse::Success(RPCResponse::DataColumnsByRange(d)),
None => {
RPCCodedResponse::StreamTermination(ResponseTermination::DataColumnsByRange)
}
Some(d) => RpcResponse::Success(RpcSuccessResponse::DataColumnsByRange(d)),
None => RpcResponse::StreamTermination(ResponseTermination::DataColumnsByRange),
},
Response::Status(s) => RPCCodedResponse::Success(RPCResponse::Status(s)),
Response::Status(s) => RpcResponse::Success(RpcSuccessResponse::Status(s)),
Response::LightClientBootstrap(b) => {
RPCCodedResponse::Success(RPCResponse::LightClientBootstrap(b))
RpcResponse::Success(RpcSuccessResponse::LightClientBootstrap(b))
}
Response::LightClientOptimisticUpdate(o) => {
RPCCodedResponse::Success(RPCResponse::LightClientOptimisticUpdate(o))
RpcResponse::Success(RpcSuccessResponse::LightClientOptimisticUpdate(o))
}
Response::LightClientFinalityUpdate(f) => {
RPCCodedResponse::Success(RPCResponse::LightClientFinalityUpdate(f))
RpcResponse::Success(RpcSuccessResponse::LightClientFinalityUpdate(f))
}
}
}

291
beacon_node/lighthouse_network/src/service/mod.rs
View File

@@ -11,9 +11,8 @@ use crate::peer_manager::{
use crate::peer_manager::{MIN_OUTBOUND_ONLY_FACTOR, PEER_EXCESS_FACTOR, PRIORITY_PEER_EXCESS};
use crate::rpc::methods::MetadataRequest;
use crate::rpc::{
methods, BlocksByRangeRequest, GoodbyeReason, HandlerErr, InboundRequest, NetworkParams,
OutboundRequest, Protocol, RPCCodedResponse, RPCError, RPCMessage, RPCReceived, RPCResponse,
RPCResponseErrorCode, ResponseTermination, RPC,
self, GoodbyeReason, HandlerErr, NetworkParams, Protocol, RPCError, RPCMessage, RPCReceived,
RequestType, ResponseTermination, RpcErrorResponse, RpcResponse, RpcSuccessResponse, RPC,
};
use crate::service::behaviour::BehaviourEvent;
pub use crate::service::behaviour::Gossipsub;
@@ -25,7 +24,7 @@ use crate::types::{
use crate::EnrExt;
use crate::Eth2Enr;
use crate::{error, metrics, Enr, NetworkGlobals, PubsubMessage, TopicHash};
use api_types::{AppRequestId, PeerRequestId, Request, RequestId, Response};
use api_types::{AppRequestId, PeerRequestId, RequestId, Response};
use futures::stream::StreamExt;
use gossipsub::{
IdentTopic as Topic, MessageAcceptance, MessageAuthenticity, MessageId, PublishError,
@@ -84,7 +83,7 @@ pub enum NetworkEvent<E: EthSpec> {
/// Identifier of the request. All responses to this request must use this id.
id: PeerRequestId,
/// Request the peer sent.
request: Request,
request: rpc::Request<E>,
},
ResponseReceived {
/// Peer that sent the response.
@@ -934,25 +933,28 @@ impl<E: EthSpec> Network<E> {
&mut self,
peer_id: PeerId,
request_id: AppRequestId,
request: Request,
request: RequestType<E>,
) -> Result<(), (AppRequestId, RPCError)> {
// Check if the peer is connected before sending an RPC request
if !self.swarm.is_connected(&peer_id) {
return Err((request_id, RPCError::Disconnected));
}
self.eth2_rpc_mut().send_request(
peer_id,
RequestId::Application(request_id),
request.into(),
);
self.eth2_rpc_mut()
.send_request(peer_id, RequestId::Application(request_id), request);
Ok(())
}
/// Send a successful response to a peer over RPC.
pub fn send_response(&mut self, peer_id: PeerId, id: PeerRequestId, response: Response<E>) {
pub fn send_response(
&mut self,
peer_id: PeerId,
id: PeerRequestId,
request_id: rpc::RequestId,
response: Response<E>,
) {
self.eth2_rpc_mut()
.send_response(peer_id, id, response.into())
.send_response(peer_id, id, request_id, response.into())
}
/// Inform the peer that their request produced an error.
@@ -960,13 +962,15 @@ impl<E: EthSpec> Network<E> {
&mut self,
peer_id: PeerId,
id: PeerRequestId,
error: RPCResponseErrorCode,
request_id: rpc::RequestId,
error: RpcErrorResponse,
reason: String,
) {
self.eth2_rpc_mut().send_response(
peer_id,
id,
RPCCodedResponse::Error(error, reason.into()),
request_id,
RpcResponse::Error(error, reason.into()),
)
}
@@ -1130,10 +1134,10 @@ impl<E: EthSpec> Network<E> {
let event = if self.fork_context.spec.is_peer_das_scheduled() {
// Nodes with higher custody will probably start advertising it
// before peerdas is activated
OutboundRequest::MetaData(MetadataRequest::new_v3())
RequestType::MetaData(MetadataRequest::new_v3())
} else {
// We always prefer sending V2 requests otherwise
OutboundRequest::MetaData(MetadataRequest::new_v2())
RequestType::MetaData(MetadataRequest::new_v2())
};
self.eth2_rpc_mut()
.send_request(peer_id, RequestId::Internal, event);
@@ -1144,12 +1148,14 @@ impl<E: EthSpec> Network<E> {
&mut self,
_req: MetadataRequest<E>,
id: PeerRequestId,
request_id: rpc::RequestId,
peer_id: PeerId,
) {
let metadata = self.network_globals.local_metadata.read().clone();
// The encoder is responsible for sending the negotiated version of the metadata
let event = RPCCodedResponse::Success(RPCResponse::MetaData(metadata));
self.eth2_rpc_mut().send_response(peer_id, id, event);
let event = RpcResponse::Success(RpcSuccessResponse::MetaData(metadata));
self.eth2_rpc_mut()
.send_response(peer_id, id, request_id, event);
}
// RPC Propagation methods
@@ -1171,56 +1177,6 @@ impl<E: EthSpec> Network<E> {
}
}
/// Convenience function to propagate a request.
#[must_use = "actually return the event"]
fn build_request(
&mut self,
id: PeerRequestId,
peer_id: PeerId,
request: Request,
) -> NetworkEvent<E> {
// Increment metrics
match &request {
Request::Status(_) => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["status"])
}
Request::LightClientBootstrap(_) => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["light_client_bootstrap"])
}
Request::LightClientOptimisticUpdate => metrics::inc_counter_vec(
&metrics::TOTAL_RPC_REQUESTS,
&["light_client_optimistic_update"],
),
Request::LightClientFinalityUpdate => metrics::inc_counter_vec(
&metrics::TOTAL_RPC_REQUESTS,
&["light_client_finality_update"],
),
Request::BlocksByRange { .. } => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["blocks_by_range"])
}
Request::BlocksByRoot { .. } => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["blocks_by_root"])
}
Request::BlobsByRange { .. } => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["blobs_by_range"])
}
Request::BlobsByRoot { .. } => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["blobs_by_root"])
}
Request::DataColumnsByRoot { .. } => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["data_columns_by_root"])
}
Request::DataColumnsByRange { .. } => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["data_columns_by_range"])
}
}
NetworkEvent::RequestReceived {
peer_id,
id,
request,
}
}
/// Dial cached Enrs in discovery service that are in the given `subnet_id` and aren't
/// in Connected, Dialing or Banned state.
fn dial_cached_enrs_in_subnet(&mut self, subnet: Subnet, spec: Arc<ChainSpec>) {
@@ -1406,7 +1362,7 @@ impl<E: EthSpec> Network<E> {
return None;
}
let handler_id = event.conn_id;
let connection_id = event.conn_id;
// The METADATA and PING RPC responses are handled within the behaviour and not propagated
match event.message {
Err(handler_err) => {
@@ -1444,21 +1400,25 @@ impl<E: EthSpec> Network<E> {
}
}
}
Ok(RPCReceived::Request(id, request)) => {
let peer_request_id = (handler_id, id);
match request {
Ok(RPCReceived::Request(request)) => {
match request.r#type {
/* Behaviour managed protocols: Ping and Metadata */
InboundRequest::Ping(ping) => {
RequestType::Ping(ping) => {
// inform the peer manager and send the response
self.peer_manager_mut().ping_request(&peer_id, ping.data);
None
}
InboundRequest::MetaData(req) => {
RequestType::MetaData(req) => {
// send the requested meta-data
self.send_meta_data_response(req, (handler_id, id), peer_id);
self.send_meta_data_response(
req,
(connection_id, request.substream_id),
request.id,
peer_id,
);
None
}
InboundRequest::Goodbye(reason) => {
RequestType::Goodbye(reason) => {
// queue for disconnection without a goodbye message
debug!(
self.log, "Peer sent Goodbye";
@@ -1473,17 +1433,19 @@ impl<E: EthSpec> Network<E> {
None
}
/* Protocols propagated to the Network */
InboundRequest::Status(msg) => {
RequestType::Status(_) => {
// inform the peer manager that we have received a status from a peer
self.peer_manager_mut().peer_statusd(&peer_id);
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["status"]);
// propagate the STATUS message upwards
let event =
self.build_request(peer_request_id, peer_id, Request::Status(msg));
Some(event)
Some(NetworkEvent::RequestReceived {
peer_id,
id: (connection_id, request.substream_id),
request,
})
}
InboundRequest::BlocksByRange(req) => {
RequestType::BlocksByRange(ref req) => {
// Still disconnect the peer if the request is naughty.
let mut count = *req.count();
if *req.step() == 0 {
self.peer_manager_mut().handle_rpc_error(
&peer_id,
@@ -1495,131 +1457,144 @@ impl<E: EthSpec> Network<E> {
);
return None;
}
// return just one block in case the step parameter is used. https://github.com/ethereum/consensus-specs/pull/2856
if *req.step() > 1 {
count = 1;
}
let request = match req {
methods::OldBlocksByRangeRequest::V1(req) => Request::BlocksByRange(
BlocksByRangeRequest::new_v1(req.start_slot, count),
),
methods::OldBlocksByRangeRequest::V2(req) => Request::BlocksByRange(
BlocksByRangeRequest::new(req.start_slot, count),
),
};
let event = self.build_request(peer_request_id, peer_id, request);
Some(event)
}
InboundRequest::BlocksByRoot(req) => {
let event = self.build_request(
peer_request_id,
peer_id,
Request::BlocksByRoot(req),
metrics::inc_counter_vec(
&metrics::TOTAL_RPC_REQUESTS,
&["blocks_by_range"],
);
Some(event)
}
InboundRequest::BlobsByRange(req) => {
let event = self.build_request(
peer_request_id,
Some(NetworkEvent::RequestReceived {
peer_id,
Request::BlobsByRange(req),
);
Some(event)
id: (connection_id, request.substream_id),
request,
})
}
InboundRequest::BlobsByRoot(req) => {
let event =
self.build_request(peer_request_id, peer_id, Request::BlobsByRoot(req));
Some(event)
}
InboundRequest::DataColumnsByRoot(req) => {
let event = self.build_request(
peer_request_id,
RequestType::BlocksByRoot(_) => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["blocks_by_root"]);
Some(NetworkEvent::RequestReceived {
peer_id,
Request::DataColumnsByRoot(req),
);
Some(event)
id: (connection_id, request.substream_id),
request,
})
}
InboundRequest::DataColumnsByRange(req) => {
let event = self.build_request(
peer_request_id,
RequestType::BlobsByRange(_) => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["blobs_by_range"]);
Some(NetworkEvent::RequestReceived {
peer_id,
Request::DataColumnsByRange(req),
);
Some(event)
id: (connection_id, request.substream_id),
request,
})
}
InboundRequest::LightClientBootstrap(req) => {
let event = self.build_request(
peer_request_id,
RequestType::BlobsByRoot(_) => {
metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["blobs_by_root"]);
Some(NetworkEvent::RequestReceived {
peer_id,
Request::LightClientBootstrap(req),
);
Some(event)
id: (connection_id, request.substream_id),
request,
})
}
InboundRequest::LightClientOptimisticUpdate => {
let event = self.build_request(
peer_request_id,
peer_id,
Request::LightClientOptimisticUpdate,
RequestType::DataColumnsByRoot(_) => {
metrics::inc_counter_vec(
&metrics::TOTAL_RPC_REQUESTS,
&["data_columns_by_root"],
);
Some(event)
Some(NetworkEvent::RequestReceived {
peer_id,
id: (connection_id, request.substream_id),
request,
})
}
InboundRequest::LightClientFinalityUpdate => {
let event = self.build_request(
peer_request_id,
peer_id,
Request::LightClientFinalityUpdate,
RequestType::DataColumnsByRange(_) => {
metrics::inc_counter_vec(
&metrics::TOTAL_RPC_REQUESTS,
&["data_columns_by_range"],
);
Some(event)
Some(NetworkEvent::RequestReceived {
peer_id,
id: (connection_id, request.substream_id),
request,
})
}
RequestType::LightClientBootstrap(_) => {
metrics::inc_counter_vec(
&metrics::TOTAL_RPC_REQUESTS,
&["light_client_bootstrap"],
);
Some(NetworkEvent::RequestReceived {
peer_id,
id: (connection_id, request.substream_id),
request,
})
}
RequestType::LightClientOptimisticUpdate => {
metrics::inc_counter_vec(
&metrics::TOTAL_RPC_REQUESTS,
&["light_client_optimistic_update"],
);
Some(NetworkEvent::RequestReceived {
peer_id,
id: (connection_id, request.substream_id),
request,
})
}
RequestType::LightClientFinalityUpdate => {
metrics::inc_counter_vec(
&metrics::TOTAL_RPC_REQUESTS,
&["light_client_finality_update"],
);
Some(NetworkEvent::RequestReceived {
peer_id,
id: (connection_id, request.substream_id),
request,
})
}
}
}
Ok(RPCReceived::Response(id, resp)) => {
match resp {
/* Behaviour managed protocols */
RPCResponse::Pong(ping) => {
RpcSuccessResponse::Pong(ping) => {
self.peer_manager_mut().pong_response(&peer_id, ping.data);
None
}
RPCResponse::MetaData(meta_data) => {
RpcSuccessResponse::MetaData(meta_data) => {
self.peer_manager_mut()
.meta_data_response(&peer_id, meta_data);
None
}
/* Network propagated protocols */
RPCResponse::Status(msg) => {
RpcSuccessResponse::Status(msg) => {
// inform the peer manager that we have received a status from a peer
self.peer_manager_mut().peer_statusd(&peer_id);
// propagate the STATUS message upwards
self.build_response(id, peer_id, Response::Status(msg))
}
RPCResponse::BlocksByRange(resp) => {
RpcSuccessResponse::BlocksByRange(resp) => {
self.build_response(id, peer_id, Response::BlocksByRange(Some(resp)))
}
RPCResponse::BlobsByRange(resp) => {
RpcSuccessResponse::BlobsByRange(resp) => {
self.build_response(id, peer_id, Response::BlobsByRange(Some(resp)))
}
RPCResponse::BlocksByRoot(resp) => {
RpcSuccessResponse::BlocksByRoot(resp) => {
self.build_response(id, peer_id, Response::BlocksByRoot(Some(resp)))
}
RPCResponse::BlobsByRoot(resp) => {
RpcSuccessResponse::BlobsByRoot(resp) => {
self.build_response(id, peer_id, Response::BlobsByRoot(Some(resp)))
}
RPCResponse::DataColumnsByRoot(resp) => {
RpcSuccessResponse::DataColumnsByRoot(resp) => {
self.build_response(id, peer_id, Response::DataColumnsByRoot(Some(resp)))
}
RPCResponse::DataColumnsByRange(resp) => {
RpcSuccessResponse::DataColumnsByRange(resp) => {
self.build_response(id, peer_id, Response::DataColumnsByRange(Some(resp)))
}
// Should never be reached
RPCResponse::LightClientBootstrap(bootstrap) => {
RpcSuccessResponse::LightClientBootstrap(bootstrap) => {
self.build_response(id, peer_id, Response::LightClientBootstrap(bootstrap))
}
RPCResponse::LightClientOptimisticUpdate(update) => self.build_response(
RpcSuccessResponse::LightClientOptimisticUpdate(update) => self.build_response(
id,
peer_id,
Response::LightClientOptimisticUpdate(update),
),
RPCResponse::LightClientFinalityUpdate(update) => self.build_response(
RpcSuccessResponse::LightClientFinalityUpdate(update) => self.build_response(
id,
peer_id,
Response::LightClientFinalityUpdate(update),

135
beacon_node/lighthouse_network/tests/rpc_tests.rs
View File

@@ -3,9 +3,9 @@
mod common;
use common::Protocol;
use lighthouse_network::rpc::methods::*;
use lighthouse_network::rpc::{methods::*, RequestType};
use lighthouse_network::service::api_types::AppRequestId;
use lighthouse_network::{rpc::max_rpc_size, NetworkEvent, ReportSource, Request, Response};
use lighthouse_network::{rpc::max_rpc_size, NetworkEvent, ReportSource, Response};
use slog::{debug, warn, Level};
use ssz::Encode;
use ssz_types::VariableList;
@@ -75,7 +75,7 @@ fn test_tcp_status_rpc() {
.await;
// Dummy STATUS RPC message
let rpc_request = Request::Status(StatusMessage {
let rpc_request = RequestType::Status(StatusMessage {
fork_digest: [0; 4],
finalized_root: Hash256::zero(),
finalized_epoch: Epoch::new(1),
@@ -128,10 +128,10 @@ fn test_tcp_status_rpc() {
id,
request,
} => {
if request == rpc_request {
if request.r#type == rpc_request {
// send the response
debug!(log, "Receiver Received");
receiver.send_response(peer_id, id, rpc_response.clone());
receiver.send_response(peer_id, id, request.id, rpc_response.clone());
}
}
_ => {} // Ignore other events
@@ -177,7 +177,12 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
.await;
// BlocksByRange Request
let rpc_request = Request::BlocksByRange(BlocksByRangeRequest::new(0, messages_to_send));
let rpc_request =
RequestType::BlocksByRange(OldBlocksByRangeRequest::V2(OldBlocksByRangeRequestV2 {
start_slot: 0,
count: messages_to_send,
step: 1,
}));
// BlocksByRange Response
let full_block = BeaconBlock::Base(BeaconBlockBase::<E>::full(&spec));
@@ -247,7 +252,7 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
id,
request,
} => {
if request == rpc_request {
if request.r#type == rpc_request {
// send the response
warn!(log, "Receiver got request");
for i in 0..messages_to_send {
@@ -260,10 +265,20 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
} else {
rpc_response_bellatrix_small.clone()
};
receiver.send_response(peer_id, id, rpc_response.clone());
receiver.send_response(
peer_id,
id,
request.id,
rpc_response.clone(),
);
}
// send the stream termination
receiver.send_response(peer_id, id, Response::BlocksByRange(None));
receiver.send_response(
peer_id,
id,
request.id,
Response::BlocksByRange(None),
);
}
}
_ => {} // Ignore other events
@@ -309,7 +324,7 @@ fn test_blobs_by_range_chunked_rpc() {
.await;
// BlobsByRange Request
let rpc_request = Request::BlobsByRange(BlobsByRangeRequest {
let rpc_request = RequestType::BlobsByRange(BlobsByRangeRequest {
start_slot: 0,
count: slot_count,
});
@@ -367,16 +382,26 @@ fn test_blobs_by_range_chunked_rpc() {
id,
request,
} => {
if request == rpc_request {
if request.r#type == rpc_request {
// send the response
warn!(log, "Receiver got request");
for _ in 0..messages_to_send {
// Send first third of responses as base blocks,
// second as altair and third as bellatrix.
receiver.send_response(peer_id, id, rpc_response.clone());
receiver.send_response(
peer_id,
id,
request.id,
rpc_response.clone(),
);
}
// send the stream termination
receiver.send_response(peer_id, id, Response::BlobsByRange(None));
receiver.send_response(
peer_id,
id,
request.id,
Response::BlobsByRange(None),
);
}
}
_ => {} // Ignore other events
@@ -422,7 +447,12 @@ fn test_tcp_blocks_by_range_over_limit() {
.await;
// BlocksByRange Request
let rpc_request = Request::BlocksByRange(BlocksByRangeRequest::new(0, messages_to_send));
let rpc_request =
RequestType::BlocksByRange(OldBlocksByRangeRequest::V1(OldBlocksByRangeRequestV1 {
start_slot: 0,
count: messages_to_send,
step: 1,
}));
// BlocksByRange Response
let full_block = bellatrix_block_large(&common::fork_context(ForkName::Bellatrix), &spec);
@@ -460,15 +490,25 @@ fn test_tcp_blocks_by_range_over_limit() {
id,
request,
} => {
if request == rpc_request {
if request.r#type == rpc_request {
// send the response
warn!(log, "Receiver got request");
for _ in 0..messages_to_send {
let rpc_response = rpc_response_bellatrix_large.clone();
receiver.send_response(peer_id, id, rpc_response.clone());
receiver.send_response(
peer_id,
id,
request.id,
rpc_response.clone(),
);
}
// send the stream termination
receiver.send_response(peer_id, id, Response::BlocksByRange(None));
receiver.send_response(
peer_id,
id,
request.id,
Response::BlocksByRange(None),
);
}
}
_ => {} // Ignore other events
@@ -514,7 +554,12 @@ fn test_tcp_blocks_by_range_chunked_rpc_terminates_correctly() {
.await;
// BlocksByRange Request
let rpc_request = Request::BlocksByRange(BlocksByRangeRequest::new(0, messages_to_send));
let rpc_request =
RequestType::BlocksByRange(OldBlocksByRangeRequest::V2(OldBlocksByRangeRequestV2 {
start_slot: 0,
count: messages_to_send,
step: 1,
}));
// BlocksByRange Response
let empty_block = BeaconBlock::empty(&spec);
@@ -583,10 +628,10 @@ fn test_tcp_blocks_by_range_chunked_rpc_terminates_correctly() {
},
_,
)) => {
if request == rpc_request {
if request.r#type == rpc_request {
// send the response
warn!(log, "Receiver got request");
message_info = Some((peer_id, id));
message_info = Some((peer_id, id, request.id));
}
}
futures::future::Either::Right((_, _)) => {} // The timeout hit, send messages if required
@@ -596,8 +641,8 @@ fn test_tcp_blocks_by_range_chunked_rpc_terminates_correctly() {
// if we need to send messages send them here. This will happen after a delay
if message_info.is_some() {
messages_sent += 1;
let (peer_id, stream_id) = message_info.as_ref().unwrap();
receiver.send_response(*peer_id, *stream_id, rpc_response.clone());
let (peer_id, stream_id, request_id) = message_info.as_ref().unwrap();
receiver.send_response(*peer_id, *stream_id, *request_id, rpc_response.clone());
debug!(log, "Sending message {}", messages_sent);
if messages_sent == messages_to_send + extra_messages_to_send {
// stop sending messages
@@ -642,7 +687,12 @@ fn test_tcp_blocks_by_range_single_empty_rpc() {
.await;
// BlocksByRange Request
let rpc_request = Request::BlocksByRange(BlocksByRangeRequest::new(0, 10));
let rpc_request =
RequestType::BlocksByRange(OldBlocksByRangeRequest::V2(OldBlocksByRangeRequestV2 {
start_slot: 0,
count: 10,
step: 1,
}));
// BlocksByRange Response
let empty_block = BeaconBlock::empty(&spec);
@@ -696,15 +746,25 @@ fn test_tcp_blocks_by_range_single_empty_rpc() {
id,
request,
} => {
if request == rpc_request {
if request.r#type == rpc_request {
// send the response
warn!(log, "Receiver got request");
for _ in 1..=messages_to_send {
receiver.send_response(peer_id, id, rpc_response.clone());
receiver.send_response(
peer_id,
id,
request.id,
rpc_response.clone(),
);
}
// send the stream termination
receiver.send_response(peer_id, id, Response::BlocksByRange(None));
receiver.send_response(
peer_id,
id,
request.id,
Response::BlocksByRange(None),
);
}
}
_ => {} // Ignore other events
@@ -750,7 +810,7 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
.await;
// BlocksByRoot Request
let rpc_request = Request::BlocksByRoot(BlocksByRootRequest::new(
let rpc_request = RequestType::BlocksByRoot(BlocksByRootRequest::new(
vec![
Hash256::zero(),
Hash256::zero(),
@@ -827,7 +887,7 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
id,
request,
} => {
if request == rpc_request {
if request.r#type == rpc_request {
// send the response
debug!(log, "Receiver got request");
@@ -840,11 +900,16 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
} else {
rpc_response_bellatrix_small.clone()
};
receiver.send_response(peer_id, id, rpc_response);
receiver.send_response(peer_id, id, request.id, rpc_response);
debug!(log, "Sending message");
}
// send the stream termination
receiver.send_response(peer_id, id, Response::BlocksByRange(None));
receiver.send_response(
peer_id,
id,
request.id,
Response::BlocksByRange(None),
);
debug!(log, "Send stream term");
}
}
@@ -888,7 +953,7 @@ fn test_tcp_blocks_by_root_chunked_rpc_terminates_correctly() {
.await;
// BlocksByRoot Request
let rpc_request = Request::BlocksByRoot(BlocksByRootRequest::new(
let rpc_request = RequestType::BlocksByRoot(BlocksByRootRequest::new(
vec![
Hash256::zero(),
Hash256::zero(),
@@ -971,10 +1036,10 @@ fn test_tcp_blocks_by_root_chunked_rpc_terminates_correctly() {
},
_,
)) => {
if request == rpc_request {
if request.r#type == rpc_request {
// send the response
warn!(log, "Receiver got request");
message_info = Some((peer_id, id));
message_info = Some((peer_id, id, request.id));
}
}
futures::future::Either::Right((_, _)) => {} // The timeout hit, send messages if required
@@ -984,8 +1049,8 @@ fn test_tcp_blocks_by_root_chunked_rpc_terminates_correctly() {
// if we need to send messages send them here. This will happen after a delay
if message_info.is_some() {
messages_sent += 1;
let (peer_id, stream_id) = message_info.as_ref().unwrap();
receiver.send_response(*peer_id, *stream_id, rpc_response.clone());
let (peer_id, stream_id, request_id) = message_info.as_ref().unwrap();
receiver.send_response(*peer_id, *stream_id, *request_id, rpc_response.clone());
debug!(log, "Sending message {}", messages_sent);
if messages_sent == messages_to_send + extra_messages_to_send {
// stop sending messages