From dea01708f0f0aa4182e3285888c0840b4788cd05 Mon Sep 17 00:00:00 2001 From: Aaro Altonen Date: Wed, 12 Jan 2022 10:21:48 +0200 Subject: [PATCH 1/4] p2p: Update protocol specification Fix couple of old fields and add nonce field to Ping/Pong so that the protocol is slightly more robust when accepting a Pong from remote peer. --- p2p/README.md | 25 ++++++++++++++++--------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/p2p/README.md b/p2p/README.md index 7aec374b97..1c7fa4984e 100644 --- a/p2p/README.md +++ b/p2p/README.md @@ -54,7 +54,7 @@ After `Hello` and `HelloAck` messages have been exchanged, the peer that sent th ### Connection maintenance -Once a minute, the node must check if the remote peer is still active by sending a `Ping` message. The peer that receives a `Ping` must respond with a `Pong` message within 10 seconds. If no response is heard, the `Ping` is sent again. If no response is heard after 3 retries, he connection is closed. `Ping`/`Pong` is exchange only if there has been no activity on the socket in the last minute. +Once a minute, the node must check if the remote peer is still active by sending a `Ping` message. The peer that receives a `Ping` must respond with a `Pong` message within 10 seconds. If no response is heard, the `Ping` is sent again. If no response is heard after 3 retries, he connection is closed. `Ping`/`Pong` is exchange only if there has been no activity on the socket in the last minute. Each `Ping` message contains a 64-bit nonce that the corresponding `Pong` sends back to ascertain that a correct `Ping` message was acknowledged. If `Pong` contains an invalid nonce, the `Ping` message is sent again and after tree failed retries the connection is closed. Additionally, once every 5 minutes, the peers shall exchange peer information they've learned within the last 5 minutes. This means that the peer which initiated the connection sends a `Pex` message which contains either an empty list (if it hasn't discovered new nodes) or a list of nodes it has discovered within the last 5 minutes. The other node responds to this message with a `PexAck` message that also contains either an empty list or a list of new nodes discovered within the last 5 minutes. @@ -69,9 +69,7 @@ Each message contains at least the header which indicates the message type it ca | Length | Description | Type | Comments | |--------|-------------|------|----------| | 4 bytes | Magic number | `u32` | Magic number that identifies a Mintlayer P2P message -| 2 bytes | Message type | `enum MessageType` | Number that identifies the message type (`Hello`, `Transaction`, etc.) -| 4 bytes | Length | `u32` | Length of the payload -| N bytes | Payload | `Vec` | Byte vector containing the SCALE-encoded representation of the message +| N bytes | Message type | `enum MessageType` | Message type (`Hello`, `Transaction`, etc.) #### Hello @@ -80,27 +78,36 @@ Each message contains at least the header which indicates the message type it ca | Length | Description | Type | Comments | |--------|-------------|------|----------| | 4 bytes | Version | `u32` | Version of the software the node is running -| 4 bytes | Network ID | `u32` | Mainnet, testnet | 4 bytes | Services | `u32` | Bitmap of services that the node provides/supports (inbound connections, validation, block relay, etc.) -| 8 bytes | Timestamp | `u64` | Unix timestamp in seconds +| 8 bytes | Timestamp | `i64` | Unix timestamp in seconds #### HelloAck `HelloAck` is used to conclude the handshake with the peer that initiated it, if they are running compatible software and are in the same network. -The format of `HelloAck` message is the same as the format of `Hello` with the exception that message type in the header is different. +| Length | Description | Type | Comments | +|--------|-------------|------|----------| +| 4 bytes | Version | `u32` | Version of the software the node is running +| 4 bytes | Services | `u32` | Bitmap of services that the node provides/supports (inbound connections, validation, block relay, etc.) +| 8 bytes | Timestamp | `i64` | Unix timestamp in seconds #### Ping Check if the peer is still alive -The `Ping` does not transfer any payload data +| Length | Description | Type | Comments | +|--------|-------------|------|----------| +| 8 bytes | Nonce | `u64` | Random nonce #### Pong Respond to an aliveness check -The `Pong` does not transfer any payload data +| Length | Description | Type | Comments | +|--------|-------------|------|----------| +| 8 bytes | Nonce | `u64` | Random nonce + +The random nonce carried in the `Pong` message must be the same that was in the `Ping` message that this `Pong` is now acknowledging. #### Pex From 4b7f8d00e3e66a2ef61607166d0fd309f8137bd8 Mon Sep 17 00:00:00 2001 From: Aaro Altonen Date: Wed, 12 Jan 2022 19:14:58 +0200 Subject: [PATCH 2/4] proto/connectivity: Implement Ping/Pong functionality Ping/Pong is used to test the aliveness of the remote peer and it works by periodically sending a Ping message and expecting a response to it within some predefined time period. This commit implements this functionality as a state machine on top of the asynchronous protocol by introducing substates to the default listening state of the `PeerState` enum. This allows the Ping/Pong functionality to follow a certain path and catch errors where appropriate while still allowing the remote peer to essentially send anything they want, as long as they reply to the Ping message in time and with correct nonce. --- p2p/src/error.rs | 4 + p2p/src/message.rs | 7 + p2p/src/peer.rs | 111 +++++++++---- p2p/src/proto/connectivity.rs | 296 ++++++++++++++++++++++++++++++++++ p2p/src/proto/handshake.rs | 12 +- p2p/src/proto/mod.rs | 1 + 6 files changed, 395 insertions(+), 36 deletions(-) create mode 100644 p2p/src/proto/connectivity.rs diff --git a/p2p/src/error.rs b/p2p/src/error.rs index ca7cc65a6e..9c954ee25c 100644 --- a/p2p/src/error.rs +++ b/p2p/src/error.rs @@ -20,6 +20,7 @@ pub enum ProtocolError { InvalidVersion, InvalidMessage, Incompatible, + Unresponsive, } #[derive(Debug, PartialEq, Eq)] @@ -66,6 +67,9 @@ impl std::fmt::Display for ProtocolError { ProtocolError::Incompatible => { write!(f, "Remote deemed us incompatible, connection closed") } + ProtocolError::Unresponsive => { + write!(f, "Remote did not reply to Pings") + } } } } diff --git a/p2p/src/message.rs b/p2p/src/message.rs index d3a0ec6ba3..c441987cf2 100644 --- a/p2p/src/message.rs +++ b/p2p/src/message.rs @@ -38,10 +38,17 @@ pub enum HandshakeMessage { }, } +#[derive(Debug, Encode, Decode, Copy, Clone, PartialEq, Eq)] +pub enum ConnectivityMessage { + Ping { nonce: u64 }, + Pong { nonce: u64 }, +} + #[derive(Debug, Encode, Decode, Copy, Clone, PartialEq, Eq)] #[allow(unused)] pub enum MessageType { Handshake(HandshakeMessage), + Connectivity(ConnectivityMessage), } #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)] diff --git a/p2p/src/peer.rs b/p2p/src/peer.rs index ff77d85ce8..0e61b3c541 100644 --- a/p2p/src/peer.rs +++ b/p2p/src/peer.rs @@ -14,24 +14,35 @@ // limitations under the License. // // Author(s): A. Altonen -use crate::error::{self, P2pError, ProtocolError}; -use crate::event::{Event, PeerEvent}; -use crate::message::{HandshakeMessage, Message, MessageType}; -use crate::net::{NetworkService, SocketService}; -use crate::proto::handshake::*; -use common::chain::ChainConfig; -use common::primitives::time; +use crate::{ + error::{self, P2pError, ProtocolError}, + event::{Event, PeerEvent}, + message::{HandshakeMessage, Message, MessageType}, + net::{NetworkService, SocketService}, + proto::{connectivity::*, handshake::*}, +}; +use common::{chain::ChainConfig, primitives::time}; use futures::{stream::FuturesUnordered, FutureExt, StreamExt}; use futures_timer::Delay; -use std::sync::Arc; -use std::time::Duration; +use std::{sync::Arc, time::Duration}; pub type PeerId = u64; pub type TaskId = u64; -struct TaskInfo { - task_id: TaskId, - period: Duration, +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub struct TaskInfo { + pub task: Task, + pub period: i64, +} + +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub enum ListeningState { + /// Listen to and handle all incoming messages + Any, + + /// Listen to and handle all incoming messages but expect + /// to receive Pong message from remote + Connectivity(ConnectivityState), } #[derive(Debug, Copy, Clone, PartialEq, Eq)] @@ -40,7 +51,7 @@ pub enum PeerState { Handshaking(HandshakeState), /// Listen to incoming messages from remote peer - Listening, + Listening(ListeningState), } #[derive(Debug, Copy, Clone, PartialEq, Eq)] @@ -52,15 +63,33 @@ pub enum PeerRole { Outbound, } -// Represents a task that will run independently of any incoming/outgoing event -// meaning the decision to run is built into the protocol and, for example, the -// network manager is not responsible for scheduling the execution of this event -const DUMMY_TASK_ID: TaskId = 1; -const DUMMY_PERIOD: Duration = Duration::from_secs(60); +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub enum ConnectivityTask { + Ping { + /// How often is Ping scheduled to happen (seconds) + period: i64, + }, + PingRetry { + /// How many times the Ping has been resent + max_retries: isize, + }, +} + +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub enum Task { + Connectivity(ConnectivityTask), +} + +pub const PING_PERIOD: i64 = 60; +pub const PING_REPLY_PERIOD: i64 = 10; +pub const PING_MAX_RETRIES: isize = 3; -async fn schedule_event(task_info: TaskInfo) -> TaskId { - Delay::new(task_info.period).await; - task_info.task_id +pub async fn schedule_event(task_info: TaskInfo) -> Task { + Delay::new(Duration::from_secs( + task_info.period.try_into().expect("Failed to convert i64 to u64"), + )) + .await; + task_info.task } #[allow(unused)] @@ -88,6 +117,9 @@ where /// Chain config pub(super) config: Arc, + + /// Last time when something was read from the socket + pub(super) last_activity: i64, } #[allow(unused)] @@ -129,6 +161,7 @@ where mgr_rx, socket, config, + last_activity: 0i64, } } @@ -152,10 +185,18 @@ where return Err(P2pError::ProtocolError(ProtocolError::DifferentNetwork)); } - if let (PeerState::Handshaking(state), MessageType::Handshake(msg)) = (self.state, msg.msg) - { - // found in src/proto/handshake.rs - self.on_handshake_event(state, msg).await?; + match (self.state, msg.msg) { + (PeerState::Handshaking(state), MessageType::Handshake(msg)) => { + // found in src/proto/handshake.rs + self.on_handshake_event(state, msg).await?; + } + (PeerState::Listening(state), MessageType::Connectivity(msg)) => { + // found in src/proto/connectivity.rs + self.on_inbound_connectivity_event(state, msg).await?; + } + (_, _) => { + println!("unhandled message"); + } } Ok(()) @@ -184,8 +225,16 @@ where /// /// This design allows the peer event loop to wait onan arbitrary number of /// timer-based events, both scheduled and one-shot. - async fn on_timer_event(&mut self, task_id: TaskId) -> error::Result> { - todo!(); + pub(super) async fn on_timer_event(&mut self, task: Task) -> error::Result> { + match (self.state, task) { + (PeerState::Listening(state), Task::Connectivity(task)) => { + // found in src/proto/connectivity.rs + self.on_outbound_connectivity_event(state, task).await + } + (PeerState::Handshaking(_), Task::Connectivity(_)) => { + Err(P2pError::ProtocolError(ProtocolError::InvalidMessage)) + } + } } /// Start event loop for the peer @@ -199,10 +248,11 @@ where /// an upper-level event loop but a task must be spawned for it pub async fn run(&mut self) -> error::Result<()> { let mut tasks = FuturesUnordered::new(); - tasks.push(schedule_event(TaskInfo { - task_id: DUMMY_TASK_ID, - period: DUMMY_PERIOD, + task: Task::Connectivity(ConnectivityTask::Ping { + period: PING_PERIOD, + }), + period: PING_PERIOD, })); // the protocol defines that the initiator of the communication, i.e., the peer @@ -226,6 +276,7 @@ where tokio::select! { event = self.socket.recv() => { self.on_peer_event(event).await?; + self.last_activity = time::get(); } event = self.mgr_rx.recv().fuse() => { self.on_manager_event(event).await?; diff --git a/p2p/src/proto/connectivity.rs b/p2p/src/proto/connectivity.rs new file mode 100644 index 0000000000..46138bfcb7 --- /dev/null +++ b/p2p/src/proto/connectivity.rs @@ -0,0 +1,296 @@ +// Copyright (c) 2021 RBB S.r.l +// opensource@mintlayer.org +// SPDX-License-Identifier: MIT +// Licensed under the MIT License; +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://spdx.org/licenses/MIT +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Author(s): A. Altonen +use crate::{ + error::{self, P2pError, ProtocolError}, + message::{ConnectivityMessage, Message, MessageType}, + net::{NetworkService, SocketService}, + peer::*, +}; +use common::primitives::time; + +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub enum ConnectivityState { + PingSent { + /// Selected nonce for the Ping + nonce: u64, + + /// How many times the Ping has been resent + retries: isize, + }, + PongReceived, +} + +impl Peer +where + NetworkingBackend: NetworkService, +{ + /// Respond to an incoming Ping with a Pong + /// + /// # Arguments + /// `nonce` - nonce that was in the Ping message + async fn on_inbound_ping_event(&mut self, nonce: u64) -> error::Result<()> { + self.socket + .send(&Message { + magic: *self.config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Pong { nonce }), + }) + .await + } + + /// Handle incoming Pong event + /// + /// This might be a new Pong which is validated against the nonce the local peer + /// sent in its Ping message and if they match, the local peer state is converted to + /// `PongReceived` to indicate that the connectivity check was completed successfully. + /// + /// There is also the possibility that the remote peer sent the Pong multiple times + /// for whatever reason and that Pong was received before the state was changed to + /// `ListeningState::Any` so just ignore the Pong message in that case. + /// + /// # Arguments + /// `state` - current connectivity state of local peer + /// `sent_nonce` - nonce that was in the Ping message that the local peer sent + async fn on_inbound_pong_event( + &mut self, + state: ConnectivityState, + sent_nonce: u64, + ) -> error::Result<()> { + match state { + ConnectivityState::PingSent { nonce, .. } => { + if sent_nonce == nonce { + self.state = PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived, + )); + } + + Ok(()) + } + ConnectivityState::PongReceived => Ok(()), + } + } + + /// Handle incoming connectivity event + /// + /// Ping can be received at any point during reception so the value of `state` does not + /// change the way the incoming Ping is processed and thus it's ignored + /// + /// Incoming Pong is can be considered valid only if the current substate is of type + /// `ConnectivityState` and processing is handled in `on_inbound_pong_event()`. + /// + /// Remote peer can also sent a stray Pong message which is considered incorrect behaviour from + /// the protocol's perspective but closing the connection may be too harsh so for now just ignore it. + /// + /// # Arguments + /// `state` - current listening state of the local peer + /// `msg` - connectivity message received from the remote peer + pub async fn on_inbound_connectivity_event( + &mut self, + state: ListeningState, + msg: ConnectivityMessage, + ) -> error::Result<()> { + match (state, msg) { + (_, ConnectivityMessage::Ping { nonce }) => self.on_inbound_ping_event(nonce).await, + (ListeningState::Connectivity(state), ConnectivityMessage::Pong { nonce }) => { + self.on_inbound_pong_event(state, nonce).await + } + (ListeningState::Any, ConnectivityMessage::Pong { .. }) => { + // Receiving a stray Pong message is invalid behaviour but closing the connection + // would be an overraction so just exit early (TODO: adjust peer reputation?) + Ok(()) + } + } + } + + /// Handle outbound Ping event + /// + /// Handling an outbound Ping event means that the 60 second timer for the Ping task + /// has expired and the local peer checks whether it should send a Ping message to remote. + /// If there has been no activity on the socket in the last 60 seconds, meaning local peer + /// has not received anything from remote peer, it sends a Ping message and schedules a PingRetry + /// task to be executed next and changes its own state to `ConnectivityState::PingSent` to indicate + /// that a response to the sent Ping message is expected. + /// + /// # Arguments + /// `state` - current listening state of the local peer + /// `period` - how often is Ping scheduled to be sent (default: 60 seconds) + /// + /// # Panics + /// The `panic!()` is added only for completeness. The logical flow of the connectivity check + /// makes it impossible for the execution to ever reach `panic!()` as `ConnectivityTask::Ping` + /// and `ConnectivityTask::PingRetry` are never simultaneously active and the act of completing + /// `ConnectivityTask::PingRetry` changes the state to `ListeningState::Any` and schedules + /// `ConnectivityTask::Ping` whereas reaching this function and actually sending the Ping message + /// changes the state to `ListeningState::Connectivity` and schedules the `ConnectivityTask::PingRetry` + /// to be executed next. + async fn on_outbound_ping_event( + &mut self, + state: ListeningState, + period: i64, + ) -> error::Result> { + match state { + ListeningState::Any => { + if time::get() - self.last_activity < period { + return Ok(Some(TaskInfo { + task: Task::Connectivity(ConnectivityTask::Ping { period }), + period, + })); + } + + let nonce: u64 = rand::random(); + self.socket + .send(&Message { + magic: *self.config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Ping { nonce }), + }) + .await?; + + self.state = PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PingSent { nonce, retries: 0 }, + )); + + Ok(Some(TaskInfo { + task: Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + }), + period: PING_REPLY_PERIOD, + })) + } + ListeningState::Connectivity(_) => { + panic!("Cannot send Ping while another connecivity check is in progress"); + } + } + } + + /// Handle valid PingRetry event + /// + /// The processing flow of PingRetry depends on one condition: + /// - has the remote send the local peer a Pong message? + /// + /// If they have, the PingRetry task schedules the Ping task to happen again + /// in 60 seconds and changes the state to `ListeningState::Any`. + /// + /// If a Pong has not been received, the execution checks if there still are + /// more retries left and if so, it resends the Ping message and schedules + /// itself again. If there are no more retries left, the code returns an + /// error indicating that the remote peer is unresponsive. + /// + /// # Arguments + /// `state` - current connecivity state of the local peer + /// `max_retries` - number of times the Ping is resent + async fn on_valid_ping_retry_event( + &mut self, + state: ConnectivityState, + max_retries: isize, + ) -> error::Result> { + match state { + ConnectivityState::PongReceived => { + self.state = PeerState::Listening(ListeningState::Any); + + Ok(Some(TaskInfo { + task: Task::Connectivity(ConnectivityTask::Ping { + period: PING_PERIOD, + }), + period: PING_PERIOD, + })) + } + ConnectivityState::PingSent { nonce, retries } => { + if retries >= max_retries { + return Err(P2pError::ProtocolError(ProtocolError::Unresponsive)); + } + + self.socket + .send(&Message { + magic: *self.config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Ping { nonce }), + }) + .await?; + + self.state = PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PingSent { + nonce, + retries: retries + 1, + }, + )); + + Ok(Some(TaskInfo { + task: Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + }), + period: PING_REPLY_PERIOD, + })) + } + } + } + + /// Handle Ping retry event issued by an expired timer + /// + /// # Arguments + /// `state` - current listening state of the local peer + /// `max_retries` - number of times the Ping is resent + /// + /// # Panics + /// The `panic!()` is added only for completeness. `ConnectivityTask::PingRetry` is the + /// only logic flow that changes the state to `ListeningState::Any` and simultaneously + /// schedules the `ConnectivityTask::Ping` to be executed next. These two tasks are never + /// scheduled simultaneously so the in this function `panic!()` is unreachable. + async fn on_ping_retry_event( + &mut self, + state: ListeningState, + max_retries: isize, + ) -> error::Result> { + match state { + ListeningState::Connectivity(state) => { + self.on_valid_ping_retry_event(state, max_retries).await + } + ListeningState::Any => { + panic!("Ping cannot be resent if peer is not in connectivity state"); + } + } + } + + /// Handle scheduled, ping-related event + /// + /// `on_outbound_connectivity_event()` either handles `ConnectivityTask::Ping` which means that + /// the 60 second timer has expired and `self.last_activity` must be checked. If the socket has + /// had activity within the last minute, the ping task is scheduled again and code returns. + /// + /// If there has been no activity within the last minute, `ConnectivityMessage::Ping` is + /// sent to remote and it must respond to it within 10 seconds. Before the code returns, + /// it schedules a new `ConnectivityTask::PingRetry` task which expires in 10 seconds and checks + /// if the response has been received. If not, and if this is the first, second or third time of + /// sending the Ping message, it's sent again and the `ConnectivityTask::PingRetry` task is also + /// scheduled again. + /// + /// If no response is heard after three retries (10 + 30 seconds), the code returns an error + /// which indicates to the caller that remote is unresponsive and connection should be + /// closed. + /// + /// If a response is heard, the check was successful, remote is responsive, + /// and the original `ConnectivityTask::Ping` is again scheduled to happen in 60 seconds. + pub async fn on_outbound_connectivity_event( + &mut self, + state: ListeningState, + task: ConnectivityTask, + ) -> error::Result> { + match task { + ConnectivityTask::Ping { period } => self.on_outbound_ping_event(state, period).await, + ConnectivityTask::PingRetry { max_retries } => { + self.on_ping_retry_event(state, max_retries).await + } + } + } +} diff --git a/p2p/src/proto/handshake.rs b/p2p/src/proto/handshake.rs index 39b2149c1b..309fdc3a58 100644 --- a/p2p/src/proto/handshake.rs +++ b/p2p/src/proto/handshake.rs @@ -17,7 +17,7 @@ use crate::error::{self, P2pError, ProtocolError}; use crate::message::{HandshakeMessage, Message, MessageType}; use crate::net::{NetworkService, SocketService}; -use crate::peer::{Peer, PeerState}; +use crate::peer::{ListeningState, Peer, PeerState}; use common::primitives::time; #[derive(Debug, Copy, Clone, PartialEq, Eq)] @@ -81,7 +81,7 @@ where }; self.socket.send(&msg).await?; - self.state = PeerState::Listening; + self.state = PeerState::Listening(ListeningState::Any); return Ok(()); } (InboundHandshakeState::WaitInitiation, HandshakeMessage::HelloAck { .. }) => { @@ -128,7 +128,7 @@ where return Err(P2pError::ProtocolError(ProtocolError::InvalidMessage)); } - self.state = PeerState::Listening; + self.state = PeerState::Listening(ListeningState::Any); return Ok(()); } (OutboundHandshakeState::WaitResponse, HandshakeMessage::Hello { .. }) => { @@ -275,15 +275,15 @@ mod tests { OutboundHandshakeState::WaitResponse )) ); - assert_eq!(remote.state, PeerState::Listening); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); // read initiator socket and parse message let msg = local.socket.recv().await; let res = local.on_peer_event(msg).await; assert!(res.is_ok()); - assert_eq!(local.state, PeerState::Listening); - assert_eq!(remote.state, PeerState::Listening); + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); } // Test that invalid magic number closes the connection diff --git a/p2p/src/proto/mod.rs b/p2p/src/proto/mod.rs index 778b07671a..76cf9bf4ae 100644 --- a/p2p/src/proto/mod.rs +++ b/p2p/src/proto/mod.rs @@ -14,4 +14,5 @@ // limitations under the License. // // Author(s): A. Altonen +pub mod connectivity; pub mod handshake; From f9ef4d4d30fb9196ed66f3d5cd3b43dc54e93c6f Mon Sep 17 00:00:00 2001 From: Aaro Altonen Date: Thu, 13 Jan 2022 07:25:34 +0200 Subject: [PATCH 3/4] proto/connectivity: Add tests for Ping/Pong --- p2p/src/proto/connectivity.rs | 778 ++++++++++++++++++++++++++++++++++ 1 file changed, 778 insertions(+) diff --git a/p2p/src/proto/connectivity.rs b/p2p/src/proto/connectivity.rs index 46138bfcb7..87f32d814f 100644 --- a/p2p/src/proto/connectivity.rs +++ b/p2p/src/proto/connectivity.rs @@ -294,3 +294,781 @@ where } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + message::HandshakeMessage, + net::mock::{MockService, MockSocket}, + peer::PeerRole, + proto::handshake::{HandshakeState, OutboundHandshakeState}, + }; + use common::chain::{config, ChainConfig}; + use std::sync::Arc; + use tokio::net::TcpStream; + + async fn create_two_peers( + config: Arc, + addr: std::net::SocketAddr, + ) -> (Peer, Peer) { + let mut server = MockService::new(addr).await.unwrap(); + let peer_fut = TcpStream::connect(addr); + + let (remote_res, local_res) = tokio::join!(server.accept(), peer_fut); + let remote_res = remote_res.unwrap(); + let local_res = local_res.unwrap(); + + let (peer_tx, _peer_rx) = tokio::sync::mpsc::channel(1); + let (_tx, rx) = tokio::sync::mpsc::channel(1); + let (_tx2, rx2) = tokio::sync::mpsc::channel(1); + + let mut local = Peer::::new( + 1, + PeerRole::Outbound, + config.clone(), + remote_res, + peer_tx.clone(), + rx, + ); + + let mut remote = Peer::::new( + 2, + PeerRole::Inbound, + config.clone(), + MockSocket::new(local_res), + peer_tx, + rx2, + ); + + // handshake with remove + local + .on_handshake_event( + HandshakeState::Outbound(OutboundHandshakeState::Initiate), + HandshakeMessage::Hello { + version: *config.version(), + services: 0u32, + timestamp: time::get(), + }, + ) + .await + .unwrap(); + + // respond to Hello with HelloAck + let msg = remote.socket.recv().await; + remote.on_peer_event(msg).await.unwrap(); + + // read HelloAck and conclude handshake + let msg = local.socket.recv().await; + local.on_peer_event(msg).await.unwrap(); + + (local, remote) + } + + // helper function to retransmit Ping and verify peer state + async fn ping_retry(peer: &mut Peer, retries: isize) { + // resend the ping `retries` times and verify on each iteration that local peer's state updates correctly + for retry in 0..retries { + // verify peer state + if let PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PingSent { nonce: _, retries }, + )) = peer.state + { + assert_eq!(retries, retry); + } else { + unreachable!(); + } + + // manually trigger `PingRetry` event + peer.on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await + .unwrap(); + } + } + + // send ping, respond with pong + #[tokio::test] + async fn test_valid_ping_pong() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11131".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // verify local state + if let PeerState::Listening(ListeningState::Connectivity(ConnectivityState::PingSent { + nonce: _, + retries, + })) = local.state + { + assert_eq!(retries, 0isize); + } else { + unreachable!(); + } + + // read Ping and verify state of remote peer + let msg = remote.socket.recv().await; + remote.on_peer_event(msg).await.unwrap(); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // read Pong and verify state of local peer + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + assert_eq!( + local.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + // trigger PingRetry event manually and verify final state + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: 3isize, + })) + .await + .unwrap(); + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + } + + // do not respond to ping at all + #[tokio::test] + async fn test_no_response() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11132".parse().unwrap(); + let (mut local, remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // transmit Ping 3 times + ping_retry(&mut local, PING_MAX_RETRIES).await; + + // verify that local peer considers remote unresponsive (as it should) + assert_eq!( + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await, + Err(P2pError::ProtocolError(ProtocolError::Unresponsive)) + ); + } + + // respond to ping on the first retry + #[tokio::test] + async fn test_late_response_v1() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11133".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // transmit Ping onces + ping_retry(&mut local, 1).await; + + // read Ping and verify state of remote peer + let msg = remote.socket.recv().await; + remote.on_peer_event(msg).await.unwrap(); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // read Pong and verify state of local peer + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + assert_eq!( + local.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + // trigger PingRetry event manually and verify final state + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: 3isize, + })) + .await + .unwrap(); + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + } + + // respond to ping on the second retry + #[tokio::test] + async fn test_late_response_v2() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11134".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // transmit Ping onces + ping_retry(&mut local, 2).await; + + // read Ping and verify state of remote peer + let msg = remote.socket.recv().await; + remote.on_peer_event(msg).await.unwrap(); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // read Pong and verify state of local peer + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + assert_eq!( + local.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + // trigger PingRetry event manually and verify final state + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: 3isize, + })) + .await + .unwrap(); + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + } + + // respond to ping on the third and last retry + #[tokio::test] + async fn test_late_response_v3() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11135".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // transmit Ping onces + ping_retry(&mut local, PING_MAX_RETRIES).await; + + // read Ping and verify state of remote peer + let msg = remote.socket.recv().await; + remote.on_peer_event(msg).await.unwrap(); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // read Pong and verify state of local peer + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + assert_eq!( + local.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + // trigger PingRetry event manually and verify final state + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: 3isize, + })) + .await + .unwrap(); + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + } + + // respond to ping but with invalid nonce + #[tokio::test] + async fn test_send_pong_invalid_nonce() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11136".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // verify local state + let nonce = if let PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PingSent { nonce, retries }, + )) = local.state + { + assert_eq!(retries, 0isize); + nonce + } else { + unreachable!(); + }; + + // resend the ping N times and respond to it each time with an invalid Pong + for retry in 0..PING_MAX_RETRIES { + // send Pong with invalid nonce + remote + .socket + .send(&Message { + magic: *config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Pong { + nonce: nonce.wrapping_add(1), + }), + }) + .await + .unwrap(); + + // read Pong from socket and verify that state remains as `PingRetry` + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + + if let PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PingSent { nonce: _, retries }, + )) = local.state + { + assert_eq!(retries, retry); + } else { + unreachable!(); + } + + // manually trigger `PingRetry` event + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await + .unwrap(); + } + + // verify that local peer considers remote unresponsive (as it should) + assert_eq!( + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await, + Err(P2pError::ProtocolError(ProtocolError::Unresponsive)) + ); + } + + // respond to ping first with invalid nonce and then with valid nonce + #[tokio::test] + async fn test_send_pong_invalid_then_valid_nonce() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11137".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // verify local state + let nonce = if let PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PingSent { nonce, retries }, + )) = local.state + { + assert_eq!(retries, 0isize); + nonce + } else { + unreachable!(); + }; + + // send Pong with invalid nonce + remote + .socket + .send(&Message { + magic: *config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Pong { + nonce: nonce.wrapping_add(1), + }), + }) + .await + .unwrap(); + + // read Pong from socket and verify that state remains as `PingRetry` + // meaning that the Pong was rejected + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + + if let PeerState::Listening(ListeningState::Connectivity(ConnectivityState::PingSent { + nonce: _, + retries, + })) = local.state + { + assert_eq!(retries, 0isize); + } else { + unreachable!(); + } + + // manually trigger `PingRetry` event + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await + .unwrap(); + + if let PeerState::Listening(ListeningState::Connectivity(ConnectivityState::PingSent { + nonce: _, + retries, + })) = local.state + { + assert_eq!(retries, 1isize); + } else { + unreachable!(); + } + + // send Pong with valid nonce + remote + .socket + .send(&Message { + magic: *config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Pong { nonce }), + }) + .await + .unwrap(); + + // read Pong from socket and verify that state has changed to `PongReceived` + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + assert_eq!( + local.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + // manually trigger `PingRetry` event and verify that has changed to `ListeningState::Any` + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await + .unwrap(); + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + } + + // respond to ping first with invalid nonce, then miss first and second retry + // and then finally respond to the last retry with correct nonce + #[tokio::test] + async fn test_send_pong_invalid_nonce_late_response() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11138".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // verify local state + let nonce = if let PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PingSent { nonce, retries }, + )) = local.state + { + assert_eq!(retries, 0isize); + nonce + } else { + unreachable!(); + }; + + // send Pong with invalid nonce + remote + .socket + .send(&Message { + magic: *config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Pong { + nonce: nonce.wrapping_add(1), + }), + }) + .await + .unwrap(); + + // read Pong from socket and verify that state remains as `PingRetry` + // meaning that the Pong was rejected + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + + if let PeerState::Listening(ListeningState::Connectivity(ConnectivityState::PingSent { + nonce: _, + retries, + })) = local.state + { + assert_eq!(retries, 0isize); + } else { + unreachable!(); + } + + // transmit Ping onces + ping_retry(&mut local, PING_MAX_RETRIES).await; + + // send Pong with valid nonce + remote + .socket + .send(&Message { + magic: *config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Pong { nonce }), + }) + .await + .unwrap(); + + // read Pong from socket and verify that state has changed to `PongReceived` + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + assert_eq!( + local.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + // manually trigger `PingRetry` event and verify that has changed to `ListeningState::Any` + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await + .unwrap(); + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + } + + // Simultaneously send ping messages and verify that both peers + // conclude the connectivity check correctly by sending a pong with + // correct nonce and expecing a pong with correct nonce back + #[tokio::test] + async fn test_simultaneous_ping() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11139".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // verify local state + if let PeerState::Listening(ListeningState::Connectivity(ConnectivityState::PingSent { + nonce: _, + retries, + })) = local.state + { + assert_eq!(retries, 0isize); + } else { + unreachable!(); + }; + + // send Ping + remote + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // verify remote state + if let PeerState::Listening(ListeningState::Connectivity(ConnectivityState::PingSent { + nonce: _, + retries, + })) = remote.state + { + assert_eq!(retries, 0isize); + } else { + unreachable!(); + }; + + // read remote peer's Ping from socket and respond with Pong + // save local peer's Ping to a temporary variable + let msg = local.socket.recv().await; + let ping: Message = remote.socket.recv().await.unwrap(); + assert_eq!(local.on_peer_event(msg).await, Ok(())); + + // read Pong from socket and verify state + let msg = remote.socket.recv().await; + assert_eq!(remote.on_peer_event(msg).await, Ok(())); + + // verify local and remote states + assert_eq!( + remote.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + if let PeerState::Listening(ListeningState::Connectivity(ConnectivityState::PingSent { + nonce: _, + retries, + })) = local.state + { + assert_eq!(retries, 0isize); + } else { + unreachable!(); + }; + + // respond to local peer's Ping and read the response + assert_eq!(remote.on_peer_event(Ok(ping)).await, Ok(())); + + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + + // once again verify local and remote states + assert_eq!( + remote.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + assert_eq!( + local.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + + // manually trigger `PingRetry` event for local and remote + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await + .unwrap(); + remote + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await + .unwrap(); + + // finally verify that both peers are back to listening the socket normally + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + } + + // Send Pong and then another Pong right after + #[tokio::test] + async fn test_duplicate_pong() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11140".parse().unwrap(); + let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + // send Ping + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + // verify local state + let nonce = if let PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PingSent { nonce, retries }, + )) = local.state + { + assert_eq!(retries, 0isize); + nonce + } else { + unreachable!(); + }; + + // send Pong two times and verify that it doens't cause an error + // but only changes the local node state to `PongReceived` + for _ in 0..2 { + remote + .socket + .send(&Message { + magic: *config.magic_bytes(), + msg: MessageType::Connectivity(ConnectivityMessage::Pong { nonce }), + }) + .await + .unwrap(); + + let msg = local.socket.recv().await; + assert_eq!(local.on_peer_event(msg).await, Ok(())); + assert_eq!( + local.state, + PeerState::Listening(ListeningState::Connectivity( + ConnectivityState::PongReceived + )) + ); + } + + // finally verify that the state is back to `ListeningState::Any` + local + .on_timer_event(Task::Connectivity(ConnectivityTask::PingRetry { + max_retries: PING_MAX_RETRIES, + })) + .await + .unwrap(); + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + } + + // Verify that recent activity on the socket cancels the schedule Ping + #[tokio::test] + async fn test_socket_activity() { + let config = Arc::new(config::create_mainnet()); + let addr = "[::1]:11141".parse().unwrap(); + let (mut local, remote) = create_two_peers(config.clone(), addr).await; + + // verify that handshake was successful + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + assert_eq!(remote.state, PeerState::Listening(ListeningState::Any)); + + local.last_activity = time::get(); + + // try to send Ping and verify that because last activity on the socket was + // less than 60 seconds ago, Ping is not sent + local + .on_timer_event(Task::Connectivity(ConnectivityTask::Ping { period: 60i64 })) + .await + .unwrap(); + + assert_eq!(local.state, PeerState::Listening(ListeningState::Any)); + } +} From bd4748991fd2680e2a61a88acafc1d6a85c0d312 Mon Sep 17 00:00:00 2001 From: Aaro Altonen Date: Fri, 14 Jan 2022 07:37:24 +0200 Subject: [PATCH 4/4] p2p: Do not match multiple variables in one expression Instead of matching both state and message type at the same time, match the state first, then call an appropriate function for that state and match the message type in that function to handle the (state, message type) combination. Additionally, add some comments, reword an error message and rename a few tests functions to be more descriptive. --- p2p/src/error.rs | 2 +- p2p/src/peer.rs | 77 ++++++++++++++++++++++++----------- p2p/src/proto/connectivity.rs | 6 +-- p2p/src/proto/handshake.rs | 19 ++++++++- 4 files changed, 76 insertions(+), 28 deletions(-) diff --git a/p2p/src/error.rs b/p2p/src/error.rs index 9c954ee25c..89156fce83 100644 --- a/p2p/src/error.rs +++ b/p2p/src/error.rs @@ -68,7 +68,7 @@ impl std::fmt::Display for ProtocolError { write!(f, "Remote deemed us incompatible, connection closed") } ProtocolError::Unresponsive => { - write!(f, "Remote did not reply to Pings") + write!(f, "No response from remote peer") } } } diff --git a/p2p/src/peer.rs b/p2p/src/peer.rs index 0e61b3c541..7c0777c986 100644 --- a/p2p/src/peer.rs +++ b/p2p/src/peer.rs @@ -41,7 +41,7 @@ pub enum ListeningState { Any, /// Listen to and handle all incoming messages but expect - /// to receive Pong message from remote + /// to receive Pong message from remote Connectivity(ConnectivityState), } @@ -75,13 +75,27 @@ pub enum ConnectivityTask { }, } +/// Task is an abstraction over some piece of code +/// that is scheduled to happen either periodically +/// (such as the ping task once a minute) or in one-shot +/// fashion (such as the ping retry task). +/// +/// It's a wrapper for an asynchronous timer which returns +/// the task type when it expires and allows the peer event +/// loop to handle it alongside with any other event received +/// either from the socket or from P2P's RX channel. #[derive(Debug, Copy, Clone, PartialEq, Eq)] pub enum Task { Connectivity(ConnectivityTask), } +/// How often (in seconds) is ping/pong scheduled to happen pub const PING_PERIOD: i64 = 60; + +/// How long is the response to ping waited until ping is sent again pub const PING_REPLY_PERIOD: i64 = 10; + +/// How many times is ping resent until the remote is considered unresponsive pub const PING_MAX_RETRIES: isize = 3; pub async fn schedule_event(task_info: TaskInfo) -> Task { @@ -165,6 +179,23 @@ where } } + /// Handle inbound message when local peer is listening + async fn on_listening_state_peer_event( + &mut self, + state: ListeningState, + msg: Message, + ) -> error::Result<()> { + match msg.msg { + MessageType::Connectivity(msg) => { + // found in src/proto/connectivity.rs + self.on_inbound_connectivity_event(state, msg).await + } + MessageType::Handshake(_) => { + Err(P2pError::ProtocolError(ProtocolError::InvalidMessage)) + } + } + } + /// Handle message coming from the remote peer /// /// This might be an invalid message (such as a stray Hello), it might be Ping in @@ -185,21 +216,10 @@ where return Err(P2pError::ProtocolError(ProtocolError::DifferentNetwork)); } - match (self.state, msg.msg) { - (PeerState::Handshaking(state), MessageType::Handshake(msg)) => { - // found in src/proto/handshake.rs - self.on_handshake_event(state, msg).await?; - } - (PeerState::Listening(state), MessageType::Connectivity(msg)) => { - // found in src/proto/connectivity.rs - self.on_inbound_connectivity_event(state, msg).await?; - } - (_, _) => { - println!("unhandled message"); - } + match self.state { + PeerState::Handshaking(state) => self.on_handshake_state_peer_event(state, msg).await, + PeerState::Listening(state) => self.on_listening_state_peer_event(state, msg).await, } - - Ok(()) } /// Handle event coming from the network manager @@ -211,6 +231,20 @@ where todo!(); } + /// Handle timer event when local peer is listening + async fn on_listening_state_timer_event( + &mut self, + state: ListeningState, + task: Task, + ) -> error::Result> { + match task { + Task::Connectivity(task) => { + // found in src/proto/connectivity.rs + self.on_outbound_connectivity_event(state, task).await + } + } + } + /// Handle event that's scheduled to happen when a timer expires /// /// This might be a Ping message that is sent periodically to verify that @@ -226,12 +260,9 @@ where /// This design allows the peer event loop to wait onan arbitrary number of /// timer-based events, both scheduled and one-shot. pub(super) async fn on_timer_event(&mut self, task: Task) -> error::Result> { - match (self.state, task) { - (PeerState::Listening(state), Task::Connectivity(task)) => { - // found in src/proto/connectivity.rs - self.on_outbound_connectivity_event(state, task).await - } - (PeerState::Handshaking(_), Task::Connectivity(_)) => { + match self.state { + PeerState::Listening(state) => self.on_listening_state_timer_event(state, task).await, + PeerState::Handshaking(_) => { Err(P2pError::ProtocolError(ProtocolError::InvalidMessage)) } } @@ -255,8 +286,8 @@ where period: PING_PERIOD, })); - // the protocol defines that the initiator of the communication, i.e., the peer - // who connected is responsible for sending the Hello message. This means that + // the protocol defines that the initiator of the communication, i.e., the outbound + // peer, is responsible for sending the Hello message. This means that // before the actual event loop is started, if the local node is initiator, // it must first send the Hello message and only then proceed to responding // to incoming events from remote peer and the network manager diff --git a/p2p/src/proto/connectivity.rs b/p2p/src/proto/connectivity.rs index 87f32d814f..7f18a17634 100644 --- a/p2p/src/proto/connectivity.rs +++ b/p2p/src/proto/connectivity.rs @@ -474,7 +474,7 @@ mod tests { // respond to ping on the first retry #[tokio::test] - async fn test_late_response_v1() { + async fn test_late_response_on_1st_retry() { let config = Arc::new(config::create_mainnet()); let addr = "[::1]:11133".parse().unwrap(); let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; @@ -519,7 +519,7 @@ mod tests { // respond to ping on the second retry #[tokio::test] - async fn test_late_response_v2() { + async fn test_late_response_on_2nd_retry() { let config = Arc::new(config::create_mainnet()); let addr = "[::1]:11134".parse().unwrap(); let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; @@ -564,7 +564,7 @@ mod tests { // respond to ping on the third and last retry #[tokio::test] - async fn test_late_response_v3() { + async fn test_late_response_on_3rd_retry() { let config = Arc::new(config::create_mainnet()); let addr = "[::1]:11135".parse().unwrap(); let (mut local, mut remote) = create_two_peers(config.clone(), addr).await; diff --git a/p2p/src/proto/handshake.rs b/p2p/src/proto/handshake.rs index 309fdc3a58..f759680c38 100644 --- a/p2p/src/proto/handshake.rs +++ b/p2p/src/proto/handshake.rs @@ -155,7 +155,7 @@ where /// /// This function assumes that the magic number of the message has been verified /// and sender and the local node are using the same chain type (Mainnet, Testnet) - pub async fn on_handshake_event( + pub(super) async fn on_handshake_event( &mut self, state: HandshakeState, msg: HandshakeMessage, @@ -165,6 +165,23 @@ where HandshakeState::Outbound(state) => self.on_outbound_handshake_event(state, msg).await, } } + + /// Handle inboud message when local peer is handshaking + pub async fn on_handshake_state_peer_event( + &mut self, + state: HandshakeState, + msg: Message, + ) -> error::Result<()> { + match msg.msg { + MessageType::Handshake(msg) => { + // found in src/proto/handshake.rs + self.on_handshake_event(state, msg).await + } + MessageType::Connectivity(_) => { + Err(P2pError::ProtocolError(ProtocolError::InvalidMessage)) + } + } + } } #[cfg(test)]