Upgrade to libp2p v0.50.0 (#3764)

I've needed to do this work in order to do some episub testing. 

This version of libp2p has not yet been released, so this is left as a draft for when we wish to update.

Co-authored-by: Diva M <divma@protonmail.com>
This commit is contained in:
Age Manning 2023-01-06 15:59:33 +00:00
parent 4e5e7ee1fc
commit 1d9a2022b4
16 changed files with 1953 additions and 1180 deletions

View File

@ -12,7 +12,7 @@ env:
# Deny warnings in CI
RUSTFLAGS: "-D warnings"
# The Nightly version used for cargo-udeps, might need updating from time to time.
PINNED_NIGHTLY: nightly-2022-05-20
PINNED_NIGHTLY: nightly-2022-12-15
# Prevent Github API rate limiting.
LIGHTHOUSE_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
jobs:
@ -332,27 +332,6 @@ jobs:
run: make lint
- name: Certify Cargo.lock freshness
run: git diff --exit-code Cargo.lock
disallowed-from-async-lint:
name: disallowed-from-async-lint
runs-on: ubuntu-latest
needs: cargo-fmt
continue-on-error: true
steps:
- uses: actions/checkout@v3
- name: Install SigP Clippy fork
run: |
cd ..
git clone https://github.com/michaelsproul/rust-clippy.git
cd rust-clippy
git checkout 31a49666ccfcd7963b63345d6ce757c373f22c2a
cargo build --release --bin cargo-clippy --bin clippy-driver
cargo build --release --bin cargo-clippy --bin clippy-driver -Zunstable-options --out-dir $(rustc --print=sysroot)/bin
- name: Install Protoc
uses: arduino/setup-protoc@e52d9eb8f7b63115df1ac544a1376fdbf5a39612
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
- name: Run Clippy with the disallowed-from-async lint
run: make nightly-lint
check-msrv:
name: check-msrv
runs-on: ubuntu-latest

2135
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -7,7 +7,13 @@ use eth2::{BeaconNodeHttpClient, Timeouts};
use http_api::{Config, Context};
use lighthouse_network::{
discv5::enr::{CombinedKey, EnrBuilder},
libp2p::{core::connection::ConnectionId, swarm::NetworkBehaviour},
libp2p::{
core::connection::ConnectionId,
swarm::{
behaviour::{ConnectionEstablished, FromSwarm},
NetworkBehaviour,
},
},
rpc::methods::{MetaData, MetaDataV2},
types::{EnrAttestationBitfield, EnrSyncCommitteeBitfield, SyncState},
ConnectedPoint, Enr, NetworkGlobals, PeerId, PeerManager,
@ -143,12 +149,18 @@ pub async fn create_api_server_on_port<T: BeaconChainTypes>(
// add a peer
let peer_id = PeerId::random();
let connected_point = ConnectedPoint::Listener {
let endpoint = &ConnectedPoint::Listener {
local_addr: EXTERNAL_ADDR.parse().unwrap(),
send_back_addr: EXTERNAL_ADDR.parse().unwrap(),
};
let con_id = ConnectionId::new(1);
pm.inject_connection_established(&peer_id, &con_id, &connected_point, None, 0);
let connection_id = ConnectionId::new(1);
pm.on_swarm_event(FromSwarm::ConnectionEstablished(ConnectionEstablished {
peer_id,
connection_id,
endpoint,
failed_addresses: &[],
other_established: 0,
}));
*network_globals.sync_state.write() = SyncState::Synced;
let eth1_service =

View File

@ -40,11 +40,12 @@ superstruct = "0.5.0"
prometheus-client = "0.18.0"
unused_port = { path = "../../common/unused_port" }
delay_map = "0.1.1"
void = "1"
[dependencies.libp2p]
version = "0.48.0"
version = "0.50.0"
default-features = false
features = ["websocket", "identify", "mplex", "yamux", "noise", "gossipsub", "dns-tokio", "tcp-tokio", "plaintext", "secp256k1"]
features = ["websocket", "identify", "mplex", "yamux", "noise", "gossipsub", "dns", "tcp", "tokio", "plaintext", "secp256k1", "macros", "ecdsa"]
[dev-dependencies]
slog-term = "2.6.0"

View File

@ -22,12 +22,13 @@ use enr::{ATTESTATION_BITFIELD_ENR_KEY, ETH2_ENR_KEY, SYNC_COMMITTEE_BITFIELD_EN
use futures::prelude::*;
use futures::stream::FuturesUnordered;
use libp2p::multiaddr::Protocol;
use libp2p::swarm::behaviour::{DialFailure, FromSwarm};
use libp2p::swarm::AddressScore;
pub use libp2p::{
core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId},
swarm::{
handler::ConnectionHandler, DialError, NetworkBehaviour,
NetworkBehaviourAction as NBAction, NotifyHandler, PollParameters, SubstreamProtocol,
dummy::ConnectionHandler, DialError, NetworkBehaviour, NetworkBehaviourAction as NBAction,
NotifyHandler, PollParameters, SubstreamProtocol,
},
};
use lru::LruCache;
@ -927,11 +928,11 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
// Discovery is not a real NetworkBehaviour...
type ConnectionHandler = libp2p::swarm::handler::DummyConnectionHandler;
type ConnectionHandler = ConnectionHandler;
type OutEvent = DiscoveredPeers;
fn new_handler(&mut self) -> Self::ConnectionHandler {
libp2p::swarm::handler::DummyConnectionHandler::default()
ConnectionHandler
}
// Handles the libp2p request to obtain multiaddrs for peer_id's in order to dial them.
@ -947,40 +948,6 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
}
}
fn inject_event(
&mut self,
_: PeerId,
_: ConnectionId,
_: <Self::ConnectionHandler as ConnectionHandler>::OutEvent,
) {
}
fn inject_dial_failure(
&mut self,
peer_id: Option<PeerId>,
_handler: Self::ConnectionHandler,
error: &DialError,
) {
if let Some(peer_id) = peer_id {
match error {
DialError::Banned
| DialError::LocalPeerId
| DialError::InvalidPeerId(_)
| DialError::ConnectionIo(_)
| DialError::NoAddresses
| DialError::Transport(_)
| DialError::WrongPeerId { .. } => {
// set peer as disconnected in discovery DHT
debug!(self.log, "Marking peer disconnected in DHT"; "peer_id" => %peer_id);
self.disconnect_peer(&peer_id);
}
DialError::ConnectionLimit(_)
| DialError::DialPeerConditionFalse(_)
| DialError::Aborted => {}
}
}
}
// Main execution loop to drive the behaviour
fn poll(
&mut self,
@ -1067,6 +1034,50 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
}
Poll::Pending
}
fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
match event {
FromSwarm::DialFailure(DialFailure { peer_id, error, .. }) => {
self.on_dial_failure(peer_id, error)
}
FromSwarm::ConnectionEstablished(_)
| FromSwarm::ConnectionClosed(_)
| FromSwarm::AddressChange(_)
| FromSwarm::ListenFailure(_)
| FromSwarm::NewListener(_)
| FromSwarm::NewListenAddr(_)
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {
// Ignore events not relevant to discovery
}
}
}
}
impl<TSpec: EthSpec> Discovery<TSpec> {
fn on_dial_failure(&mut self, peer_id: Option<PeerId>, error: &DialError) {
if let Some(peer_id) = peer_id {
match error {
DialError::Banned
| DialError::LocalPeerId
| DialError::InvalidPeerId(_)
| DialError::ConnectionIo(_)
| DialError::NoAddresses
| DialError::Transport(_)
| DialError::WrongPeerId { .. } => {
// set peer as disconnected in discovery DHT
debug!(self.log, "Marking peer disconnected in DHT"; "peer_id" => %peer_id);
self.disconnect_peer(&peer_id);
}
DialError::ConnectionLimit(_)
| DialError::DialPeerConditionFalse(_)
| DialError::Aborted => {}
}
}
}
}
#[cfg(test)]

