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

Merge branch 'unstable' into deneb-merge-from-unstable-20230627

Browse Source 
# Conflicts:
#	beacon_node/beacon_chain/src/beacon_chain.rs
#	beacon_node/beacon_chain/src/block_verification.rs
#	beacon_node/beacon_chain/src/lib.rs
#	beacon_node/beacon_chain/src/test_utils.rs
#	beacon_node/beacon_chain/tests/block_verification.rs
#	beacon_node/beacon_chain/tests/store_tests.rs
#	beacon_node/beacon_chain/tests/tests.rs
#	beacon_node/http_api/src/publish_blocks.rs
#	beacon_node/lighthouse_network/src/rpc/codec/ssz_snappy.rs
#	beacon_node/lighthouse_network/src/rpc/methods.rs
#	beacon_node/lighthouse_network/src/rpc/outbound.rs
#	beacon_node/lighthouse_network/src/rpc/protocol.rs
#	beacon_node/lighthouse_network/src/service/api_types.rs
#	beacon_node/network/src/beacon_processor/worker/gossip_methods.rs
#	beacon_node/network/src/beacon_processor/worker/rpc_methods.rs
#	beacon_node/network/src/beacon_processor/worker/sync_methods.rs
#	beacon_node/network/src/sync/block_lookups/single_block_lookup.rs
#	beacon_node/network/src/sync/network_context.rs
#	beacon_node/network/src/sync/range_sync/batch.rs
#	beacon_node/network/src/sync/range_sync/chain.rs
#	common/eth2/src/types.rs
#	consensus/fork_choice/src/fork_choice.rs
This commit is contained in:
Jimmy Chen
2023-06-27 08:38:54 +10:00
parent e1af24b470 cc780aae3e
commit 97c4660761
81 changed files with 1675 additions and 1243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
beacon_node/lighthouse_network/src/config.rs
View File

@@ -163,7 +163,7 @@ impl Config {
udp_port,
tcp_port,
});
self.discv5_config.ip_mode = discv5::IpMode::Ip4;
self.discv5_config.listen_config = discv5::ListenConfig::from_ip(addr.into(), udp_port);
self.discv5_config.table_filter = |enr| enr.ip4().as_ref().map_or(false, is_global_ipv4)
}
@@ -176,9 +176,8 @@ impl Config {
udp_port,
tcp_port,
});
self.discv5_config.ip_mode = discv5::IpMode::Ip6 {
enable_mapped_addresses: false,
};
self.discv5_config.listen_config = discv5::ListenConfig::from_ip(addr.into(), udp_port);
self.discv5_config.table_filter = |enr| enr.ip6().as_ref().map_or(false, is_global_ipv6)
}
@@ -206,10 +205,10 @@ impl Config {
tcp_port: tcp6_port,
},
);
self.discv5_config.listen_config = discv5::ListenConfig::default()
.with_ipv4(v4_addr, udp4_port)
.with_ipv6(v6_addr, udp6_port);
self.discv5_config.ip_mode = discv5::IpMode::Ip6 {
enable_mapped_addresses: true,
};
self.discv5_config.table_filter = |enr| match (&enr.ip4(), &enr.ip6()) {
(None, None) => false,
(None, Some(ip6)) => is_global_ipv6(ip6),
@@ -279,9 +278,17 @@ impl Default for Config {
.build()
.expect("The total rate limit has been specified"),
);
let listen_addresses = ListenAddress::V4(ListenAddr {
addr: Ipv4Addr::UNSPECIFIED,
udp_port: 9000,
tcp_port: 9000,
});
let discv5_listen_config =
discv5::ListenConfig::from_ip(Ipv4Addr::UNSPECIFIED.into(), 9000);
// discv5 configuration
let discv5_config = Discv5ConfigBuilder::new()
let discv5_config = Discv5ConfigBuilder::new(discv5_listen_config)
.enable_packet_filter()
.session_cache_capacity(5000)
.request_timeout(Duration::from_secs(1))
@@ -304,12 +311,9 @@ impl Default for Config {
// NOTE: Some of these get overridden by the corresponding CLI default values.
Config {
network_dir,
listen_addresses: ListenAddress::V4(ListenAddr {
addr: Ipv4Addr::UNSPECIFIED,
udp_port: 9000,
tcp_port: 9000,
}),
listen_addresses,
enr_address: (None, None),
enr_udp4_port: None,
enr_tcp4_port: None,
enr_udp6_port: None,

6
beacon_node/lighthouse_network/src/discovery/enr.rs
View File

@@ -213,13 +213,17 @@ pub fn build_enr<T: EthSpec>(
fn compare_enr(local_enr: &Enr, disk_enr: &Enr) -> bool {
// take preference over disk_enr address if one is not specified
(local_enr.ip4().is_none() || local_enr.ip4() == disk_enr.ip4())
&&
(local_enr.ip6().is_none() || local_enr.ip6() == disk_enr.ip6())
// tcp ports must match
&& local_enr.tcp4() == disk_enr.tcp4()
&& local_enr.tcp6() == disk_enr.tcp6()
// must match on the same fork
&& local_enr.get(ETH2_ENR_KEY) == disk_enr.get(ETH2_ENR_KEY)
// take preference over disk udp port if one is not specified
&& (local_enr.udp4().is_none() || local_enr.udp4() == disk_enr.udp4())
// we need the ATTESTATION_BITFIELD_ENR_KEY and SYNC_COMMITTEE_BITFIELD_ENR_KEY key to match,
&& (local_enr.udp6().is_none() || local_enr.udp6() == disk_enr.udp6())
// we need the ATTESTATION_BITFIELD_ENR_KEY and SYNC_COMMITTEE_BITFIELD_ENR_KEY key to match,
// otherwise we use a new ENR. This will likely only be true for non-validating nodes
&& local_enr.get(ATTESTATION_BITFIELD_ENR_KEY) == disk_enr.get(ATTESTATION_BITFIELD_ENR_KEY)
&& local_enr.get(SYNC_COMMITTEE_BITFIELD_ENR_KEY) == disk_enr.get(SYNC_COMMITTEE_BITFIELD_ENR_KEY)

28
beacon_node/lighthouse_network/src/discovery/enr_ext.rs
View File

@@ -198,7 +198,7 @@ impl CombinedKeyPublicExt for CombinedPublicKey {
fn as_peer_id(&self) -> PeerId {
match self {
Self::Secp256k1(pk) => {
let pk_bytes = pk.to_bytes();
let pk_bytes = pk.to_sec1_bytes();
let libp2p_pk = libp2p::core::PublicKey::Secp256k1(
libp2p::core::identity::secp256k1::PublicKey::decode(&pk_bytes)
.expect("valid public key"),
@@ -222,14 +222,16 @@ impl CombinedKeyExt for CombinedKey {
match key {
Keypair::Secp256k1(key) => {
let secret =
discv5::enr::k256::ecdsa::SigningKey::from_bytes(&key.secret().to_bytes())
discv5::enr::k256::ecdsa::SigningKey::from_slice(&key.secret().to_bytes())
.expect("libp2p key must be valid");
Ok(CombinedKey::Secp256k1(secret))
}
Keypair::Ed25519(key) => {
let ed_keypair =
discv5::enr::ed25519_dalek::SecretKey::from_bytes(&key.encode()[..32])
.expect("libp2p key must be valid");
let ed_keypair = discv5::enr::ed25519_dalek::SigningKey::from_bytes(
&(key.encode()[..32])
.try_into()
.expect("libp2p key must be valid"),
);
Ok(CombinedKey::from(ed_keypair))
}
Keypair::Ecdsa(_) => Err("Ecdsa keypairs not supported"),
@@ -281,7 +283,7 @@ mod tests {
fn test_secp256k1_peer_id_conversion() {
let sk_hex = "df94a73d528434ce2309abb19c16aedb535322797dbd59c157b1e04095900f48";
let sk_bytes = hex::decode(sk_hex).unwrap();
let secret_key = discv5::enr::k256::ecdsa::SigningKey::from_bytes(&sk_bytes).unwrap();
let secret_key = discv5::enr::k256::ecdsa::SigningKey::from_slice(&sk_bytes).unwrap();
let libp2p_sk = libp2p::identity::secp256k1::SecretKey::from_bytes(sk_bytes).unwrap();
let secp256k1_kp: libp2p::identity::secp256k1::Keypair = libp2p_sk.into();
@@ -300,16 +302,18 @@ mod tests {
fn test_ed25519_peer_conversion() {
let sk_hex = "4dea8a5072119927e9d243a7d953f2f4bc95b70f110978e2f9bc7a9000e4b261";
let sk_bytes = hex::decode(sk_hex).unwrap();
let secret = discv5::enr::ed25519_dalek::SecretKey::from_bytes(&sk_bytes).unwrap();
let public = discv5::enr::ed25519_dalek::PublicKey::from(&secret);
let keypair = discv5::enr::ed25519_dalek::Keypair { secret, public };
let secret_key = discv5::enr::ed25519_dalek::SigningKey::from_bytes(
&sk_bytes.clone().try_into().unwrap(),
);
let libp2p_sk = libp2p::identity::ed25519::SecretKey::from_bytes(sk_bytes).unwrap();
let ed25519_kp: libp2p::identity::ed25519::Keypair = libp2p_sk.into();
let libp2p_kp = Keypair::Ed25519(ed25519_kp);
let secp256k1_kp: libp2p::identity::ed25519::Keypair = libp2p_sk.into();
let libp2p_kp = Keypair::Ed25519(secp256k1_kp);
let peer_id = libp2p_kp.public().to_peer_id();
let enr = discv5::enr::EnrBuilder::new("v4").build(&keypair).unwrap();
let enr = discv5::enr::EnrBuilder::new("v4")
.build(&secret_key)
.unwrap();
let node_id = peer_id_to_node_id(&peer_id).unwrap();
assert_eq!(enr.node_id(), node_id);

39
beacon_node/lighthouse_network/src/discovery/mod.rs
View File

@@ -209,13 +209,6 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
info!(log, "ENR Initialised"; "enr" => local_enr.to_base64(), "seq" => local_enr.seq(), "id"=> %local_enr.node_id(),
"ip4" => ?local_enr.ip4(), "udp4"=> ?local_enr.udp4(), "tcp4" => ?local_enr.tcp4(), "tcp6" => ?local_enr.tcp6(), "udp6" => ?local_enr.udp6()
);
let listen_socket = match config.listen_addrs() {
crate::listen_addr::ListenAddress::V4(v4_addr) => v4_addr.udp_socket_addr(),
crate::listen_addr::ListenAddress::V6(v6_addr) => v6_addr.udp_socket_addr(),
crate::listen_addr::ListenAddress::DualStack(_v4_addr, v6_addr) => {
v6_addr.udp_socket_addr()
}
};
// convert the keypair into an ENR key
let enr_key: CombinedKey = CombinedKey::from_libp2p(local_key)?;
@@ -251,10 +244,7 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
// Start the discv5 service and obtain an event stream
let event_stream = if !config.disable_discovery {
discv5
.start(listen_socket)
.map_err(|e| e.to_string())
.await?;
discv5.start().map_err(|e| e.to_string()).await?;
debug!(log, "Discovery service started");
EventStream::Awaiting(Box::pin(discv5.event_stream()))
} else {
@@ -413,7 +403,7 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
/// If the external address needs to be modified, use `update_enr_udp_socket.
pub fn update_enr_tcp_port(&mut self, port: u16) -> Result<(), String> {
self.discv5
.enr_insert("tcp", &port.to_be_bytes())
.enr_insert("tcp", &port)
.map_err(|e| format!("{:?}", e))?;
// replace the global version
@@ -428,29 +418,12 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
/// This is with caution. Discovery should automatically maintain this. This should only be
/// used when automatic discovery is disabled.
pub fn update_enr_udp_socket(&mut self, socket_addr: SocketAddr) -> Result<(), String> {
match socket_addr {
SocketAddr::V4(socket) => {
self.discv5
.enr_insert("ip", &socket.ip().octets())
.map_err(|e| format!("{:?}", e))?;
self.discv5
.enr_insert("udp", &socket.port().to_be_bytes())
.map_err(|e| format!("{:?}", e))?;
}
SocketAddr::V6(socket) => {
self.discv5
.enr_insert("ip6", &socket.ip().octets())
.map_err(|e| format!("{:?}", e))?;
self.discv5
.enr_insert("udp6", &socket.port().to_be_bytes())
.map_err(|e| format!("{:?}", e))?;
}
const IS_TCP: bool = false;
if self.discv5.update_local_enr_socket(socket_addr, IS_TCP) {
// persist modified enr to disk
enr::save_enr_to_disk(Path::new(&self.enr_dir), &self.local_enr(), &self.log);
}
// replace the global version
*self.network_globals.local_enr.write() = self.discv5.local_enr();
// persist modified enr to disk
enr::save_enr_to_disk(Path::new(&self.enr_dir), &self.local_enr(), &self.log);
Ok(())
}

9
beacon_node/lighthouse_network/src/rpc/codec/base.rs
View File

@@ -217,8 +217,7 @@ mod tests {
let mut buf = BytesMut::new();
buf.extend_from_slice(&message);
let snappy_protocol_id =
ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZSnappy);
let snappy_protocol_id = ProtocolId::new(SupportedProtocol::StatusV1, Encoding::SSZSnappy);
let fork_context = Arc::new(fork_context(ForkName::Base));
let mut snappy_outbound_codec = SSZSnappyOutboundCodec::<Spec>::new(
@@ -252,8 +251,7 @@ mod tests {
// Insert length-prefix
uvi_codec.encode(len, &mut dst).unwrap();
let snappy_protocol_id =
ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZSnappy);
let snappy_protocol_id = ProtocolId::new(SupportedProtocol::StatusV1, Encoding::SSZSnappy);
let fork_context = Arc::new(fork_context(ForkName::Base));
let mut snappy_outbound_codec = SSZSnappyOutboundCodec::<Spec>::new(
@@ -280,8 +278,7 @@ mod tests {
dst
}
let protocol_id =
ProtocolId::new(Protocol::BlocksByRange, Version::V1, Encoding::SSZSnappy);
let protocol_id = ProtocolId::new(SupportedProtocol::BlocksByRangeV1, Encoding::SSZSnappy);
// Response limits
let fork_context = Arc::new(fork_context(ForkName::Base));

632
beacon_node/lighthouse_network/src/rpc/codec/ssz_snappy.rs
View File

File diff suppressed because it is too large Load Diff

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

@@ -7,6 +7,7 @@ use super::protocol::{max_rpc_size, InboundRequest, Protocol, RPCError, RPCProto
use super::{RPCReceived, RPCSend, ReqId};
use crate::rpc::outbound::{OutboundFramed, OutboundRequest};
use crate::rpc::protocol::InboundFramed;
use crate::rpc::ResponseTermination;
use fnv::FnvHashMap;
use futures::prelude::*;
use futures::{Sink, SinkExt};
@@ -245,7 +246,7 @@ where
while let Some((id, req)) = self.dial_queue.pop() {
self.events_out.push(Err(HandlerErr::Outbound {
error: RPCError::Disconnected,
proto: req.protocol(),
proto: req.versioned_protocol().protocol(),
id,
}));
}
@@ -269,7 +270,7 @@ where
}
_ => self.events_out.push(Err(HandlerErr::Outbound {
error: RPCError::Disconnected,
proto: req.protocol(),
proto: req.versioned_protocol().protocol(),
id,
})),
}
@@ -334,7 +335,7 @@ where
) {
self.dial_negotiated -= 1;
let (id, request) = request_info;
let proto = request.protocol();
let proto = request.versioned_protocol().protocol();
// accept outbound connections only if the handler is not deactivated
if matches!(self.state, HandlerState::Deactivated) {
@@ -414,7 +415,7 @@ where
128,
) as usize),
delay_key: Some(delay_key),
protocol: req.protocol(),
protocol: req.versioned_protocol().protocol(),
request_start_time: Instant::now(),
remaining_chunks: expected_responses,
},
@@ -422,7 +423,7 @@ where
} else {
self.events_out.push(Err(HandlerErr::Inbound {
id: self.current_inbound_substream_id,
proto: req.protocol(),
proto: req.versioned_protocol().protocol(),
error: RPCError::HandlerRejected,
}));
return self.shutdown(None);
@@ -498,7 +499,7 @@ where
};
self.events_out.push(Err(HandlerErr::Outbound {
error,
proto: req.protocol(),
proto: req.versioned_protocol().protocol(),
id,
}));
}
@@ -895,7 +896,7 @@ where
// else we return an error, stream should not have closed early.
let outbound_err = HandlerErr::Outbound {
id: request_id,
proto: request.protocol(),
proto: request.versioned_protocol().protocol(),
error: RPCError::IncompleteStream,
};
return Poll::Ready(ConnectionHandlerEvent::Custom(Err(outbound_err)));
@@ -933,8 +934,13 @@ where
// continue sending responses beyond what we would expect. Here
// we simply terminate the stream and report a stream
// termination to the application
let termination = match protocol {
Protocol::BlocksByRange => Some(ResponseTermination::BlocksByRange),
Protocol::BlocksByRoot => Some(ResponseTermination::BlocksByRoot),
_ => None, // all other protocols are do not have multiple responses and we do not inform the user, we simply drop the stream.
};
if let Some(termination) = protocol.terminator() {
if let Some(termination) = termination {
return Poll::Ready(ConnectionHandlerEvent::Custom(Ok(
RPCReceived::EndOfStream(request_id, termination),
)));

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

@@ -3,11 +3,13 @@
use crate::types::{EnrAttestationBitfield, EnrSyncCommitteeBitfield};
use regex::bytes::Regex;
use serde::Serialize;
use ssz::Encode;
use ssz_derive::{Decode, Encode};
use ssz_types::{
typenum::{U1024, U256, U768},
VariableList,
};
use std::marker::PhantomData;
use std::ops::Deref;
use std::sync::Arc;
use strum::IntoStaticStr;
@@ -96,6 +98,30 @@ pub struct Ping {
pub data: u64,
}
/// The METADATA request structure.
#[superstruct(
variants(V1, V2),
variant_attributes(derive(Clone, Debug, PartialEq, Serialize),)
)]
#[derive(Clone, Debug, PartialEq)]
pub struct MetadataRequest<T: EthSpec> {
_phantom_data: PhantomData<T>,
}
impl<T: EthSpec> MetadataRequest<T> {
pub fn new_v1() -> Self {
Self::V1(MetadataRequestV1 {
_phantom_data: PhantomData,
})
}
pub fn new_v2() -> Self {
Self::V2(MetadataRequestV2 {
_phantom_data: PhantomData,
})
}
}
/// The METADATA response structure.
#[superstruct(
variants(V1, V2),
@@ -104,9 +130,8 @@ pub struct Ping {
serde(bound = "T: EthSpec", deny_unknown_fields),
)
)]
#[derive(Clone, Debug, PartialEq, Serialize, Encode)]
#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(bound = "T: EthSpec")]
#[ssz(enum_behaviour = "transparent")]
pub struct MetaData<T: EthSpec> {
/// A sequential counter indicating when data gets modified.
pub seq_number: u64,
@@ -117,6 +142,38 @@ pub struct MetaData<T: EthSpec> {
pub syncnets: EnrSyncCommitteeBitfield<T>,
}
impl<T: EthSpec> MetaData<T> {
/// Returns a V1 MetaData response from self.
pub fn metadata_v1(&self) -> Self {
match self {
md @ MetaData::V1(_) => md.clone(),
MetaData::V2(metadata) => MetaData::V1(MetaDataV1 {
seq_number: metadata.seq_number,
attnets: metadata.attnets.clone(),
}),
}
}
/// Returns a V2 MetaData response from self by filling unavailable fields with default.
pub fn metadata_v2(&self) -> Self {
match self {
MetaData::V1(metadata) => MetaData::V2(MetaDataV2 {
seq_number: metadata.seq_number,
attnets: metadata.attnets.clone(),
syncnets: Default::default(),
}),
md @ MetaData::V2(_) => md.clone(),
}
}
pub fn as_ssz_bytes(&self) -> Vec<u8> {
match self {
MetaData::V1(md) => md.as_ssz_bytes(),
MetaData::V2(md) => md.as_ssz_bytes(),
}
}
}
/// The reason given for a `Goodbye` message.
///
/// Note: any unknown `u64::into(n)` will resolve to `Goodbye::Unknown` for any unknown `n`,
@@ -208,7 +265,11 @@ impl ssz::Decode for GoodbyeReason {
}
/// Request a number of beacon block roots from a peer.
#[derive(Encode, Decode, Clone, Debug, PartialEq)]
#[superstruct(
variants(V1, V2),
variant_attributes(derive(Encode, Decode, Clone, Debug, PartialEq))
)]
#[derive(Clone, Debug, PartialEq)]
pub struct BlocksByRangeRequest {
/// The starting slot to request blocks.
pub start_slot: u64,
@@ -217,6 +278,17 @@ pub struct BlocksByRangeRequest {
pub count: u64,
}
impl BlocksByRangeRequest {
/// The default request is V2
pub fn new(start_slot: u64, count: u64) -> Self {
Self::V2(BlocksByRangeRequestV2 { start_slot, count })
}
pub fn new_v1(start_slot: u64, count: u64) -> Self {
Self::V1(BlocksByRangeRequestV1 { start_slot, count })
}
}
/// Request a number of beacon blobs from a peer.
#[derive(Encode, Decode, Clone, Debug, PartialEq)]
pub struct BlobsByRangeRequest {
@@ -228,7 +300,11 @@ pub struct BlobsByRangeRequest {
}
/// Request a number of beacon block roots from a peer.
#[derive(Encode, Decode, Clone, Debug, PartialEq)]
#[superstruct(
variants(V1, V2),
variant_attributes(derive(Encode, Decode, Clone, Debug, PartialEq))
)]
#[derive(Clone, Debug, PartialEq)]
pub struct OldBlocksByRangeRequest {
/// The starting slot to request blocks.
pub start_slot: u64,
@@ -244,13 +320,43 @@ pub struct OldBlocksByRangeRequest {
pub step: u64,
}
impl OldBlocksByRangeRequest {
/// The default request is V2
pub fn new(start_slot: u64, count: u64, step: u64) -> Self {
Self::V2(OldBlocksByRangeRequestV2 {
start_slot,
count,
step,
})
}
pub fn new_v1(start_slot: u64, count: u64, step: u64) -> Self {
Self::V1(OldBlocksByRangeRequestV1 {
start_slot,
count,
step,
})
}
}
/// Request a number of beacon block bodies from a peer.
#[derive(Encode, Decode, Clone, Debug, PartialEq)]
#[superstruct(variants(V1, V2), variant_attributes(derive(Clone, Debug, PartialEq)))]
#[derive(Clone, Debug, PartialEq)]
pub struct BlocksByRootRequest {
/// The list of beacon block bodies being requested.
pub block_roots: VariableList<Hash256, MaxRequestBlocks>,
}
impl BlocksByRootRequest {
pub fn new(block_roots: VariableList<Hash256, MaxRequestBlocks>) -> Self {
Self::V2(BlocksByRootRequestV2 { block_roots })
}
pub fn new_v1(block_roots: VariableList<Hash256, MaxRequestBlocks>) -> Self {
Self::V1(BlocksByRootRequestV1 { block_roots })
}
}
/// Request a number of beacon blocks and blobs from a peer.
#[derive(Clone, Debug, PartialEq)]
pub struct BlobsByRootRequest {
@@ -492,7 +598,12 @@ impl std::fmt::Display for GoodbyeReason {
impl std::fmt::Display for BlocksByRangeRequest {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Start Slot: {}, Count: {}", self.start_slot, self.count)
write!(
f,
"Start Slot: {}, Count: {}",
self.start_slot(),
self.count()
)
}
}
@@ -501,7 +612,9 @@ impl std::fmt::Display for OldBlocksByRangeRequest {
write!(
f,
"Start Slot: {}, Count: {}, Step: {}",
self.start_slot, self.count, self.step
self.start_slot(),
self.count(),
self.step()
)
}
}

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

@@ -247,7 +247,7 @@ where
}
Err(RateLimitedErr::TooLarge) => {
// we set the batch sizes, so this is a coding/config err for most protocols
let protocol = req.protocol();
let protocol = req.versioned_protocol().protocol();
if matches!(protocol, Protocol::BlocksByRange) {
debug!(self.log, "Blocks by range request will never be processed"; "request" => %req);
} else {
@@ -335,7 +335,7 @@ where
serializer.emit_arguments("peer_id", &format_args!("{}", self.peer_id))?;
let (msg_kind, protocol) = match &self.event {
Ok(received) => match received {
RPCReceived::Request(_, req) => ("request", req.protocol()),
RPCReceived::Request(_, req) => ("request", req.versioned_protocol().protocol()),
RPCReceived::Response(_, res) => ("response", res.protocol()),
RPCReceived::EndOfStream(_, end) => (
"end_of_stream",

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

@@ -1,11 +1,8 @@
use std::marker::PhantomData;
use super::methods::*;
use super::protocol::Protocol;
use super::protocol::ProtocolId;
use super::protocol::SupportedProtocol;
use super::RPCError;
use crate::rpc::protocol::Encoding;
use crate::rpc::protocol::Version;
use crate::rpc::{
codec::{base::BaseOutboundCodec, ssz_snappy::SSZSnappyOutboundCodec, OutboundCodec},
methods::ResponseTermination,
@@ -40,9 +37,8 @@ pub enum OutboundRequest<TSpec: EthSpec> {
BlocksByRoot(BlocksByRootRequest),
BlobsByRange(BlobsByRangeRequest),
BlobsByRoot(BlobsByRootRequest),
LightClientBootstrap(LightClientBootstrapRequest),
Ping(Ping),
MetaData(PhantomData<TSpec>),
MetaData(MetadataRequest<TSpec>),
}
impl<TSpec: EthSpec> UpgradeInfo for OutboundRequestContainer<TSpec> {
@@ -61,46 +57,37 @@ impl<TSpec: EthSpec> OutboundRequest<TSpec> {
match self {
// add more protocols when versions/encodings are supported
OutboundRequest::Status(_) => vec![ProtocolId::new(
Protocol::Status,
Version::V1,
SupportedProtocol::StatusV1,
Encoding::SSZSnappy,
)],
OutboundRequest::Goodbye(_) => vec![ProtocolId::new(
Protocol::Goodbye,
Version::V1,
SupportedProtocol::GoodbyeV1,
Encoding::SSZSnappy,
)],
OutboundRequest::BlocksByRange(_) => vec![
ProtocolId::new(Protocol::BlocksByRange, Version::V2, Encoding::SSZSnappy),
ProtocolId::new(Protocol::BlocksByRange, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlocksByRangeV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlocksByRangeV1, Encoding::SSZSnappy),
],
OutboundRequest::BlocksByRoot(_) => vec![
ProtocolId::new(Protocol::BlocksByRoot, Version::V2, Encoding::SSZSnappy),
ProtocolId::new(Protocol::BlocksByRoot, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlocksByRootV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlocksByRootV1, Encoding::SSZSnappy),
],
OutboundRequest::BlobsByRange(_) => vec![ProtocolId::new(
Protocol::BlobsByRange,
Version::V1,
SupportedProtocol::BlobsByRangeV1,
Encoding::SSZSnappy,
)],
OutboundRequest::BlobsByRoot(_) => vec![ProtocolId::new(
Protocol::BlobsByRoot,
Version::V1,
SupportedProtocol::BlobsByRootV1,
Encoding::SSZSnappy,
)],
OutboundRequest::Ping(_) => vec![ProtocolId::new(
Protocol::Ping,
Version::V1,
SupportedProtocol::PingV1,
Encoding::SSZSnappy,
)],
OutboundRequest::MetaData(_) => vec![
ProtocolId::new(Protocol::MetaData, Version::V2, Encoding::SSZSnappy),
ProtocolId::new(Protocol::MetaData, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::MetaDataV2, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::MetaDataV1, Encoding::SSZSnappy),
],
// Note: This match arm is technically unreachable as we only respond to light client requests
// that we generate from the beacon state.
// We do not make light client rpc requests from the beacon node
OutboundRequest::LightClientBootstrap(_) => vec![],
}
}
/* These functions are used in the handler for stream management */
@@ -110,28 +97,35 @@ impl<TSpec: EthSpec> OutboundRequest<TSpec> {
match self {
OutboundRequest::Status(_) => 1,
OutboundRequest::Goodbye(_) => 0,
OutboundRequest::BlocksByRange(req) => req.count,
OutboundRequest::BlocksByRoot(req) => req.block_roots.len() as u64,
OutboundRequest::BlocksByRange(req) => *req.count(),
OutboundRequest::BlocksByRoot(req) => req.block_roots().len() as u64,
OutboundRequest::BlobsByRange(req) => req.count * TSpec::max_blobs_per_block() as u64,
OutboundRequest::BlobsByRoot(req) => req.blob_ids.len() as u64,
OutboundRequest::Ping(_) => 1,
OutboundRequest::MetaData(_) => 1,
OutboundRequest::LightClientBootstrap(_) => 1,
}
}
/// Gives the corresponding `Protocol` to this request.
pub fn protocol(&self) -> Protocol {
/// Gives the corresponding `SupportedProtocol` to this request.
pub fn versioned_protocol(&self) -> SupportedProtocol {
match self {
OutboundRequest::Status(_) => Protocol::Status,
OutboundRequest::Goodbye(_) => Protocol::Goodbye,
OutboundRequest::BlocksByRange(_) => Protocol::BlocksByRange,
OutboundRequest::BlocksByRoot(_) => Protocol::BlocksByRoot,
OutboundRequest::BlobsByRange(_) => Protocol::BlobsByRange,
OutboundRequest::BlobsByRoot(_) => Protocol::BlobsByRoot,
OutboundRequest::Ping(_) => Protocol::Ping,
OutboundRequest::MetaData(_) => Protocol::MetaData,
OutboundRequest::LightClientBootstrap(_) => Protocol::LightClientBootstrap,
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::Ping(_) => SupportedProtocol::PingV1,
OutboundRequest::MetaData(req) => match req {
MetadataRequest::V1(_) => SupportedProtocol::MetaDataV1,
MetadataRequest::V2(_) => SupportedProtocol::MetaDataV2,
},
}
}
@@ -145,7 +139,6 @@ impl<TSpec: EthSpec> OutboundRequest<TSpec> {
OutboundRequest::BlocksByRoot(_) => ResponseTermination::BlocksByRoot,
OutboundRequest::BlobsByRange(_) => ResponseTermination::BlobsByRange,
OutboundRequest::BlobsByRoot(_) => ResponseTermination::BlobsByRoot,
OutboundRequest::LightClientBootstrap(_) => unreachable!(),
OutboundRequest::Status(_) => unreachable!(),
OutboundRequest::Goodbye(_) => unreachable!(),
OutboundRequest::Ping(_) => unreachable!(),
@@ -205,9 +198,6 @@ impl<TSpec: EthSpec> std::fmt::Display for OutboundRequest<TSpec> {
OutboundRequest::BlobsByRoot(req) => write!(f, "Blobs by root: {:?}", req),
OutboundRequest::Ping(ping) => write!(f, "Ping: {}", ping.data),
OutboundRequest::MetaData(_) => write!(f, "MetaData request"),
OutboundRequest::LightClientBootstrap(bootstrap) => {
write!(f, "Lightclient Bootstrap: {}", bootstrap.root)
}
}
}
}

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

@@ -205,35 +205,80 @@ pub enum Protocol {
LightClientBootstrap,
}
/// RPC Versions
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Version {
/// Version 1 of RPC
V1,
/// Version 2 of RPC
V2,
}
/// RPC Encondings supported.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Encoding {
SSZSnappy,
}
impl Protocol {
pub(crate) fn terminator(self) -> Option<ResponseTermination> {
/// All valid protocol name and version combinations.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum SupportedProtocol {
StatusV1,
GoodbyeV1,
BlocksByRangeV1,
BlocksByRangeV2,
BlocksByRootV1,
BlocksByRootV2,
BlobsByRangeV1,
BlobsByRootV1,
PingV1,
MetaDataV1,
MetaDataV2,
LightClientBootstrapV1,
}
impl SupportedProtocol {
pub fn version_string(&self) -> &'static str {
match self {
Protocol::Status => None,
Protocol::Goodbye => None,
Protocol::BlocksByRange => Some(ResponseTermination::BlocksByRange),
Protocol::BlocksByRoot => Some(ResponseTermination::BlocksByRoot),
Protocol::BlobsByRange => Some(ResponseTermination::BlobsByRange),
Protocol::BlobsByRoot => Some(ResponseTermination::BlobsByRoot),
Protocol::Ping => None,
Protocol::MetaData => None,
Protocol::LightClientBootstrap => None,
SupportedProtocol::StatusV1 => "1",
SupportedProtocol::GoodbyeV1 => "1",
SupportedProtocol::BlocksByRangeV1 => "1",
SupportedProtocol::BlocksByRangeV2 => "2",
SupportedProtocol::BlocksByRootV1 => "1",
SupportedProtocol::BlocksByRootV2 => "2",
SupportedProtocol::BlobsByRangeV1 => "1",
SupportedProtocol::BlobsByRootV1 => "1",
SupportedProtocol::PingV1 => "1",
SupportedProtocol::MetaDataV1 => "1",
SupportedProtocol::MetaDataV2 => "2",
SupportedProtocol::LightClientBootstrapV1 => "1",
}
}
pub fn protocol(&self) -> Protocol {
match self {
SupportedProtocol::StatusV1 => Protocol::Status,
SupportedProtocol::GoodbyeV1 => Protocol::Goodbye,
SupportedProtocol::BlocksByRangeV1 => Protocol::BlocksByRange,
SupportedProtocol::BlocksByRangeV2 => Protocol::BlocksByRange,
SupportedProtocol::BlocksByRootV1 => Protocol::BlocksByRoot,
SupportedProtocol::BlocksByRootV2 => Protocol::BlocksByRoot,
SupportedProtocol::BlobsByRangeV1 => Protocol::BlobsByRange,
SupportedProtocol::BlobsByRootV1 => Protocol::BlobsByRoot,
SupportedProtocol::PingV1 => Protocol::Ping,
SupportedProtocol::MetaDataV1 => Protocol::MetaData,
SupportedProtocol::MetaDataV2 => Protocol::MetaData,
SupportedProtocol::LightClientBootstrapV1 => Protocol::LightClientBootstrap,
}
}
fn currently_supported() -> Vec<ProtocolId> {
vec![
ProtocolId::new(Self::StatusV1, Encoding::SSZSnappy),
ProtocolId::new(Self::GoodbyeV1, Encoding::SSZSnappy),
// V2 variants have higher preference then V1
ProtocolId::new(Self::BlocksByRangeV2, Encoding::SSZSnappy),
ProtocolId::new(Self::BlocksByRangeV1, Encoding::SSZSnappy),
ProtocolId::new(Self::BlocksByRootV2, Encoding::SSZSnappy),
ProtocolId::new(Self::BlocksByRootV1, Encoding::SSZSnappy),
ProtocolId::new(Self::BlobsByRangeV1, Encoding::SSZSnappy),
ProtocolId::new(Self::BlobsByRootV1, Encoding::SSZSnappy),
ProtocolId::new(Self::PingV1, Encoding::SSZSnappy),
ProtocolId::new(Self::MetaDataV2, Encoding::SSZSnappy),
ProtocolId::new(Self::MetaDataV1, Encoding::SSZSnappy),
]
}
}
impl std::fmt::Display for Encoding {
@@ -245,16 +290,6 @@ impl std::fmt::Display for Encoding {
}
}
impl std::fmt::Display for Version {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let repr = match self {
Version::V1 => "1",
Version::V2 => "2",
};
f.write_str(repr)
}
}
#[derive(Debug, Clone)]
pub struct RPCProtocol<TSpec: EthSpec> {
pub fork_context: Arc<ForkContext>,
@@ -269,30 +304,17 @@ impl<TSpec: EthSpec> UpgradeInfo for RPCProtocol<TSpec> {
/// The list of supported RPC protocols for Lighthouse.
fn protocol_info(&self) -> Self::InfoIter {
let mut supported_protocols = vec![
ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(Protocol::Goodbye, Version::V1, Encoding::SSZSnappy),
// V2 variants have higher preference then V1
ProtocolId::new(Protocol::BlocksByRange, Version::V2, Encoding::SSZSnappy),
ProtocolId::new(Protocol::BlocksByRange, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(Protocol::BlocksByRoot, Version::V2, Encoding::SSZSnappy),
ProtocolId::new(Protocol::BlocksByRoot, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(Protocol::Ping, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(Protocol::MetaData, Version::V2, Encoding::SSZSnappy),
ProtocolId::new(Protocol::MetaData, Version::V1, Encoding::SSZSnappy),
];
let mut supported_protocols = SupportedProtocol::currently_supported();
if let ForkName::Deneb = self.fork_context.current_fork() {
supported_protocols.extend_from_slice(&[
ProtocolId::new(Protocol::BlobsByRoot, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(Protocol::BlobsByRange, Version::V1, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlobsByRootV1, Encoding::SSZSnappy),
ProtocolId::new(SupportedProtocol::BlobsByRangeV1, Encoding::SSZSnappy),
]);
}
if self.enable_light_client_server {
supported_protocols.push(ProtocolId::new(
Protocol::LightClientBootstrap,
Version::V1,
SupportedProtocol::LightClientBootstrapV1,
Encoding::SSZSnappy,
));
}
@@ -322,11 +344,8 @@ impl RpcLimits {
/// Tracks the types in a protocol id.
#[derive(Clone, Debug)]
pub struct ProtocolId {
/// The RPC message type/name.
pub message_name: Protocol,
/// The version of the RPC.
pub version: Version,
/// The protocol name and version
pub versioned_protocol: SupportedProtocol,
/// The encoding of the RPC.
pub encoding: Encoding,
@@ -338,7 +357,7 @@ pub struct ProtocolId {
impl ProtocolId {
/// Returns min and max size for messages of given protocol id requests.
pub fn rpc_request_limits(&self) -> RpcLimits {
match self.message_name {
match self.versioned_protocol.protocol() {
Protocol::Status => RpcLimits::new(
<StatusMessage as Encode>::ssz_fixed_len(),
<StatusMessage as Encode>::ssz_fixed_len(),
@@ -347,9 +366,10 @@ impl ProtocolId {
<GoodbyeReason as Encode>::ssz_fixed_len(),
<GoodbyeReason as Encode>::ssz_fixed_len(),
),
// V1 and V2 requests are the same
Protocol::BlocksByRange => RpcLimits::new(
<OldBlocksByRangeRequest as Encode>::ssz_fixed_len(),
<OldBlocksByRangeRequest as Encode>::ssz_fixed_len(),
<OldBlocksByRangeRequestV2 as Encode>::ssz_fixed_len(),
<OldBlocksByRangeRequestV2 as Encode>::ssz_fixed_len(),
),
Protocol::BlocksByRoot => {
RpcLimits::new(*BLOCKS_BY_ROOT_REQUEST_MIN, *BLOCKS_BY_ROOT_REQUEST_MAX)
@@ -376,7 +396,7 @@ impl ProtocolId {
/// Returns min and max size for messages of given protocol id responses.
pub fn rpc_response_limits<T: EthSpec>(&self, fork_context: &ForkContext) -> RpcLimits {
match self.message_name {
match self.versioned_protocol.protocol() {
Protocol::Status => RpcLimits::new(
<StatusMessage as Encode>::ssz_fixed_len(),
<StatusMessage as Encode>::ssz_fixed_len(),
@@ -407,31 +427,36 @@ impl ProtocolId {
/// Returns `true` if the given `ProtocolId` should expect `context_bytes` in the
/// beginning of the stream, else returns `false`.
pub fn has_context_bytes(&self) -> bool {
match self.message_name {
Protocol::BlocksByRange | Protocol::BlocksByRoot => match self.version {
Version::V2 => true,
Version::V1 => false,
},
Protocol::LightClientBootstrap => match self.version {
Version::V2 | Version::V1 => true,
},
Protocol::BlobsByRoot | Protocol::BlobsByRange => true,
Protocol::Goodbye | Protocol::Ping | Protocol::Status | Protocol::MetaData => false,
match self.versioned_protocol {
SupportedProtocol::BlocksByRangeV2
| SupportedProtocol::BlocksByRootV2
| SupportedProtocol::BlobsByRangeV1
| SupportedProtocol::BlobsByRootV1
| SupportedProtocol::LightClientBootstrapV1 => true,
SupportedProtocol::StatusV1
| SupportedProtocol::BlocksByRootV1
| SupportedProtocol::BlocksByRangeV1
| SupportedProtocol::PingV1
| SupportedProtocol::MetaDataV1
| SupportedProtocol::MetaDataV2
| SupportedProtocol::GoodbyeV1 => false,
}
}
}
/// An RPC protocol ID.
impl ProtocolId {
pub fn new(message_name: Protocol, version: Version, encoding: Encoding) -> Self {
pub fn new(versioned_protocol: SupportedProtocol, encoding: Encoding) -> Self {
let protocol_id = format!(
"{}/{}/{}/{}",
PROTOCOL_PREFIX, message_name, version, encoding
PROTOCOL_PREFIX,
versioned_protocol.protocol(),
versioned_protocol.version_string(),
encoding
);
ProtocolId {
message_name,
version,
versioned_protocol,
encoding,
protocol_id,
}
@@ -464,7 +489,7 @@ where
fn upgrade_inbound(self, socket: TSocket, protocol: ProtocolId) -> Self::Future {
async move {
let protocol_name = protocol.message_name;
let versioned_protocol = protocol.versioned_protocol;
// convert the socket to tokio compatible socket
let socket = socket.compat();
let codec = match protocol.encoding {
@@ -483,8 +508,13 @@ where
let socket = Framed::new(Box::pin(timed_socket), codec);
// MetaData requests should be empty, return the stream
match protocol_name {
Protocol::MetaData => Ok((InboundRequest::MetaData(PhantomData), socket)),
match versioned_protocol {
SupportedProtocol::MetaDataV1 => {
Ok((InboundRequest::MetaData(MetadataRequest::new_v1()), socket))
}
SupportedProtocol::MetaDataV2 => {
Ok((InboundRequest::MetaData(MetadataRequest::new_v2()), socket))
}
_ => {
match tokio::time::timeout(
Duration::from_secs(REQUEST_TIMEOUT),
@@ -514,7 +544,7 @@ pub enum InboundRequest<TSpec: EthSpec> {
BlobsByRoot(BlobsByRootRequest),
LightClientBootstrap(LightClientBootstrapRequest),
Ping(Ping),
MetaData(PhantomData<TSpec>),
MetaData(MetadataRequest<TSpec>),
}
/// Implements the encoding per supported protocol for `RPCRequest`.
@@ -526,8 +556,8 @@ impl<TSpec: EthSpec> InboundRequest<TSpec> {
match self {
InboundRequest::Status(_) => 1,
InboundRequest::Goodbye(_) => 0,
InboundRequest::BlocksByRange(req) => req.count,
InboundRequest::BlocksByRoot(req) => req.block_roots.len() as u64,
InboundRequest::BlocksByRange(req) => *req.count(),
InboundRequest::BlocksByRoot(req) => req.block_roots().len() as u64,
InboundRequest::BlobsByRange(req) => req.count * TSpec::max_blobs_per_block() as u64,
InboundRequest::BlobsByRoot(req) => req.blob_ids.len() as u64,
InboundRequest::Ping(_) => 1,
@@ -536,18 +566,27 @@ impl<TSpec: EthSpec> InboundRequest<TSpec> {
}
}
/// Gives the corresponding `Protocol` to this request.
pub fn protocol(&self) -> Protocol {
/// Gives the corresponding `SupportedProtocol` to this request.
pub fn versioned_protocol(&self) -> SupportedProtocol {
match self {
InboundRequest::Status(_) => Protocol::Status,
InboundRequest::Goodbye(_) => Protocol::Goodbye,
InboundRequest::BlocksByRange(_) => Protocol::BlocksByRange,
InboundRequest::BlocksByRoot(_) => Protocol::BlocksByRoot,
InboundRequest::BlobsByRange(_) => Protocol::BlobsByRange,
InboundRequest::BlobsByRoot(_) => Protocol::BlobsByRoot,
InboundRequest::Ping(_) => Protocol::Ping,
InboundRequest::MetaData(_) => Protocol::MetaData,
InboundRequest::LightClientBootstrap(_) => Protocol::LightClientBootstrap,
InboundRequest::Status(_) => SupportedProtocol::StatusV1,
InboundRequest::Goodbye(_) => SupportedProtocol::GoodbyeV1,
InboundRequest::BlocksByRange(req) => match req {
OldBlocksByRangeRequest::V1(_) => SupportedProtocol::BlocksByRangeV1,
OldBlocksByRangeRequest::V2(_) => SupportedProtocol::BlocksByRangeV2,
},
InboundRequest::BlocksByRoot(req) => match req {
BlocksByRootRequest::V1(_) => SupportedProtocol::BlocksByRootV1,
BlocksByRootRequest::V2(_) => SupportedProtocol::BlocksByRootV2,
},
InboundRequest::BlobsByRange(_) => SupportedProtocol::BlobsByRangeV1,
InboundRequest::BlobsByRoot(_) => SupportedProtocol::BlobsByRootV1,
InboundRequest::Ping(_) => SupportedProtocol::PingV1,
InboundRequest::MetaData(req) => match req {
MetadataRequest::V1(_) => SupportedProtocol::MetaDataV1,
MetadataRequest::V2(_) => SupportedProtocol::MetaDataV2,
},
InboundRequest::LightClientBootstrap(_) => SupportedProtocol::LightClientBootstrapV1,
}
}

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

@@ -214,7 +214,7 @@ pub trait RateLimiterItem {
impl<T: EthSpec> RateLimiterItem for super::InboundRequest<T> {
fn protocol(&self) -> Protocol {
self.protocol()
self.versioned_protocol().protocol()
}
fn expected_responses(&self) -> u64 {
@@ -224,7 +224,7 @@ impl<T: EthSpec> RateLimiterItem for super::InboundRequest<T> {
impl<T: EthSpec> RateLimiterItem for super::OutboundRequest<T> {
fn protocol(&self) -> Protocol {
self.protocol()
self.versioned_protocol().protocol()
}
fn expected_responses(&self) -> u64 {

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

@@ -72,7 +72,7 @@ impl<Id: ReqId, TSpec: EthSpec> SelfRateLimiter<Id, TSpec> {
request_id: Id,
req: OutboundRequest<TSpec>,
) -> Result<BehaviourAction<Id, TSpec>, Error> {
let protocol = req.protocol();
let protocol = req.versioned_protocol().protocol();
// First check that there are not already other requests waiting to be sent.
if let Some(queued_requests) = self.delayed_requests.get_mut(&(peer_id, protocol)) {
queued_requests.push_back(QueuedRequest { req, request_id });
@@ -111,7 +111,7 @@ impl<Id: ReqId, TSpec: EthSpec> SelfRateLimiter<Id, TSpec> {
event: RPCSend::Request(request_id, req),
}),
Err(e) => {
let protocol = req.protocol();
let protocol = req.versioned_protocol();
match e {
RateLimitedErr::TooLarge => {
// this should never happen with default parameters. Let's just send the request.
@@ -119,7 +119,7 @@ impl<Id: ReqId, TSpec: EthSpec> SelfRateLimiter<Id, TSpec> {
crit!(
log,
"Self rate limiting error for a batch that will never fit. Sending request anyway. Check configuration parameters.";
"protocol" => %req.protocol()
"protocol" => %req.versioned_protocol().protocol()
);
Ok(BehaviourAction::NotifyHandler {
peer_id,
@@ -128,7 +128,7 @@ impl<Id: ReqId, TSpec: EthSpec> SelfRateLimiter<Id, TSpec> {
})
}
RateLimitedErr::TooSoon(wait_time) => {
debug!(log, "Self rate limiting"; "protocol" => %protocol, "wait_time_ms" => wait_time.as_millis(), "peer_id" => %peer_id);
debug!(log, "Self rate limiting"; "protocol" => %protocol.protocol(), "wait_time_ms" => wait_time.as_millis(), "peer_id" => %peer_id);
Err((QueuedRequest { req, request_id }, wait_time))
}
}

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

@@ -8,7 +8,8 @@ use crate::rpc::methods::{BlobsByRangeRequest, BlobsByRootRequest};
use crate::rpc::{
methods::{
BlocksByRangeRequest, BlocksByRootRequest, LightClientBootstrapRequest,
OldBlocksByRangeRequest, RPCCodedResponse, RPCResponse, ResponseTermination, StatusMessage,
OldBlocksByRangeRequest, OldBlocksByRangeRequestV1, OldBlocksByRangeRequestV2,
RPCCodedResponse, RPCResponse, ResponseTermination, StatusMessage,
},
OutboundRequest, SubstreamId,
};
@@ -48,15 +49,26 @@ impl<TSpec: EthSpec> std::convert::From<Request> for OutboundRequest<TSpec> {
fn from(req: Request) -> OutboundRequest<TSpec> {
match req {
Request::BlocksByRoot(r) => OutboundRequest::BlocksByRoot(r),
Request::BlocksByRange(BlocksByRangeRequest { start_slot, count }) => {
OutboundRequest::BlocksByRange(OldBlocksByRangeRequest {
start_slot,
count,
step: 1,
})
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(_) => {
unreachable!("Lighthouse never makes an outbound light client request")
}
Request::BlobsByRange(r) => OutboundRequest::BlobsByRange(r),
Request::LightClientBootstrap(b) => OutboundRequest::LightClientBootstrap(b),
Request::BlobsByRoot(r) => OutboundRequest::BlobsByRoot(r),
Request::Status(s) => OutboundRequest::Status(s),
}

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

@@ -9,6 +9,7 @@ use crate::peer_manager::{
ConnectionDirection, PeerManager, PeerManagerEvent,
};
use crate::peer_manager::{MIN_OUTBOUND_ONLY_FACTOR, PEER_EXCESS_FACTOR, PRIORITY_PEER_EXCESS};
use crate::rpc::methods::MetadataRequest;
use crate::rpc::*;
use crate::service::behaviour::BehaviourEvent;
pub use crate::service::behaviour::Gossipsub;
@@ -37,7 +38,6 @@ use slog::{crit, debug, info, o, trace, warn};
use std::path::PathBuf;
use std::pin::Pin;
use std::{
marker::PhantomData,
sync::Arc,
task::{Context, Poll},
};
@@ -949,16 +949,25 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
/// Sends a METADATA request to a peer.
fn send_meta_data_request(&mut self, peer_id: PeerId) {
let event = OutboundRequest::MetaData(PhantomData);
// We always prefer sending V2 requests
let event = OutboundRequest::MetaData(MetadataRequest::new_v2());
self.eth2_rpc_mut()
.send_request(peer_id, RequestId::Internal, event);
}
/// Sends a METADATA response to a peer.
fn send_meta_data_response(&mut self, id: PeerRequestId, peer_id: PeerId) {
let event = RPCCodedResponse::Success(RPCResponse::MetaData(
self.network_globals.local_metadata.read().clone(),
));
fn send_meta_data_response(
&mut self,
req: MetadataRequest<TSpec>,
id: PeerRequestId,
peer_id: PeerId,
) {
let metadata = self.network_globals.local_metadata.read().clone();
let metadata = match req {
MetadataRequest::V1(_) => metadata.metadata_v1(),
MetadataRequest::V2(_) => metadata,
};
let event = RPCCodedResponse::Success(RPCResponse::MetaData(metadata));
self.eth2_rpc_mut().send_response(peer_id, id, event);
}
@@ -1207,9 +1216,9 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
self.pong(peer_request_id, peer_id);
None
}
InboundRequest::MetaData(_) => {
InboundRequest::MetaData(req) => {
// send the requested meta-data
self.send_meta_data_response((handler_id, id), peer_id);
self.send_meta_data_response(req, (handler_id, id), peer_id);
None
}
InboundRequest::Goodbye(reason) => {
@@ -1236,13 +1245,9 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
Some(event)
}
InboundRequest::BlocksByRange(req) => {
let methods::OldBlocksByRangeRequest {
start_slot,
mut count,
step,
} = req;
// Still disconnect the peer if the request is naughty.
if step == 0 {
let mut count = *req.count();
if *req.step() == 0 {
self.peer_manager_mut().handle_rpc_error(
&peer_id,
Protocol::BlocksByRange,
@@ -1254,14 +1259,18 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
return None;
}
// return just one block in case the step parameter is used. https://github.com/ethereum/consensus-specs/pull/2856
if step > 1 {
if *req.step() > 1 {
count = 1;
}
let event = self.build_request(
peer_request_id,
peer_id,
Request::BlocksByRange(BlocksByRangeRequest { start_slot, count }),
);
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) => {

8
beacon_node/lighthouse_network/src/service/utils.rs
View File

@@ -276,9 +276,11 @@ pub(crate) fn save_metadata_to_disk<E: EthSpec>(
log: &slog::Logger,
) {
let _ = std::fs::create_dir_all(dir);
match File::create(dir.join(METADATA_FILENAME))
.and_then(|mut f| f.write_all(&metadata.as_ssz_bytes()))
{
let metadata_bytes = match metadata {
MetaData::V1(md) => md.as_ssz_bytes(),
MetaData::V2(md) => md.as_ssz_bytes(),
};
match File::create(dir.join(METADATA_FILENAME)).and_then(|mut f| f.write_all(&metadata_bytes)) {
Ok(_) => {
debug!(log, "Metadata written to disk");
}