View File

@ -7,7 +7,7 @@ use crate::{NetworkGlobals, PeerId};
use crate::{Subnet, SubnetDiscovery};
use delay_map::HashSetDelay;
use discv5::Enr;
use libp2p::identify::IdentifyInfo;
use libp2p::identify::Info as IdentifyInfo;
use peerdb::{client::ClientKind, BanOperation, BanResult, ScoreUpdateResult};
use rand::seq::SliceRandom;
use slog::{debug, error, trace, warn};

View File

@ -1,14 +1,12 @@
use std::task::{Context, Poll};
use futures::StreamExt;
use libp2p::core::connection::ConnectionId;
use libp2p::core::ConnectedPoint;
use libp2p::swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm};
use libp2p::swarm::dial_opts::{DialOpts, PeerCondition};
use libp2p::swarm::handler::DummyConnectionHandler;
use libp2p::swarm::{
ConnectionHandler, DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
};
use libp2p::{Multiaddr, PeerId};
use libp2p::swarm::dummy::ConnectionHandler;
use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use libp2p::PeerId;
use slog::{debug, error};
use types::EthSpec;
@ -20,23 +18,14 @@ use super::peerdb::BanResult;
use super::{ConnectingType, PeerManager, PeerManagerEvent, ReportSource};
impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
type ConnectionHandler = DummyConnectionHandler;
type ConnectionHandler = ConnectionHandler;
type OutEvent = PeerManagerEvent;
/* Required trait members */
fn new_handler(&mut self) -> Self::ConnectionHandler {
DummyConnectionHandler::default()
}
fn inject_event(
&mut self,
_: PeerId,
_: ConnectionId,
_: <DummyConnectionHandler as ConnectionHandler>::OutEvent,
) {
unreachable!("Dummy handler does not emit events")
ConnectionHandler
}
fn poll(
@ -114,19 +103,46 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
Poll::Pending
}
/* Overwritten trait members */
fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
match event {
FromSwarm::ConnectionEstablished(ConnectionEstablished {
peer_id,
endpoint,
other_established,
..
}) => self.on_connection_established(peer_id, endpoint, other_established),
FromSwarm::ConnectionClosed(ConnectionClosed {
peer_id,
remaining_established,
..
}) => self.on_connection_closed(peer_id, remaining_established),
FromSwarm::DialFailure(DialFailure { peer_id, .. }) => self.on_dial_failure(peer_id),
FromSwarm::AddressChange(_)
| FromSwarm::ListenFailure(_)
| FromSwarm::NewListener(_)
| FromSwarm::NewListenAddr(_)
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {
// The rest of the events we ignore since they are handled in their associated
// `SwarmEvent`
}
}
}
}
fn inject_connection_established(
impl<TSpec: EthSpec> PeerManager<TSpec> {
fn on_connection_established(
&mut self,
peer_id: &PeerId,
_connection_id: &ConnectionId,
peer_id: PeerId,
endpoint: &ConnectedPoint,
_failed_addresses: Option<&Vec<Multiaddr>>,
other_established: usize,
) {
debug!(self.log, "Connection established"; "peer_id" => %peer_id, "connection" => ?endpoint.to_endpoint());
if other_established == 0 {
self.events.push(PeerManagerEvent::MetaData(*peer_id));
self.events.push(PeerManagerEvent::MetaData(peer_id));
}
// Check NAT if metrics are enabled
@ -135,20 +151,20 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
}
// Check to make sure the peer is not supposed to be banned
match self.ban_status(peer_id) {
match self.ban_status(&peer_id) {
// TODO: directly emit the ban event?
BanResult::BadScore => {
// This is a faulty state
error!(self.log, "Connected to a banned peer. Re-banning"; "peer_id" => %peer_id);
// Reban the peer
self.goodbye_peer(peer_id, GoodbyeReason::Banned, ReportSource::PeerManager);
self.goodbye_peer(&peer_id, GoodbyeReason::Banned, ReportSource::PeerManager);
return;
}
BanResult::BannedIp(ip_addr) => {
// A good peer has connected to us via a banned IP address. We ban the peer and
// prevent future connections.
debug!(self.log, "Peer connected via banned IP. Banning"; "peer_id" => %peer_id, "banned_ip" => %ip_addr);
self.goodbye_peer(peer_id, GoodbyeReason::BannedIP, ReportSource::PeerManager);
self.goodbye_peer(&peer_id, GoodbyeReason::BannedIP, ReportSource::PeerManager);
return;
}
BanResult::NotBanned => {}
@ -162,11 +178,11 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
.network_globals
.peers
.read()
.peer_info(peer_id)
.peer_info(&peer_id)
.map_or(true, |peer| !peer.has_future_duty())
{
// Gracefully disconnect the peer.
self.disconnect_peer(*peer_id, GoodbyeReason::TooManyPeers);
self.disconnect_peer(peer_id, GoodbyeReason::TooManyPeers);
return;
}
@ -174,14 +190,14 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
// does not need to know about these peers.
match endpoint {
ConnectedPoint::Listener { send_back_addr, .. } => {
self.inject_connect_ingoing(peer_id, send_back_addr.clone(), None);
self.inject_connect_ingoing(&peer_id, send_back_addr.clone(), None);
self.events
.push(PeerManagerEvent::PeerConnectedIncoming(*peer_id));
.push(PeerManagerEvent::PeerConnectedIncoming(peer_id));
}
ConnectedPoint::Dialer { address, .. } => {
self.inject_connect_outgoing(peer_id, address.clone(), None);
self.inject_connect_outgoing(&peer_id, address.clone(), None);
self.events
.push(PeerManagerEvent::PeerConnectedOutgoing(*peer_id));
.push(PeerManagerEvent::PeerConnectedOutgoing(peer_id));
}
}
@ -189,14 +205,8 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
self.update_connected_peer_metrics();
metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT);
}
fn inject_connection_closed(
&mut self,
peer_id: &PeerId,
_: &ConnectionId,
_: &ConnectedPoint,
_: DummyConnectionHandler,
remaining_established: usize,
) {
fn on_connection_closed(&mut self, peer_id: PeerId, remaining_established: usize) {
if remaining_established > 0 {
return;
}
@ -206,62 +216,33 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
.network_globals
.peers
.read()
.is_connected_or_disconnecting(peer_id)
.is_connected_or_disconnecting(&peer_id)
{
// We are disconnecting the peer or the peer has already been connected.
// Both these cases, the peer has been previously registered by the peer manager and
// potentially the application layer.
// Inform the application.
self.events
.push(PeerManagerEvent::PeerDisconnected(*peer_id));
.push(PeerManagerEvent::PeerDisconnected(peer_id));
debug!(self.log, "Peer disconnected"; "peer_id" => %peer_id);
}
// NOTE: It may be the case that a rejected node, due to too many peers is disconnected
// here and the peer manager has no knowledge of its connection. We insert it here for
// reference so that peer manager can track this peer.
self.inject_disconnect(peer_id);
self.inject_disconnect(&peer_id);
// Update the prometheus metrics
self.update_connected_peer_metrics();
metrics::inc_counter(&metrics::PEER_DISCONNECT_EVENT_COUNT);
}
fn inject_address_change(
&mut self,
_peer_id: &PeerId,
_connection_id: &ConnectionId,
old: &ConnectedPoint,
new: &ConnectedPoint,
) {
debug_assert!(
matches!(
(old, new),
(
// inbound remains inbound
ConnectedPoint::Listener { .. },
ConnectedPoint::Listener { .. }
) | (
// outbound remains outbound
ConnectedPoint::Dialer { .. },
ConnectedPoint::Dialer { .. }
)
),
"A peer has changed between inbound and outbound"
)
}
/// A dial attempt has failed.
///
/// NOTE: It can be the case that we are dialing a peer and during the dialing process the peer
/// connects and the dial attempt later fails. To handle this, we only update the peer_db if
/// the peer is not already connected.
fn inject_dial_failure(
&mut self,
peer_id: Option<PeerId>,
_handler: DummyConnectionHandler,
_error: &DialError,
) {
fn on_dial_failure(&mut self, peer_id: Option<PeerId>) {
if let Some(peer_id) = peer_id {
if !self.network_globals.peers.read().is_connected(&peer_id) {
self.inject_disconnect(&peer_id);

View File

@ -2,7 +2,7 @@
//!
//! Currently using identify to fingerprint.
use libp2p::identify::IdentifyInfo;
use libp2p::identify::Info as IdentifyInfo;
use serde::Serialize;
use strum::{AsRefStr, EnumIter, IntoStaticStr};

View File

@ -7,8 +7,8 @@ use libp2p::gossipsub::subscription_filter::{
MaxCountSubscriptionFilter, WhitelistSubscriptionFilter,
};
use libp2p::gossipsub::Gossipsub as BaseGossipsub;
use libp2p::identify::Identify;
use libp2p::NetworkBehaviour;
use libp2p::identify::Behaviour as Identify;
use libp2p::swarm::NetworkBehaviour;
use types::EthSpec;
use super::api_types::RequestId;

View File

@ -26,7 +26,7 @@ use libp2p::gossipsub::subscription_filter::MaxCountSubscriptionFilter;
use libp2p::gossipsub::{
GossipsubEvent, IdentTopic as Topic, MessageAcceptance, MessageAuthenticity, MessageId,
};
use libp2p::identify::{Identify, IdentifyConfig, IdentifyEvent};
use libp2p::identify::{Behaviour as Identify, Config as IdentifyConfig, Event as IdentifyEvent};
use libp2p::multiaddr::{Multiaddr, Protocol as MProtocol};
use libp2p::swarm::{ConnectionLimits, Swarm, SwarmBuilder, SwarmEvent};
use libp2p::PeerId;
@ -316,7 +316,7 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
// use the executor for libp2p
struct Executor(task_executor::TaskExecutor);
impl libp2p::core::Executor for Executor {
impl libp2p::swarm::Executor for Executor {
fn exec(&self, f: Pin<Box<dyn futures::Future<Output = ()> + Send>>) {
self.0.spawn(f, "libp2p");
}
@ -341,12 +341,16 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
.with_max_established_per_peer(Some(MAX_CONNECTIONS_PER_PEER));
(
SwarmBuilder::new(transport, behaviour, local_peer_id)
.notify_handler_buffer_size(std::num::NonZeroUsize::new(7).expect("Not zero"))
.connection_event_buffer_size(64)
.connection_limits(limits)
.executor(Box::new(Executor(executor)))
.build(),
SwarmBuilder::with_executor(
transport,
behaviour,
local_peer_id,
Executor(executor),
)
.notify_handler_buffer_size(std::num::NonZeroUsize::new(7).expect("Not zero"))
.connection_event_buffer_size(64)
.connection_limits(limits)
.build(),
bandwidth,
)
};

View File

@ -44,8 +44,7 @@ type BoxedTransport = Boxed<(PeerId, StreamMuxerBox)>;
pub fn build_transport(
local_private_key: Keypair,
) -> std::io::Result<(BoxedTransport, Arc<BandwidthSinks>)> {
let tcp =
libp2p::tcp::TokioTcpTransport::new(libp2p::tcp::GenTcpConfig::default().nodelay(true));
let tcp = libp2p::tcp::tokio::Transport::new(libp2p::tcp::Config::default().nodelay(true));
let transport = libp2p::dns::TokioDnsConfig::system(tcp)?;
#[cfg(feature = "libp2p-websocket")]
let transport = {

View File

@ -15,13 +15,6 @@ use types::{
};
use unused_port::unused_tcp_port;
#[allow(clippy::type_complexity)]
#[allow(unused)]
pub mod behaviour;
#[allow(clippy::type_complexity)]
#[allow(unused)]
pub mod swarm;
type E = MinimalEthSpec;
type ReqId = usize;

View File

@ -1,395 +0,0 @@
// NOTE: Taken from libp2p's swarm's testing utils.
//
// Copyright 2020 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use std::collections::HashMap;
use std::task::{Context, Poll};
use libp2p::core::connection::{ConnectedPoint, ConnectionId};
use libp2p::core::transport::ListenerId;
use libp2p::swarm::handler::{ConnectionHandler, DummyConnectionHandler, IntoConnectionHandler};
use libp2p::swarm::{DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use libp2p::{Multiaddr, PeerId};
/// A `MockBehaviour` is a `NetworkBehaviour` that allows for
/// the instrumentation of return values, without keeping
/// any further state.
pub struct MockBehaviour<
THandler = DummyConnectionHandler,
TOutEvent = <DummyConnectionHandler as ConnectionHandler>::OutEvent,
> where
THandler: ConnectionHandler,
{
/// The prototype protocols handler that is cloned for every
/// invocation of `new_handler`.
pub handler_proto: THandler,
/// The addresses to return from `addresses_of_peer`.
pub addresses: HashMap<PeerId, Vec<Multiaddr>>,
/// The next action to return from `poll`.
///
/// An action is only returned once.
pub next_action: Option<NetworkBehaviourAction<TOutEvent, THandler>>,
}
impl<THandler, TOutEvent> MockBehaviour<THandler, TOutEvent>
where
THandler: ConnectionHandler,
{
pub fn new(handler_proto: THandler) -> Self {
MockBehaviour {
handler_proto,
addresses: HashMap::new(),
next_action: None,
}
}
}
impl<THandler, TOutEvent> NetworkBehaviour for MockBehaviour<THandler, TOutEvent>
where
THandler: ConnectionHandler + Clone,
THandler::OutEvent: Clone,
TOutEvent: Send + 'static,
{
type ConnectionHandler = THandler;
type OutEvent = TOutEvent;
fn new_handler(&mut self) -> Self::ConnectionHandler {
self.handler_proto.clone()
}
fn addresses_of_peer(&mut self, p: &PeerId) -> Vec<Multiaddr> {
self.addresses.get(p).map_or(Vec::new(), |v| v.clone())
}
fn inject_event(&mut self, _: PeerId, _: ConnectionId, _: THandler::OutEvent) {}
fn poll(
&mut self,
_: &mut Context,
_: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
Option::take(&mut self.next_action).map_or(Poll::Pending, Poll::Ready)
}
}
/// A `CallTraceBehaviour` is a `NetworkBehaviour` that tracks invocations of callback methods and
/// their arguments, wrapping around an inner behaviour. It ensures certain invariants are met.
pub struct CallTraceBehaviour<TInner>
where
TInner: NetworkBehaviour,
{
inner: TInner,
pub addresses_of_peer: Vec<PeerId>,
pub inject_connection_established: Vec<(PeerId, ConnectionId, ConnectedPoint, usize)>,
pub inject_connection_closed: Vec<(PeerId, ConnectionId, ConnectedPoint, usize)>,
pub inject_event: Vec<(
PeerId,
ConnectionId,
<<TInner::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent,
)>,
pub inject_dial_failure: Vec<Option<PeerId>>,
pub inject_new_listener: Vec<ListenerId>,
pub inject_new_listen_addr: Vec<(ListenerId, Multiaddr)>,
pub inject_new_external_addr: Vec<Multiaddr>,
pub inject_expired_listen_addr: Vec<(ListenerId, Multiaddr)>,
pub inject_expired_external_addr: Vec<Multiaddr>,
pub inject_listener_error: Vec<ListenerId>,
pub inject_listener_closed: Vec<(ListenerId, bool)>,
pub poll: usize,
}
impl<TInner> CallTraceBehaviour<TInner>
where
TInner: NetworkBehaviour,
{
pub fn new(inner: TInner) -> Self {
Self {
inner,
addresses_of_peer: Vec::new(),
inject_connection_established: Vec::new(),
inject_connection_closed: Vec::new(),
inject_event: Vec::new(),
inject_dial_failure: Vec::new(),
inject_new_listener: Vec::new(),
inject_new_listen_addr: Vec::new(),
inject_new_external_addr: Vec::new(),
inject_expired_listen_addr: Vec::new(),
inject_expired_external_addr: Vec::new(),
inject_listener_error: Vec::new(),
inject_listener_closed: Vec::new(),
poll: 0,
}
}
#[allow(dead_code)]
pub fn reset(&mut self) {
self.addresses_of_peer = Vec::new();
self.inject_connection_established = Vec::new();
self.inject_connection_closed = Vec::new();
self.inject_event = Vec::new();
self.inject_dial_failure = Vec::new();
self.inject_new_listen_addr = Vec::new();
self.inject_new_external_addr = Vec::new();
self.inject_expired_listen_addr = Vec::new();
self.inject_listener_error = Vec::new();
self.inject_listener_closed = Vec::new();
self.poll = 0;
}
pub fn inner(&mut self) -> &mut TInner {
&mut self.inner
}
/// Checks that when the expected number of closed connection notifications are received, a
/// given number of expected disconnections have been received as well.
///
/// Returns if the first condition is met.
pub fn assert_disconnected(
&self,
expected_closed_connections: usize,
expected_disconnections: usize,
) -> bool {
if self.inject_connection_closed.len() == expected_closed_connections {
assert_eq!(
self.inject_connection_closed
.iter()
.filter(|(.., remaining_established)| { *remaining_established == 0 })
.count(),
expected_disconnections
);
return true;
}
false
}
/// Checks that when the expected number of established connection notifications are received,
/// a given number of expected connections have been received as well.
///
/// Returns if the first condition is met.
pub fn assert_connected(
&self,
expected_established_connections: usize,
expected_connections: usize,
) -> bool {
if self.inject_connection_established.len() == expected_established_connections {
assert_eq!(
self.inject_connection_established
.iter()
.filter(|(.., reported_aditional_connections)| {
*reported_aditional_connections == 0
})
.count(),
expected_connections
);
return true;
}
false
}
}
impl<TInner> NetworkBehaviour for CallTraceBehaviour<TInner>
where
TInner: NetworkBehaviour,
<<TInner::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent:
Clone,
{
type ConnectionHandler = TInner::ConnectionHandler;
type OutEvent = TInner::OutEvent;
fn new_handler(&mut self) -> Self::ConnectionHandler {
self.inner.new_handler()
}
fn addresses_of_peer(&mut self, p: &PeerId) -> Vec<Multiaddr> {
self.addresses_of_peer.push(*p);
self.inner.addresses_of_peer(p)
}
fn inject_connection_established(
&mut self,
p: &PeerId,
c: &ConnectionId,
e: &ConnectedPoint,
errors: Option<&Vec<Multiaddr>>,
other_established: usize,
) {
let mut other_peer_connections = self
.inject_connection_established
.iter()
.rev() // take last to first
.filter_map(|(peer, .., other_established)| {
if p == peer {
Some(other_established)
} else {
None
}
})
.take(other_established);
// We are informed that there are `other_established` additional connections. Ensure that the
// number of previous connections is consistent with this
if let Some(&prev) = other_peer_connections.next() {
if prev < other_established {
assert_eq!(
prev,
other_established - 1,
"Inconsistent connection reporting"
)
}
assert_eq!(other_peer_connections.count(), other_established - 1);
} else {
assert_eq!(other_established, 0)
}
self.inject_connection_established
.push((*p, *c, e.clone(), other_established));
self.inner
.inject_connection_established(p, c, e, errors, other_established);
}
fn inject_connection_closed(
&mut self,
p: &PeerId,
c: &ConnectionId,
e: &ConnectedPoint,
handler: <Self::ConnectionHandler as IntoConnectionHandler>::Handler,
remaining_established: usize,
) {
let mut other_closed_connections = self
.inject_connection_established
.iter()
.rev() // take last to first
.filter_map(|(peer, .., remaining_established)| {
if p == peer {
Some(remaining_established)
} else {
None
}
})
.take(remaining_established);
// We are informed that there are `other_established` additional connections. Ensure that the
// number of previous connections is consistent with this
if let Some(&prev) = other_closed_connections.next() {
if prev < remaining_established {
assert_eq!(
prev,
remaining_established - 1,
"Inconsistent closed connection reporting"
)
}
assert_eq!(other_closed_connections.count(), remaining_established - 1);
} else {
assert_eq!(remaining_established, 0)
}
assert!(
self.inject_connection_established
.iter()
.any(|(peer, conn_id, endpoint, _)| (peer, conn_id, endpoint) == (p, c, e)),
"`inject_connection_closed` is called only for connections for \
which `inject_connection_established` was called first."
);
self.inject_connection_closed
.push((*p, *c, e.clone(), remaining_established));
self.inner
.inject_connection_closed(p, c, e, handler, remaining_established);
}
fn inject_event(
&mut self,
p: PeerId,
c: ConnectionId,
e: <<Self::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent,
) {
assert!(
self.inject_connection_established
.iter()
.any(|(peer_id, conn_id, ..)| *peer_id == p && c == *conn_id),
"`inject_event` is called for reported connections."
);
assert!(
!self
.inject_connection_closed
.iter()
.any(|(peer_id, conn_id, ..)| *peer_id == p && c == *conn_id),
"`inject_event` is never called for closed connections."
);
self.inject_event.push((p, c, e.clone()));
self.inner.inject_event(p, c, e);
}
fn inject_dial_failure(
&mut self,
p: Option<PeerId>,
handler: Self::ConnectionHandler,
error: &DialError,
) {
self.inject_dial_failure.push(p);
self.inner.inject_dial_failure(p, handler, error);
}
fn inject_new_listener(&mut self, id: ListenerId) {
self.inject_new_listener.push(id);
self.inner.inject_new_listener(id);
}
fn inject_new_listen_addr(&mut self, id: ListenerId, a: &Multiaddr) {
self.inject_new_listen_addr.push((id, a.clone()));
self.inner.inject_new_listen_addr(id, a);
}
fn inject_expired_listen_addr(&mut self, id: ListenerId, a: &Multiaddr) {
self.inject_expired_listen_addr.push((id, a.clone()));
self.inner.inject_expired_listen_addr(id, a);
}
fn inject_new_external_addr(&mut self, a: &Multiaddr) {
self.inject_new_external_addr.push(a.clone());
self.inner.inject_new_external_addr(a);
}
fn inject_expired_external_addr(&mut self, a: &Multiaddr) {
self.inject_expired_external_addr.push(a.clone());
self.inner.inject_expired_external_addr(a);
}
fn inject_listener_error(&mut self, l: ListenerId, e: &(dyn std::error::Error + 'static)) {
self.inject_listener_error.push(l);
self.inner.inject_listener_error(l, e);
}
fn inject_listener_closed(&mut self, l: ListenerId, r: Result<(), &std::io::Error>) {
self.inject_listener_closed.push((l, r.is_ok()));
self.inner.inject_listener_closed(l, r);
}
fn poll(
&mut self,
cx: &mut Context,
args: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
self.poll += 1;
self.inner.poll(cx, args)
}
}

View File

@ -1,99 +0,0 @@
use std::collections::HashMap;
use std::pin::Pin;
use super::behaviour::{CallTraceBehaviour, MockBehaviour};
use futures::stream::Stream;
use futures::task::{Context, Poll};
use libp2p::swarm::handler::ConnectionHandler;
use libp2p::swarm::{IntoConnectionHandler, NetworkBehaviour, Swarm, SwarmBuilder, SwarmEvent};
use libp2p::{PeerId, Transport};
use futures::StreamExt;
pub fn new_test_swarm<B>(behaviour: B) -> Swarm<B>
where
B: NetworkBehaviour,
{
let id_keys = libp2p::identity::Keypair::generate_ed25519();
let local_public_key = id_keys.public();
let transport = libp2p::core::transport::MemoryTransport::default()
.upgrade(libp2p::core::upgrade::Version::V1)
.authenticate(libp2p::plaintext::PlainText2Config {
local_public_key: local_public_key.clone(),
})
.multiplex(libp2p::yamux::YamuxConfig::default())
.boxed();
SwarmBuilder::new(transport, behaviour, local_public_key.into()).build()
}
pub fn random_multiaddr() -> libp2p::multiaddr::Multiaddr {
libp2p::multiaddr::Protocol::Memory(rand::random::<u64>()).into()
}
/// Bind a memory multiaddr to a compatible swarm.
pub async fn bind_listener<B: NetworkBehaviour>(
swarm: &mut Swarm<B>,
) -> libp2p::multiaddr::Multiaddr {
swarm.listen_on(random_multiaddr()).unwrap();
match swarm.select_next_some().await {
SwarmEvent::NewListenAddr {
listener_id: _,
address,
} => address,
_ => panic!("Testing swarm's first event should be a new listener"),
}
}
#[derive(Default)]
pub struct SwarmPool<B: NetworkBehaviour> {
swarms: HashMap<PeerId, Swarm<B>>,
}
impl<B: NetworkBehaviour> SwarmPool<B> {
pub fn with_capacity(capacity: usize) -> Self {
Self {
swarms: HashMap::with_capacity(capacity),
}
}
pub fn insert(&mut self, swarm: Swarm<B>) -> PeerId {
let peer_id = *swarm.local_peer_id();
self.swarms.insert(peer_id, swarm);
peer_id
}
pub fn remove(&mut self, peer_id: &PeerId) {
self.swarms.remove(peer_id);
}
pub fn get_mut(&mut self, peer_id: &PeerId) -> Option<&mut Swarm<B>> {
self.swarms.get_mut(peer_id)
}
pub fn swarms(&self) -> &HashMap<PeerId, Swarm<B>> {
&self.swarms
}
pub fn swarms_mut(&mut self) -> &mut HashMap<PeerId, Swarm<B>> {
&mut self.swarms
}
}
impl<B> Stream for SwarmPool<B>
where
B: NetworkBehaviour,
<B as NetworkBehaviour>::ConnectionHandler: ConnectionHandler,
{
type Item = (PeerId,
SwarmEvent<<B as NetworkBehaviour>::OutEvent, <<<B as NetworkBehaviour>::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::Error>);
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut polls = self
.get_mut()
.swarms
.iter_mut()
.map(|(&peer_id, swarm)| swarm.map(move |ev| (peer_id, ev)))
.collect::<futures::stream::SelectAll<_>>();
polls.poll_next_unpin(cx)
}
}

View File

@ -1,203 +0,0 @@
#![cfg(not(debug_assertions))]
mod common;
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use common::{
behaviour::{CallTraceBehaviour, MockBehaviour},
swarm,
};
use lighthouse_network::{
peer_manager::{config::Config, PeerManagerEvent},
NetworkGlobals, PeerAction, PeerInfo, PeerManager, ReportSource,
};
use types::MinimalEthSpec as E;
use futures::StreamExt;
use libp2p::{
core::either::EitherError,
swarm::SwarmEvent,
swarm::{handler::DummyConnectionHandler, DummyBehaviour, KeepAlive, Swarm},
NetworkBehaviour,
};
use slog::debug;
/// Struct that mimics the lighthouse_network::Service with respect to handling peer manager
/// events.
// TODO: make this a real struct for more accurate testing.
struct Service {
swarm: Swarm<Behaviour>,
}
impl Service {
async fn select_next_some(&mut self) -> SwarmEvent<Ev, EitherError<void::Void, void::Void>> {
let ev = self.swarm.select_next_some().await;
match &ev {
SwarmEvent::Behaviour(Ev(PeerManagerEvent::Banned(peer_id, _addr_vec))) => {
self.swarm.ban_peer_id(*peer_id);
}
SwarmEvent::Behaviour(Ev(PeerManagerEvent::UnBanned(peer_id, _addr_vec))) => {
self.swarm.unban_peer_id(*peer_id);
}
SwarmEvent::Behaviour(Ev(PeerManagerEvent::DisconnectPeer(peer_id, _reason))) => {
// directly disconnect here.
let _ = self.swarm.disconnect_peer_id(*peer_id);
}
_ => {}
}
ev
}
}
#[derive(Debug)]
struct Ev(PeerManagerEvent);
impl From<void::Void> for Ev {
fn from(_: void::Void) -> Self {
unreachable!("No events are emmited")
}
}
impl From<PeerManagerEvent> for Ev {
fn from(ev: PeerManagerEvent) -> Self {
Ev(ev)
}
}
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "Ev")]
struct Behaviour {
pm_call_trace: CallTraceBehaviour<PeerManager<E>>,
sibling: MockBehaviour,
}
impl Behaviour {
fn new(pm: PeerManager<E>) -> Self {
Behaviour {
pm_call_trace: CallTraceBehaviour::new(pm),
sibling: MockBehaviour::new(DummyConnectionHandler {
// The peer manager votes No, so we make sure the combined handler stays alive this
// way.
keep_alive: KeepAlive::Yes,
}),
}
}
}
#[tokio::test]
async fn banned_peers_consistency() {
let log = common::build_log(slog::Level::Debug, false);
let pm_log = log.new(slog::o!("who" => "[PM]"));
let globals: Arc<NetworkGlobals<E>> = Arc::new(NetworkGlobals::new_test_globals(&log));
// Build the peer manager.
let (mut pm_service, pm_addr) = {
let pm_config = Config {
discovery_enabled: false,
..Default::default()
};
let pm = PeerManager::new(pm_config, globals.clone(), &pm_log).unwrap();
let mut pm_swarm = swarm::new_test_swarm(Behaviour::new(pm));
let pm_addr = swarm::bind_listener(&mut pm_swarm).await;
let service = Service { swarm: pm_swarm };
(service, pm_addr)
};
let excess_banned_peers = 15;
let peers_to_ban =
lighthouse_network::peer_manager::peerdb::MAX_BANNED_PEERS + excess_banned_peers;
// Build all the dummy peers needed.
let (mut swarm_pool, peers) = {
let mut pool = swarm::SwarmPool::with_capacity(peers_to_ban);
let mut peers = HashSet::with_capacity(peers_to_ban);
for _ in 0..peers_to_ban {
let mut peer_swarm =
swarm::new_test_swarm(DummyBehaviour::with_keep_alive(KeepAlive::Yes));
let _peer_addr = swarm::bind_listener(&mut peer_swarm).await;
// It is ok to dial all at the same time since the swarm handles an event at a time.
peer_swarm.dial(pm_addr.clone()).unwrap();
let peer_id = pool.insert(peer_swarm);
peers.insert(peer_id);
}
(pool, peers)
};
// we track banned peers at the swarm level here since there is no access to that info.
let mut swarm_banned_peers = HashMap::with_capacity(peers_to_ban);
let mut peers_unbanned = 0;
let timeout = tokio::time::sleep(tokio::time::Duration::from_secs(30));
futures::pin_mut!(timeout);
loop {
// poll the pm and dummy swarms.
tokio::select! {
pm_event = pm_service.select_next_some() => {
debug!(log, "[PM] {:?}", pm_event);
match pm_event {
SwarmEvent::Behaviour(Ev(ev)) => match ev {
PeerManagerEvent::Banned(peer_id, _) => {
let has_been_unbanned = false;
swarm_banned_peers.insert(peer_id, has_been_unbanned);
}
PeerManagerEvent::UnBanned(peer_id, _) => {
*swarm_banned_peers.get_mut(&peer_id).expect("Unbanned peer must be banned first") = true;
peers_unbanned += 1;
}
_ => {}
}
SwarmEvent::ConnectionEstablished {
peer_id,
endpoint: _,
num_established: _,
concurrent_dial_errors: _,
} => {
assert!(peers.contains(&peer_id));
// now we report the peer as banned.
pm_service
.swarm
.behaviour_mut()
.pm_call_trace
.inner()
.report_peer(
&peer_id,
PeerAction::Fatal,
ReportSource::Processor,
None,
""
);
},
_ => {}
}
}
Some((_peer_id, _peer_ev)) = swarm_pool.next() => {
// we need to poll the swarms to keep the peers going
}
_ = timeout.as_mut() => {
panic!("Test timeout.")
}
}
if peers_unbanned == excess_banned_peers {
let pdb = globals.peers.read();
let inconsistencies = swarm_banned_peers
.into_iter()
.map(|(peer_id, was_unbanned)| {
was_unbanned
!= pdb.peer_info(&peer_id).map_or(
false, /* We forgot about a banned peer */
PeerInfo::is_banned,
)
});
assert_eq!(
inconsistencies
.filter(|is_consistent| *is_consistent)
.count(),
peers_to_ban
);
return;
}
}
}

View File

@ -23,7 +23,6 @@ status = [
"check-msrv",
"slasher-tests",
"syncing-simulator-ubuntu",
"disallowed-from-async-lint",
"compile-with-beta-compiler"
]
use_squash_merge = true