Repository: mintlayer/mintlayer-core
Pull request #659 (2023-01-30 07:03:33)

Add ping messages
This PR will add a check that peers are online and responding in time. This should make p2p more robust and prevent some problems caused by malicious/slow clients (such as unlimited send buffer growth).
Ping requests are sent to all connected peers every 60 seconds, and if no response is received after 150 seconds, the peer is disconnected. A duplicate request is sent until a response is received.
I have moved some functionality from PeerDb to PeerManager. I think PeerDb should be only deal with peer addresses (so it would be more like AddrMan in Bitcoin Core).


Patch:
diff --git a/node/src/config_files/mod.rs b/node/src/config_files/mod.rs
index e9b58625b..16c0c779f 100644
--- a/node/src/config_files/mod.rs
+++ b/node/src/config_files/mod.rs
@@ -139,12 +139,16 @@ fn p2p_config(config: P2pConfigFile, options: &RunOptions) -> P2pConfigFile {
         ban_threshold,
         ban_duration,
         outbound_connection_timeout,
+        ping_check_period,
+        ping_timeout,
         node_type,
     } = config;
 
     let bind_addresses = options.p2p_addr.clone().or(bind_addresses);
     let added_nodes = options.p2p_add_node.clone().or(added_nodes);
     let ban_threshold = options.p2p_ban_threshold.or(ban_threshold);
+    let ping_check_period = options.p2p_ping_check_period.or(ping_check_period);
+    let ping_timeout = options.p2p_ping_timeout.or(ping_timeout);
     let outbound_connection_timeout =
         options.p2p_outbound_connection_timeout.or(outbound_connection_timeout);
     let node_type = options.node_type.or(node_type);
@@ -155,6 +159,8 @@ fn p2p_config(config: P2pConfigFile, options: &RunOptions) -> P2pConfigFile {
         ban_threshold,
         ban_duration,
         outbound_connection_timeout,
+        ping_check_period,
+        ping_timeout,
         node_type,
     }
 }
diff --git a/node/src/config_files/p2p.rs b/node/src/config_files/p2p.rs
index a8c6b0ce4..d9e83071b 100644
--- a/node/src/config_files/p2p.rs
+++ b/node/src/config_files/p2p.rs
@@ -61,6 +61,10 @@ pub struct P2pConfigFile {
     pub ban_duration: Option<u64>,
     /// The outbound connection timeout value in seconds.
     pub outbound_connection_timeout: Option<NonZeroU64>,
+    /// How often send ping requests to peers.
+    pub ping_check_period: Option<u64>,
+    /// When a peer is detected as dead and disconnected.
+    pub ping_timeout: Option<u64>,
     /// A node type.
     pub node_type: Option<NodeTypeConfigFile>,
 }
@@ -76,6 +80,8 @@ impl From<P2pConfigFile> for P2pConfig {
                 .outbound_connection_timeout
                 .map(|t| Duration::from_secs(t.into()))
                 .into(),
+            ping_check_period: c.ping_check_period.map(Duration::from_secs).into(),
+            ping_timeout: c.ping_timeout.map(Duration::from_secs).into(),
             node_type: c.node_type.map(Into::into).into(),
             allow_discover_private_ips: Default::default(),
         }
diff --git a/node/src/main.rs b/node/src/main.rs
index d22f5d02a..0a64dc8f1 100644
--- a/node/src/main.rs
+++ b/node/src/main.rs
@@ -25,7 +25,7 @@ async fn run() -> anyhow::Result<()> {
 #[tokio::main]
 async fn main() {
     run().await.unwrap_or_else(|err| {
-        logging::log::error!("Mintlayer node launch failed: {err:?}");
+        eprintln!("Mintlayer node launch failed: {err:?}");
         std::process::exit(1)
     })
 }
diff --git a/node/src/options.rs b/node/src/options.rs
index 34fd35c8c..cb07d40b4 100644
--- a/node/src/options.rs
+++ b/node/src/options.rs
@@ -95,6 +95,15 @@ pub struct RunOptions {
     #[clap(long)]
     pub p2p_outbound_connection_timeout: Option<NonZeroU64>,
 
+    /// How often send ping requests to peers (in seconds).
+    /// Set to 0 to disable sending ping requests.
+    #[clap(long)]
+    pub p2p_ping_check_period: Option<u64>,
+
+    /// After what time a peer is detected as dead and is disconnected (in seconds).
+    #[clap(long)]
+    pub p2p_ping_timeout: Option<u64>,
+
     /// A maximum tip age in seconds.
     ///
     /// The initial block download is finished if the difference between the current time and the
diff --git a/p2p/src/config.rs b/p2p/src/config.rs
index a405c9934..e4562feb2 100644
--- a/p2p/src/config.rs
+++ b/p2p/src/config.rs
@@ -31,6 +31,8 @@ make_config_setting!(
 );
 make_config_setting!(NodeTypeSetting, NodeType, NodeType::Full);
 make_config_setting!(AllowDiscoverPrivateIps, bool, false);
+make_config_setting!(PingCheckPeriod, Duration, Duration::from_secs(60));
+make_config_setting!(PingTimeout, Duration, Duration::from_secs(150));
 
 /// A node type.
 #[derive(Debug, Copy, Clone)]
@@ -70,6 +72,10 @@ pub struct P2pConfig {
     pub ban_duration: BanDuration,
     /// The outbound connection timeout value in seconds.
     pub outbound_connection_timeout: OutboundConnectionTimeout,
+    /// How often send ping requests to peers
+    pub ping_check_period: PingCheckPeriod,
+    /// When a peer is detected as dead and disconnected
+    pub ping_timeout: PingTimeout,
     /// A node type.
     pub node_type: NodeTypeSetting,
     /// Allow announcing and discovering local and private IPs. Should be used for testing only.
diff --git a/p2p/src/message.rs b/p2p/src/message.rs
index 3e8b428c9..8a5eaff10 100644
--- a/p2p/src/message.rs
+++ b/p2p/src/message.rs
@@ -68,6 +68,11 @@ pub struct AnnounceAddrRequest {
     pub address: PeerAddress,
 }
 
+#[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
+pub struct PingRequest {
+    pub nonce: u64,
+}
+
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub enum Request {
     #[codec(index = 0)]
@@ -78,6 +83,8 @@ pub enum Request {
     AddrListRequest(AddrListRequest),
     #[codec(index = 3)]
     AnnounceAddrRequest(AnnounceAddrRequest),
+    #[codec(index = 4)]
+    PingRequest(PingRequest),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -90,6 +97,7 @@ pub enum SyncRequest {
 pub enum PeerManagerRequest {
     AddrListRequest(AddrListRequest),
     AnnounceAddrRequest(AnnounceAddrRequest),
+    PingRequest(PingRequest),
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
@@ -138,6 +146,11 @@ pub struct AddrListResponse {
     pub addresses: Vec<PeerAddress>,
 }
 
+#[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
+pub struct PingResponse {
+    pub nonce: u64,
+}
+
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub enum Response {
     #[codec(index = 0)]
@@ -148,6 +161,8 @@ pub enum Response {
     AddrListResponse(AddrListResponse),
     #[codec(index = 3)]
     AnnounceAddrResponse(AnnounceAddrResponse),
+    #[codec(index = 4)]
+    PingResponse(PingResponse),
 }
 
 #[derive(Debug, Clone)]
@@ -160,6 +175,7 @@ pub enum SyncResponse {
 pub enum PeerManagerResponse {
     AddrListResponse(AddrListResponse),
     AnnounceAddrResponse(AnnounceAddrResponse),
+    PingResponse(PingResponse),
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
@@ -175,6 +191,7 @@ impl From<PeerManagerRequest> for Request {
             PeerManagerRequest::AnnounceAddrRequest(request) => {
                 Request::AnnounceAddrRequest(request)
             }
+            PeerManagerRequest::PingRequest(request) => Request::PingRequest(request),
         }
     }
 }
@@ -186,6 +203,7 @@ impl From<PeerManagerResponse> for Response {
             PeerManagerResponse::AnnounceAddrResponse(response) => {
                 Response::AnnounceAddrResponse(response)
             }
+            PeerManagerResponse::PingResponse(response) => Response::PingResponse(response),
         }
     }
 }
diff --git a/p2p/src/net/default_backend/backend.rs b/p2p/src/net/default_backend/backend.rs
index 0de4090ff..77a98a948 100644
--- a/p2p/src/net/default_backend/backend.rs
+++ b/p2p/src/net/default_backend/backend.rs
@@ -301,6 +301,15 @@ where
                 })
                 .await
                 .map_err(P2pError::from),
+            message::Request::PingRequest(request) => self
+                .conn_tx
+                .send(ConnectivityEvent::Request {
+                    peer_id,
+                    request_id,
+                    request: PeerManagerRequest::PingRequest(request),
+                })
+                .await
+                .map_err(P2pError::from),
         }
     }
 
@@ -350,6 +359,15 @@ where
                 })
                 .await
                 .map_err(P2pError::from),
+            message::Response::PingResponse(response) => self
+                .conn_tx
+                .send(ConnectivityEvent::Response {
+                    peer_id,
+                    request_id,
+                    response: PeerManagerResponse::PingResponse(response),
+                })
+                .await
+                .map_err(P2pError::from),
         }
     }
 
diff --git a/p2p/src/net/default_backend/mod.rs b/p2p/src/net/default_backend/mod.rs
index 88a7e8b87..0e63bb170 100644
--- a/p2p/src/net/default_backend/mod.rs
+++ b/p2p/src/net/default_backend/mod.rs
@@ -60,6 +60,21 @@ pub struct ConnectivityHandle<S: NetworkingService, T: TransportSocket> {
     _marker: PhantomData<fn() -> S>,
 }
 
+impl<S: NetworkingService, T: TransportSocket> ConnectivityHandle<S, T> {
+    pub fn new(
+        local_addresses: Vec<S::Address>,
+        cmd_tx: mpsc::Sender<types::Command<T>>,
+        conn_rx: mpsc::Receiver<types::ConnectivityEvent<T>>,
+    ) -> Self {
+        Self {
+            local_addresses,
+            cmd_tx,
+            conn_rx,
+            _marker: PhantomData,
+        }
+    }
+}
+
 pub struct PubSubHandle<S, T>
 where
     S: NetworkingService,
@@ -127,12 +142,7 @@ impl<T: TransportSocket> NetworkingService for DefaultNetworkingService<T> {
         });
 
         Ok((
-            Self::ConnectivityHandle {
-                local_addresses,
-                cmd_tx: cmd_tx.clone(),
-                conn_rx,
-                _marker: Default::default(),
-            },
+            ConnectivityHandle::new(local_addresses, cmd_tx.clone(), conn_rx),
             Self::SyncingMessagingHandle {
                 cmd_tx,
                 sync_rx,
diff --git a/p2p/src/net/default_backend/types.rs b/p2p/src/net/default_backend/types.rs
index 29797dbec..cde8872d8 100644
--- a/p2p/src/net/default_backend/types.rs
+++ b/p2p/src/net/default_backend/types.rs
@@ -35,6 +35,7 @@ use crate::{
     types::peer_address::PeerAddress,
 };
 
+#[derive(Debug)]
 pub enum Command<T: TransportSocket> {
     Connect {
         address: T::Address,
diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs
index f1de22f32..2cc7f5969 100644
--- a/p2p/src/peer_manager/mod.rs
+++ b/p2p/src/peer_manager/mod.rs
@@ -19,17 +19,17 @@
 //!
 //!
 
-#![allow(rustdoc::private_intra_doc_links)]
-
 mod global_ip;
+pub mod peer_context;
 pub mod peerdb;
 
 use std::{
-    collections::{HashMap, HashSet},
+    collections::{BTreeMap, HashMap, HashSet},
     sync::Arc,
     time::Duration,
 };
 
+use crypto::random::{make_pseudo_rng, Rng, SliceRandom};
 use tokio::{
     sync::{mpsc, oneshot},
     time::Instant,
@@ -44,9 +44,10 @@ use crate::{
     config::P2pConfig,
     error::{P2pError, PeerError, ProtocolError},
     event::{PeerManagerEvent, SyncControlEvent},
+    interface::types::ConnectedPeer,
     message::{
         AddrListRequest, AddrListResponse, AnnounceAddrRequest, AnnounceAddrResponse,
-        PeerManagerRequest, PeerManagerResponse,
+        PeerManagerRequest, PeerManagerResponse, PingRequest, PingResponse,
     },
     net::{
         self,
@@ -58,7 +59,11 @@ use crate::{
     types::peer_address::{PeerAddress, PeerAddressIp4, PeerAddressIp6},
 };
 
-use self::{global_ip::IsGlobalIp, peerdb::storage::PeerDbStorage};
+use self::{
+    global_ip::IsGlobalIp,
+    peer_context::{PeerContext, SentPing},
+    peerdb::storage::PeerDbStorage,
+};
 
 /// Maximum number of connections the [`PeerManager`] is allowed to have open
 const MAX_ACTIVE_CONNECTIONS: usize = 128;
@@ -94,7 +99,13 @@ where
     tx_sync: mpsc::UnboundedSender<SyncControlEvent<T>>,
 
     /// Hashmap of pending outbound connections
-    pending: HashMap<T::Address, Option<oneshot::Sender<crate::Result<()>>>>,
+    pending_connects: HashMap<T::Address, Option<oneshot::Sender<crate::Result<()>>>>,
+
+    /// Hashmap of pending disconnect requests
+    pending_disconnects: HashMap<T::PeerId, Option<oneshot::Sender<crate::Result<()>>>>,
+
+    /// Set of active peers
+    peers: BTreeMap<T::PeerId, PeerContext<T>>,
 
     /// Peer database
     peerdb: peerdb::PeerDb<T, S>,
@@ -124,15 +135,22 @@ where
         peerdb_storage: S,
     ) -> crate::Result<Self> {
         let peerdb = peerdb::PeerDb::new(Arc::clone(&p2p_config), time_getter, peerdb_storage)?;
+        let now = tokio::time::Instant::now();
+        utils::ensure!(
+            !p2p_config.ping_timeout.is_zero(),
+            P2pError::Other("ping timeout can't be 0")
+        );
         Ok(Self {
             peer_connectivity_handle: handle,
             rx_peer_manager,
             tx_sync,
             peerdb,
-            pending: HashMap::new(),
+            peers: BTreeMap::new(),
+            pending_connects: HashMap::new(),
+            pending_disconnects: HashMap::new(),
             chain_config,
             p2p_config,
-            last_heartbeat: tokio::time::Instant::now(),
+            last_heartbeat: now,
             announced_addresses: HashMap::new(),
         })
     }
@@ -255,7 +273,7 @@ where
             ))
         );
         ensure!(
-            !self.peerdb.is_active_peer(&info.peer_id),
+            !self.is_peer_connected(&info.peer_id),
             P2pError::PeerError(PeerError::PeerAlreadyExists),
         );
         ensure!(
@@ -276,8 +294,26 @@ where
                 .await?;
         }
 
-        log::debug!("peer {peer_id} connected, address {address:?}, {info}");
-        self.peerdb.peer_connected(address, role, info);
+        log::info!(
+            "peer connected, peer_id: {}, address: {address:?}, {:?}",
+            info.peer_id,
+            role
+        );
+
+        let old_value = self.peers.insert(
+            info.peer_id,
+            PeerContext {
+                info,
+                address: address.clone(),
+                role,
+                score: 0,
+                sent_ping: None,
+            },
+        );
+        assert!(old_value.is_none());
+
+        self.peerdb.peer_connected(address);
+
         self.tx_sync.send(SyncControlEvent::Connected(peer_id)).map_err(P2pError::from)
     }
 
@@ -299,7 +335,7 @@ where
         log::debug!("validate inbound connection, inbound address {address:?}");
 
         ensure!(
-            !self.peerdb.is_active_peer(&info.peer_id),
+            !self.is_peer_connected(&info.peer_id),
             P2pError::PeerError(PeerError::PeerAlreadyExists),
         );
 
@@ -313,43 +349,61 @@ where
         // accepted even if it's valid. The peer is still reported to the PeerDb which
         // knows of all peers and later on if the number of connections falls below
         // the desired threshold, `PeerManager::heartbeat()` may connect to this peer.
-        if self.peerdb.active_peer_count() >= MAX_ACTIVE_CONNECTIONS {
+        if self.active_peer_count() >= MAX_ACTIVE_CONNECTIONS {
             return Err(P2pError::PeerError(PeerError::TooManyPeers));
         }
 
         self.accept_connection(address, Role::Inbound, info, receiver_address).await
     }
 
-    /// Close connection to a remote peer
+    /// The connection to a remote peer is reported as closed.
     ///
-    /// The decision to close the connection is made either by the user via RPC
-    /// or by the [`PeerManager::heartbeat()`] function which has decided to cull
-    /// this connection in favor of another potential connection.
-    fn close_connection(&mut self, peer_id: T::PeerId) -> crate::Result<()> {
+    /// This can happen when the remote peer has dropped its connection
+    /// or if a disconnect request has been sent by PeerManager to the backend.
+    fn connection_closed(&mut self, peer_id: T::PeerId) -> crate::Result<()> {
         // The backend is always sending ConnectionClosed event when somebody disconnects, ensure that the peer is active
-        if self.peerdb.is_active_peer(&peer_id) {
-            log::debug!("connection closed for peer {peer_id}");
+        if let Some(peer) = self.peers.remove(&peer_id) {
+            log::info!(
+                "peer disconnected, peer_id: {}, address: {:?}",
+                peer.info.peer_id,
+                peer.address
+            );
 
             self.tx_sync.send(SyncControlEvent::Disconnected(peer_id))?;
 
-            self.peerdb.peer_disconnected(&peer_id);
-        }
+            if let Some(Some(response)) = self.pending_disconnects.remove(&peer_id) {
+                response.send(Ok(())).map_err(|_| P2pError::ChannelClosed)?;
+            }
 
-        self.announced_addresses.remove(&peer_id);
+            self.peerdb.peer_disconnected(peer.address);
+
+            self.announced_addresses.remove(&peer_id);
+        }
 
         Ok(())
     }
 
     /// Adjust peer score
     ///
-    /// If after adjustment the peer score is more than the ban threshold, the peer is banned
-    /// which makes the `PeerDb` mark is banned and prevents any further connections with the peer
-    /// and also bans the peer in the networking backend.
+    /// If the peer is known, update its existing peer score and report
+    /// if it should be disconnected when score reached the threshold.
+    /// Unknown peers are reported as to be disconnected.
+    ///
+    /// If peer is banned, it is removed from the connected peers
+    /// and its address is marked as banned.
     async fn adjust_peer_score(&mut self, peer_id: T::PeerId, score: u32) -> crate::Result<()> {
         log::debug!("adjusting score for peer {peer_id}, adjustment {score}");
 
-        if self.peerdb.adjust_peer_score(&peer_id, score)? {
-            let _ = self.peer_connectivity_handle.disconnect(peer_id).await;
+        let peer = match self.peers.get_mut(&peer_id) {
+            Some(peer) => peer,
+            None => return Ok(()),
+        };
+
+        peer.score = peer.score.saturating_add(score);
+
+        if peer.score >= *self.p2p_config.ban_threshold {
+            self.peerdb.ban_peer(&peer.address)?;
+            self.disconnect(peer_id, None).await?;
         }
 
         Ok(())
@@ -364,7 +418,7 @@ where
     /// Inform the [`crate::peer_manager::peerdb::PeerDb`] about the address failure so it knows to
     /// update its own records.
     fn handle_outbound_error(&mut self, address: T::Address, error: P2pError) -> crate::Result<()> {
-        if let Some(Some(channel)) = self.pending.remove(&address) {
+        if let Some(Some(channel)) = self.pending_connects.remove(&address) {
             channel.send(Err(error)).map_err(|_| P2pError::ChannelClosed)?;
         }
 
@@ -377,9 +431,9 @@ where
     /// This function doesn't block on the call but sends a command to the
     /// networking backend which then reports at some point in the future
     /// whether the connection failed or succeeded.
-    async fn connect(&mut self, address: T::Address) -> crate::Result<()> {
+    async fn try_connect(&mut self, address: T::Address) -> crate::Result<()> {
         ensure!(
-            !self.pending.contains_key(&address),
+            !self.pending_connects.contains_key(&address),
             P2pError::PeerError(PeerError::Pending(address.to_string())),
         );
 
@@ -397,6 +451,72 @@ where
         self.peer_connectivity_handle.connect(address).await
     }
 
+    /// Establish an outbound connection
+    async fn connect(
+        &mut self,
+        address: T::Address,
+        response: Option<oneshot::Sender<crate::Result<()>>>,
+    ) -> crate::Result<()> {
+        log::debug!("try to establish outbound connection to peer at address {address:?}");
+
+        let res = self.try_connect(address.clone()).await;
+
+        match res {
+            Ok(()) => {
+                self.pending_connects.insert(address, response);
+            }
+            Err(e) => {
+                if let Some(response) = response {
+                    response.send(Err(e)).map_err(|_| P2pError::ChannelClosed)?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn try_disconnect(&mut self, peer_id: T::PeerId) -> crate::Result<()> {
+        ensure!(
+            !self.pending_disconnects.contains_key(&peer_id),
+            P2pError::PeerError(PeerError::Pending(peer_id.to_string())),
+        );
+
+        ensure!(
+            self.peers.contains_key(&peer_id),
+            P2pError::PeerError(PeerError::PeerDisconnected),
+        );
+
+        self.peer_connectivity_handle.disconnect(peer_id).await
+    }
+
+    /// Disconnect an existing connection (inbound or outbound)
+    ///
+    /// The decision to close the connection is made either by the user via RPC
+    /// or by the [`PeerManager::heartbeat()`] function which has decided to cull
+    /// this connection in favor of another potential connection.
+    async fn disconnect(
+        &mut self,
+        peer_id: T::PeerId,
+        response: Option<oneshot::Sender<crate::Result<()>>>,
+    ) -> crate::Result<()> {
+        log::debug!("disconnect peer {peer_id}");
+
+        let res = self.try_disconnect(peer_id).await;
+
+        match res {
+            Ok(()) => {
+                self.pending_disconnects.insert(peer_id, response);
+            }
+            Err(e) => {
+                if let Some(response) = response {
+                    response.send(Err(e)).map_err(|_| P2pError::ChannelClosed)?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+
     /// Maintains the peer manager state.
     ///
     /// `PeerManager::heartbeat()` is called every time a network/control event is received
@@ -423,21 +543,13 @@ where
             self.peerdb.available_addresses_count(),
             MAX_ACTIVE_CONNECTIONS
                 .saturating_sub(self.peerdb.available_addresses_count())
-                .saturating_sub(self.pending.len()),
+                .saturating_sub(self.pending_connects.len()),
         );
 
         let addresses = self.peerdb.random_known_addresses(count);
 
         for address in addresses {
-            match self.connect(address.clone()).await {
-                Ok(_) => {
-                    self.pending.insert(address, None);
-                }
-                Err(err) => {
-                    self.peerdb.report_outbound_failure(address.clone());
-                    self.handle_result(None, Err(err)).await?;
-                }
-            }
+            self.connect(address, None).await?;
         }
 
         // TODO: update peer scores
@@ -480,19 +592,27 @@ where
 
                     self.announced_addresses.entry(peer_id).or_default().insert(address.clone());
 
-                    let peer_ids = self.peerdb.random_peer_ids(ANNOUNCED_RESEND_COUNT);
+                    let peer_ids = self.random_peer_ids(ANNOUNCED_RESEND_COUNT);
                     for new_peer_id in peer_ids {
                         self.send_announced_address(new_peer_id, address.clone()).await?;
                     }
                 }
                 Ok(())
             }
+            PeerManagerRequest::PingRequest(PingRequest { nonce }) => {
+                self.peer_connectivity_handle
+                    .send_response(
+                        request_id,
+                        PeerManagerResponse::PingResponse(PingResponse { nonce }),
+                    )
+                    .await
+            }
         }
     }
 
     fn handle_incoming_response(
         &mut self,
-        _peer_id: T::PeerId,
+        peer_id: T::PeerId,
         _request_id: T::PeerRequestId,
         response: PeerManagerResponse,
     ) -> crate::Result<()> {
@@ -510,6 +630,15 @@ where
                 Ok(())
             }
             PeerManagerResponse::AnnounceAddrResponse(AnnounceAddrResponse {}) => Ok(()),
+            PeerManagerResponse::PingResponse(PingResponse { nonce }) => {
+                if let Some(peer) = self.peers.get_mut(&peer_id) {
+                    if peer.sent_ping.as_ref().map(|sent_ping| sent_ping.nonce) == Some(nonce) {
+                        // Correct reply received, clear pending request.
+                        peer.sent_ping = None;
+                    }
+                }
+                Ok(())
+            }
         }
     }
 
@@ -554,20 +683,11 @@ where
     /// Handle events from an outside controller (rpc, for example) that sets/gets values for PeerManager
     async fn handle_control_event(&mut self, event: PeerManagerEvent<T>) -> crate::Result<()> {
         match event {
-            PeerManagerEvent::Connect(addr, response) => {
-                log::debug!("try to establish outbound connection to peer at address {addr:?}");
-
-                let res = self.connect(addr.clone()).await.map(|_| {
-                    self.pending.insert(addr.clone(), Some(response));
-                });
-                self.handle_result(None, res).await?;
+            PeerManagerEvent::Connect(address, response) => {
+                self.connect(address, Some(response)).await?;
             }
             PeerManagerEvent::Disconnect(peer_id, response) => {
-                log::debug!("disconnect peer {peer_id}");
-
-                response
-                    .send(self.peer_connectivity_handle.disconnect(peer_id).await)
-                    .map_err(|_| P2pError::ChannelClosed)?;
+                self.disconnect(peer_id, Some(response)).await?;
             }
             PeerManagerEvent::AdjustPeerScore(peer_id, score, response) => {
                 log::debug!("adjust peer {peer_id} score: {score}");
@@ -577,9 +697,7 @@ where
                     .map_err(|_| P2pError::ChannelClosed)?;
             }
             PeerManagerEvent::GetPeerCount(response) => {
-                response
-                    .send(self.peerdb.active_peer_count())
-                    .map_err(|_| P2pError::ChannelClosed)?;
+                response.send(self.active_peer_count()).map_err(|_| P2pError::ChannelClosed)?;
             }
             PeerManagerEvent::GetBindAddresses(response) => {
                 let addr = self
@@ -591,7 +709,7 @@ where
                 response.send(addr).map_err(|_| P2pError::ChannelClosed)?;
             }
             PeerManagerEvent::GetConnectedPeers(response) => {
-                let peers = self.peerdb.get_connected_peers();
+                let peers = self.get_connected_peers();
                 response.send(peers).map_err(|_| P2pError::ChannelClosed)?
             }
         }
@@ -660,7 +778,7 @@ where
                         .await;
                     self.handle_result(Some(peer_id), res).await?;
 
-                    match self.pending.remove(&address) {
+                    match self.pending_connects.remove(&address) {
                         Some(Some(channel)) => {
                             channel.send(Ok(())).map_err(|_| P2pError::ChannelClosed)?
                         }
@@ -669,7 +787,7 @@ where
                     }
                 }
                 net::types::ConnectivityEvent::ConnectionClosed { peer_id } => {
-                    let res = self.close_connection(peer_id);
+                    let res = self.connection_closed(peer_id);
                     self.handle_result(Some(peer_id), res).await?;
                 }
                 net::types::ConnectivityEvent::ConnectionError { address, error } => {
@@ -690,13 +808,80 @@ where
         Ok(())
     }
 
+    /// Get the number of active peers
+    pub fn active_peer_count(&self) -> usize {
+        self.peers.len()
+    }
+
+    /// Returns short info about all connected peers
+    pub fn get_connected_peers(&self) -> Vec<ConnectedPeer> {
+        self.peers.values().map(Into::into).collect()
+    }
+
+    /// Selects requested count of connected peer ids randomly.
+    ///
+    /// It can be used to distribute data in the gossip protocol
+    /// (for example, to relay announced addresses to a small group of peers).
+    pub fn random_peer_ids(&self, count: usize) -> Vec<T::PeerId> {
+        // TODO: Optimise this
+        let all_peer_ids = self.peers.keys().cloned().collect::<Vec<_>>();
+        all_peer_ids
+            .choose_multiple(&mut make_pseudo_rng(), count)
+            .cloned()
+            .collect::<Vec<_>>()
+    }
+
+    /// Checks if the peer is in active state
+    pub fn is_peer_connected(&self, peer_id: &T::PeerId) -> bool {
+        self.peers.get(peer_id).is_some()
+    }
+
+    /// Sends ping requests and disconnects peers that do not respond in time
+    async fn ping_check(&mut self) -> crate::Result<()> {
+        let now = Instant::now();
+        let mut dead_peers = Vec::new();
+        for (peer_id, peer) in self.peers.iter_mut() {
+            // If a ping has already been sent, wait for a reply first, do not send another ping request!
+            match &peer.sent_ping {
+                Some(sent_ping) => {
+                    if now.duration_since(sent_ping.timestamp) >= *self.p2p_config.ping_timeout {
+                        log::info!("ping check: dead peer detected: {peer_id}");
+                        dead_peers.push(*peer_id);
+                    } else {
+                        log::debug!("ping check: slow peer detected: {peer_id}");
+                    }
+                }
+                None => {
+                    let nonce = make_pseudo_rng().gen();
+                    self.peer_connectivity_handle
+                        .send_request(
+                            *peer_id,
+                            PeerManagerRequest::PingRequest(PingRequest { nonce }),
+                        )
+                        .await?;
+                    peer.sent_ping = Some(SentPing {
+                        nonce,
+                        timestamp: now,
+                    });
+                }
+            }
+        }
+
+        for peer_id in dead_peers {
+            self.disconnect(peer_id, None).await?;
+        }
+
+        Ok(())
+    }
+
     /// Runs the `PeerManager` event loop.
     ///
-    /// The event loop has three main responsibilities:
+    /// The event loop has this main responsibilities:
     /// - listening to and handling control events from [`crate::sync::SyncManager`]/
     /// [`crate::pubsub::PubSubMessageHandler`]/RPC
     /// - listening to network events
     /// - updating internal state
+    /// - sending and checking ping requests
     ///
     /// After handling an event from one of the aforementioned sources, the event loop
     /// handles the error (if any) and runs the [`PeerManager::heartbeat()`] function
@@ -705,6 +890,14 @@ where
     /// This is done to prevent the `PeerManager` from stalling in case the network doesn't
     /// have any events.
     pub async fn run(&mut self) -> crate::Result<void::Void> {
+        let ping_check_enabled = !self.p2p_config.ping_check_period.is_zero();
+        let mut ping_check_interval = if ping_check_enabled {
+            tokio::time::interval(*self.p2p_config.ping_check_period)
+        } else {
+            // Use any valid (non-zero) value
+            tokio::time::interval(Duration::MAX)
+        };
+
         loop {
             tokio::select! {
                 event = self.rx_peer_manager.recv() => {
@@ -713,6 +906,9 @@ where
                 event_res = self.peer_connectivity_handle.poll_next() => {
                     self.handle_connectivity_event_result(event_res).await?;
                 },
+                _event = ping_check_interval.tick(), if ping_check_enabled => {
+                    self.ping_check().await?;
+                }
                 _event = tokio::time::sleep(PEER_MGR_HEARTBEAT_INTERVAL_MAX) => {}
             }
 
diff --git a/p2p/src/peer_manager/peer_context.rs b/p2p/src/peer_manager/peer_context.rs
new file mode 100644
index 000000000..daafbbd2d
--- /dev/null
+++ b/p2p/src/peer_manager/peer_context.rs
@@ -0,0 +1,59 @@
+// Copyright (c) 2023 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
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// 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.
+
+use tokio::time::Instant;
+
+use crate::{
+    interface::types::ConnectedPeer,
+    net::{
+        types::{self, Role},
+        NetworkingService,
+    },
+};
+
+#[derive(Debug)]
+pub struct SentPing {
+    pub nonce: u64,
+    pub timestamp: Instant,
+}
+
+#[derive(Debug)]
+pub struct PeerContext<T: NetworkingService> {
+    /// Peer information
+    pub info: types::PeerInfo<T::PeerId>,
+
+    /// Peer's address
+    pub address: T::Address,
+
+    /// Peer's role (inbound or outbound)
+    pub role: Role,
+
+    /// Peer score
+    pub score: u32,
+
+    /// Sent ping details
+    pub sent_ping: Option<SentPing>,
+}
+
+impl<T: NetworkingService> From<&PeerContext<T>> for ConnectedPeer {
+    fn from(context: &PeerContext<T>) -> Self {
+        ConnectedPeer {
+            peer_id: context.info.peer_id.to_string(),
+            address: context.address.to_string(),
+            inbound: context.role == Role::Inbound,
+            ban_score: context.score,
+        }
+    }
+}
diff --git a/p2p/src/peer_manager/peerdb/mod.rs b/p2p/src/peer_manager/peerdb/mod.rs
index d2c3aaa16..a433f98ba 100644
--- a/p2p/src/peer_manager/peerdb/mod.rs
+++ b/p2p/src/peer_manager/peerdb/mod.rs
@@ -36,56 +36,21 @@ use std::{
 
 use common::time_getter::TimeGetter;
 use crypto::random::{make_pseudo_rng, SliceRandom};
-use logging::log;
 
 use crate::{
     config,
     error::{ConversionError, P2pError},
-    interface::types::ConnectedPeer,
-    net::{
-        types::{self, Role},
-        AsBannableAddress, NetworkingService,
-    },
+    net::{AsBannableAddress, NetworkingService},
 };
 
 use self::storage::{
     PeerDbStorage, PeerDbStorageRead, PeerDbStorageWrite, PeerDbTransactionRo, PeerDbTransactionRw,
 };
 
-#[derive(Debug)]
-pub struct PeerContext<T: NetworkingService> {
-    /// Peer information
-    pub info: types::PeerInfo<T::PeerId>,
-
-    /// Peer's address
-    pub address: T::Address,
-
-    /// Peer's role (inbound or outbound)
-    pub role: Role,
-
-    /// Peer score
-    pub score: u32,
-}
-
-impl<T: NetworkingService> From<&PeerContext<T>> for ConnectedPeer {
-    fn from(context: &PeerContext<T>) -> Self {
-        ConnectedPeer {
-            peer_id: context.info.peer_id.to_string(),
-            address: context.address.to_string(),
-            inbound: context.role == Role::Inbound,
-            ban_score: context.score,
-        }
-    }
-}
-
-// TODO: Store available addresses in a binary heap (sorting by their availability).
 pub struct PeerDb<T: NetworkingService, S> {
     /// P2P configuration
     p2p_config: Arc<config::P2pConfig>,
 
-    /// Set of active peers
-    peers: BTreeMap<T::PeerId, PeerContext<T>>,
-
     /// Set of currently connected addresses
     connected_addresses: BTreeSet<T::Address>,
 
@@ -139,7 +104,6 @@ impl<T: NetworkingService, S: PeerDbStorage> PeerDb<T, S> {
             .collect();
 
         Ok(Self {
-            peers: Default::default(),
             connected_addresses: Default::default(),
             known_addresses,
             banned_addresses,
@@ -154,16 +118,6 @@ impl<T: NetworkingService, S: PeerDbStorage> PeerDb<T, S> {
         self.known_addresses.len()
     }
 
-    /// Get the number of active peers
-    pub fn active_peer_count(&self) -> usize {
-        self.peers.len()
-    }
-
-    /// Returns short info about all connected peers
-    pub fn get_connected_peers(&self) -> Vec<ConnectedPeer> {
-        self.peers.values().map(Into::into).collect()
-    }
-
     /// Checks if the given address is already connected.
     pub fn is_address_connected(&self, address: &T::Address) -> bool {
         self.connected_addresses.contains(address)
@@ -181,19 +135,6 @@ impl<T: NetworkingService, S: PeerDbStorage> PeerDb<T, S> {
             .collect::<Vec<_>>()
     }
 
-    /// Selects requested count of connected peer ids randomly.
-    ///
-    /// It can be used to distribute data in the gossip protocol
-    /// (for example, to relay announced addresses to a small group of peers).
-    pub fn random_peer_ids(&self, count: usize) -> Vec<T::PeerId> {
-        // There are normally not many connected peers, so iterating over the whole list should be OK
-        let all_peer_ids = self.peers.keys().cloned().collect::<Vec<_>>();
-        all_peer_ids
-            .choose_multiple(&mut make_pseudo_rng(), count)
-            .cloned()
-            .collect::<Vec<_>>()
-    }
-
     /// Checks if the given address is banned.
     pub fn is_address_banned(&mut self, address: &T::BannableAddress) -> crate::Result<bool> {
         if let Some(banned_till) = self.banned_addresses.get(address) {
@@ -212,11 +153,6 @@ impl<T: NetworkingService, S: PeerDbStorage> PeerDb<T, S> {
         Ok(false)
     }
 
-    /// Checks if the peer is active
-    pub fn is_active_peer(&self, peer_id: &T::PeerId) -> bool {
-        self.peers.get(peer_id).is_some()
-    }
-
     /// Get socket address of the next best peer (TODO: in terms of peer score).
     // TODO: Rewrite this.
     pub fn get_best_peer_addr(&mut self) -> Option<T::Address> {
@@ -245,93 +181,30 @@ impl<T: NetworkingService, S: PeerDbStorage> PeerDb<T, S> {
     ///
     /// After `PeerManager` has established either an inbound or an outbound connection,
     /// it informs the `PeerDb` about it.
-    pub fn peer_connected(
-        &mut self,
-        address: T::Address,
-        role: Role,
-        info: types::PeerInfo<T::PeerId>,
-    ) {
-        log::info!(
-            "peer connected, peer_id: {}, address: {address:?}, {:?}",
-            info.peer_id,
-            role
-        );
-
-        let old_value = self.peers.insert(
-            info.peer_id,
-            PeerContext {
-                info,
-                address: address.clone(),
-                role,
-                score: 0,
-            },
-        );
-        assert!(old_value.is_none());
-
-        let old_value = self.connected_addresses.insert(address);
-        assert!(old_value);
+    pub fn peer_connected(&mut self, address: T::Address) {
+        let is_inserted = self.connected_addresses.insert(address);
+        assert!(is_inserted);
     }
 
     /// Handle peer disconnection event
     ///
     /// Close the connection to an active peer.
-    pub fn peer_disconnected(&mut self, peer_id: &T::PeerId) {
-        let removed = self.peers.remove(peer_id);
-        let peer = removed.expect("peer must be known");
-
-        let removed = self.connected_addresses.remove(&peer.address);
-        assert!(removed);
-
-        log::info!(
-            "peer disconnected, peer_id: {}, address: {:?}",
-            peer.info.peer_id,
-            peer.address
-        );
+    pub fn peer_disconnected(&mut self, address: T::Address) {
+        let is_removed = self.connected_addresses.remove(&address);
+        assert!(is_removed);
     }
 
     /// Changes the peer state to `Peer::Banned` and bans it for 24 hours.
-    fn ban_peer(&mut self, peer_id: &T::PeerId) -> crate::Result<()> {
-        if let Some(peer) = self.peers.remove(peer_id) {
-            let bannable_address = peer.address.as_bannable();
-            let ban_till = self.time_getter.get_time() + *self.p2p_config.ban_duration;
-            let mut tx = self.storage.transaction_rw()?;
-            tx.add_banned_address(&bannable_address.to_string(), ban_till)?;
-            tx.commit()?;
-            self.banned_addresses.insert(bannable_address, ban_till);
-        } else {
-            log::error!("Failed to get address for peer {}", peer_id);
-        }
+    pub fn ban_peer(&mut self, address: &T::Address) -> crate::Result<()> {
+        let bannable_address = address.as_bannable();
+        let ban_till = self.time_getter.get_time() + *self.p2p_config.ban_duration;
+        let mut tx = self.storage.transaction_rw()?;
+        tx.add_banned_address(&bannable_address.to_string(), ban_till)?;
+        tx.commit()?;
+        self.banned_addresses.insert(bannable_address, ban_till);
         Ok(())
     }
 
-    /// Adjust peer score
-    ///
-    /// If the peer is known, update its existing peer score and report
-    /// if it should be disconnected when score reached the threshold.
-    /// Unknown peers are reported as to be disconnected.
-    ///
-    /// If peer is banned, it is removed from the connected peers
-    /// and its address is marked as banned.
-    pub fn adjust_peer_score(&mut self, peer_id: &T::PeerId, score: u32) -> crate::Result<bool> {
-        let peer = match self.peers.get_mut(peer_id) {
-            Some(peer) => peer,
-            None => return Ok(true),
-        };
-
-        peer.score = peer.score.saturating_add(score);
-
-        if peer.score >= *self.p2p_config.ban_threshold {
-            self.ban_peer(peer_id)?;
-            return Ok(true);
-        }
-
-        Ok(false)
-    }
-
-    pub fn peer_address(&self, id: &T::PeerId) -> Option<&T::Address> {
-        self.peers.get(id).map(|c| &c.address)
-    }
-
     #[cfg(feature = "testing_utils")]
     pub fn get_storage_mut(&mut self) -> &mut S {
         &mut self.storage


Test patch:
diff --git a/node/tests/cli.rs b/node/tests/cli.rs
index 8588a9073..36ccc0bce 100644
--- a/node/tests/cli.rs
+++ b/node/tests/cli.rs
@@ -90,6 +90,8 @@ fn read_config_override_values() {
     let p2p_add_node = "add_node";
     let p2p_ban_threshold = 3;
     let p2p_timeout = NonZeroU64::new(10000).unwrap();
+    let p2p_ping_check_period = 30;
+    let p2p_ping_timeout = 60;
     let http_rpc_addr = SocketAddr::from_str("127.0.0.1:5432").unwrap();
     let ws_rpc_addr = SocketAddr::from_str("127.0.0.1:5433").unwrap();
     let backend_type = StorageBackendConfigFile::InMemory;
@@ -97,6 +99,8 @@ fn read_config_override_values() {
     let max_tip_age = 1000;
 
     let options = RunOptions {
+        storage_backend: Some(backend_type.clone()),
+        node_type: Some(node_type),
         max_db_commit_attempts: Some(max_db_commit_attempts),
         max_orphan_blocks: Some(max_orphan_blocks),
         tx_index_enabled: Some(false),
@@ -104,13 +108,13 @@ fn read_config_override_values() {
         p2p_add_node: Some(vec![p2p_add_node.to_owned()]),
         p2p_ban_threshold: Some(p2p_ban_threshold),
         p2p_outbound_connection_timeout: Some(p2p_timeout),
+        p2p_ping_check_period: Some(p2p_ping_check_period),
+        p2p_ping_timeout: Some(p2p_ping_timeout),
+        max_tip_age: Some(max_tip_age),
         http_rpc_addr: Some(http_rpc_addr),
         http_rpc_enabled: Some(true),
         ws_rpc_addr: Some(ws_rpc_addr),
         ws_rpc_enabled: Some(false),
-        storage_backend: Some(backend_type.clone()),
-        node_type: Some(node_type),
-        max_tip_age: Some(max_tip_age),
     };
     let datadir_opt = Some(data_dir.path().into());
     let config = NodeConfigFile::read(&config_path, &datadir_opt, &options).unwrap();
@@ -138,6 +142,8 @@ fn read_config_override_values() {
     assert_eq!(config.p2p.added_nodes, Some(vec!(p2p_add_node.to_owned())));
     assert_eq!(config.p2p.ban_threshold, Some(p2p_ban_threshold));
     assert_eq!(config.p2p.outbound_connection_timeout, Some(p2p_timeout));
+    assert_eq!(config.p2p.ping_check_period, Some(p2p_ping_check_period));
+    assert_eq!(config.p2p.ping_timeout, Some(p2p_ping_timeout));
     assert_eq!(config.p2p.node_type, Some(node_type));
 
     assert_eq!(config.rpc.http_bind_address, Some(http_rpc_addr));
@@ -196,6 +202,8 @@ fn custom_config_path_and_data_dir() {
 
 fn default_run_options() -> RunOptions {
     RunOptions {
+        storage_backend: None,
+        node_type: None,
         max_db_commit_attempts: None,
         max_orphan_blocks: None,
         tx_index_enabled: None,
@@ -203,12 +211,12 @@ fn default_run_options() -> RunOptions {
         p2p_add_node: None,
         p2p_ban_threshold: None,
         p2p_outbound_connection_timeout: None,
+        p2p_ping_check_period: None,
+        p2p_ping_timeout: None,
+        max_tip_age: None,
         http_rpc_addr: None,
         http_rpc_enabled: None,
         ws_rpc_addr: None,
         ws_rpc_enabled: None,
-        storage_backend: None,
-        node_type: None,
-        max_tip_age: None,
     }
 }
diff --git a/p2p/backend-test-suite/src/block_announcement.rs b/p2p/backend-test-suite/src/block_announcement.rs
index 621d5fb1c..45c052011 100644
--- a/p2p/backend-test-suite/src/block_announcement.rs
+++ b/p2p/backend-test-suite/src/block_announcement.rs
@@ -135,6 +135,8 @@ where
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
+        ping_check_period: Default::default(),
+        ping_timeout: Default::default(),
         node_type: NodeType::Inactive.into(),
         allow_discover_private_ips: Default::default(),
     });
diff --git a/p2p/backend-test-suite/src/peer_manager_peerdb.rs b/p2p/backend-test-suite/src/peer_manager_peerdb.rs
index dd969df4b..5dfcda682 100644
--- a/p2p/backend-test-suite/src/peer_manager_peerdb.rs
+++ b/p2p/backend-test-suite/src/peer_manager_peerdb.rs
@@ -20,97 +20,15 @@ use std::{sync::Arc, time::Duration};
 
 use p2p::{
     config::P2pConfig,
-    net::{
-        default_backend::types::PeerId,
-        types::{PeerInfo, PubSubTopic, Role},
-        AsBannableAddress, NetworkingService,
-    },
+    net::{default_backend::types::PeerId, AsBannableAddress, NetworkingService},
     peer_manager::peerdb::{
-        storage::{PeerDbStorage, PeerDbStorageRead, PeerDbTransactional},
+        storage::{PeerDbStorageRead, PeerDbTransactional},
         PeerDb,
     },
-    testing_utils::{peerdb_inmemory_store, RandomAddressMaker},
+    testing_utils::{peerdb_inmemory_store, P2pTestTimeGetter, RandomAddressMaker},
 };
 
-tests![
-    adjust_peer_score_normal_threshold,
-    adjust_peer_score_higher_threshold,
-    adjust_peer_score_lower_threshold,
-    unban_peer,
-];
-
-async fn adjust_peer_score_normal_threshold<T, N, A>()
-where
-    N: NetworkingService<PeerId = PeerId>,
-    A: RandomAddressMaker<Address = N::Address>,
-{
-    let mut peerdb = PeerDb::<N, _>::new(
-        Arc::new(P2pConfig::default()),
-        Default::default(),
-        peerdb_inmemory_store(),
-    )
-    .unwrap();
-
-    let peer_id = add_active_peer::<T, N, A, _>(&mut peerdb);
-    let address = peerdb.peer_address(&peer_id).unwrap().as_bannable();
-    assert!(peerdb.adjust_peer_score(&peer_id, 100).unwrap());
-    assert!(peerdb.is_address_banned(&address).unwrap());
-}
-
-async fn adjust_peer_score_higher_threshold<T, N, A>()
-where
-    N: NetworkingService<PeerId = PeerId>,
-    A: RandomAddressMaker<Address = N::Address>,
-{
-    let config = P2pConfig {
-        bind_addresses: Default::default(),
-        added_nodes: Default::default(),
-        ban_threshold: 200.into(),
-        ban_duration: Default::default(),
-        outbound_connection_timeout: Default::default(),
-        node_type: Default::default(),
-        allow_discover_private_ips: Default::default(),
-    };
-    let mut peerdb = PeerDb::<N, _>::new(
-        Arc::new(config),
-        Default::default(),
-        peerdb_inmemory_store(),
-    )
-    .unwrap();
-
-    let peer_id = add_active_peer::<T, N, A, _>(&mut peerdb);
-    assert!(!peerdb.adjust_peer_score(&peer_id, 100).unwrap());
-
-    let address = peerdb.peer_address(&peer_id).unwrap().as_bannable();
-    assert!(!peerdb.is_address_banned(&address).unwrap());
-}
-
-async fn adjust_peer_score_lower_threshold<T, N, A>()
-where
-    N: NetworkingService<PeerId = PeerId>,
-    A: RandomAddressMaker<Address = N::Address>,
-{
-    let config = P2pConfig {
-        bind_addresses: Default::default(),
-        added_nodes: Default::default(),
-        ban_threshold: 20.into(),
-        ban_duration: Default::default(),
-        outbound_connection_timeout: Default::default(),
-        node_type: Default::default(),
-        allow_discover_private_ips: Default::default(),
-    };
-    let mut peerdb = PeerDb::<N, _>::new(
-        Arc::new(config),
-        Default::default(),
-        peerdb_inmemory_store(),
-    )
-    .unwrap();
-
-    let peer_id = add_active_peer::<T, N, A, _>(&mut peerdb);
-    let address = peerdb.peer_address(&peer_id).unwrap().as_bannable();
-    assert!(peerdb.adjust_peer_score(&peer_id, 30).unwrap());
-    assert!(peerdb.is_address_banned(&address).unwrap());
-}
+tests![unban_peer,];
 
 async fn unban_peer<T, N, A>()
 where
@@ -118,26 +36,28 @@ where
     A: RandomAddressMaker<Address = N::Address>,
 {
     let db_store = peerdb_inmemory_store();
+    let time_getter = P2pTestTimeGetter::new();
     let mut peerdb = PeerDb::<N, _>::new(
         Arc::new(P2pConfig {
             bind_addresses: Default::default(),
             added_nodes: Default::default(),
             ban_threshold: Default::default(),
-            ban_duration: Duration::from_secs(2).into(),
+            ban_duration: Duration::from_secs(60).into(),
             outbound_connection_timeout: Default::default(),
+            ping_check_period: Default::default(),
+            ping_timeout: Default::default(),
             node_type: Default::default(),
             allow_discover_private_ips: Default::default(),
         }),
-        Default::default(),
+        time_getter.get_time_getter(),
         db_store,
     )
     .unwrap();
 
-    let peer_id = add_active_peer::<T, N, A, _>(&mut peerdb);
-    let address = peerdb.peer_address(&peer_id).unwrap().as_bannable();
-    assert!(peerdb.adjust_peer_score(&peer_id, 100).unwrap());
+    let address = A::new();
+    peerdb.ban_peer(&address).unwrap();
 
-    assert!(peerdb.is_address_banned(&address).unwrap());
+    assert!(peerdb.is_address_banned(&address.as_bannable()).unwrap());
     let banned_addresses = peerdb
         .get_storage_mut()
         .transaction_ro()
@@ -146,9 +66,9 @@ where
         .unwrap();
     assert_eq!(banned_addresses.len(), 1);
 
-    tokio::time::sleep(Duration::from_secs(4)).await;
+    time_getter.advance_time(Duration::from_secs(120)).await;
 
-    assert!(!peerdb.is_address_banned(&address).unwrap());
+    assert!(!peerdb.is_address_banned(&address.as_bannable()).unwrap());
     let banned_addresses = peerdb
         .get_storage_mut()
         .transaction_ro()
@@ -157,34 +77,3 @@ where
         .unwrap();
     assert_eq!(banned_addresses.len(), 0);
 }
-
-fn make_peer_info<T, N, A>() -> (N::PeerId, N::Address, PeerInfo<N::PeerId>)
-where
-    N: NetworkingService<PeerId = PeerId>,
-    A: RandomAddressMaker<Address = N::Address>,
-{
-    let peer_id = PeerId::new();
-
-    (
-        peer_id,
-        A::new(),
-        PeerInfo::<N::PeerId> {
-            peer_id,
-            network: [1, 2, 3, 4],
-            version: common::primitives::semver::SemVer::new(0, 1, 0),
-            agent: None,
-            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
-        },
-    )
-}
-
-fn add_active_peer<T, N, A, S>(peerdb: &mut PeerDb<N, S>) -> N::PeerId
-where
-    N: NetworkingService<PeerId = PeerId>,
-    A: RandomAddressMaker<Address = N::Address>,
-    S: PeerDbStorage,
-{
-    let (peer_id, address, info) = make_peer_info::<T, N, A>();
-    peerdb.peer_connected(address, Role::Inbound, info);
-    peer_id
-}
diff --git a/p2p/src/peer_manager/tests/ban.rs b/p2p/src/peer_manager/tests/ban.rs
index 14d48d418..4825c18fd 100644
--- a/p2p/src/peer_manager/tests/ban.rs
+++ b/p2p/src/peer_manager/tests/ban.rs
@@ -24,6 +24,7 @@ use crate::{
     },
 };
 use common::{chain::config, primitives::semver::SemVer};
+use tokio::sync::oneshot;
 
 use crate::{
     error::{P2pError, PeerError},
@@ -164,32 +165,27 @@ where
     let peer_id = peer_info1.peer_id;
     pm2.accept_inbound_connection(address, peer_info1, None).await.unwrap();
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
-    let addr1 = pm1.peer_connectivity_handle.local_addresses()[0].clone().as_bannable();
-    assert!(pm2.peerdb.is_address_banned(&addr1).unwrap());
-    let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
-    assert!(matches!(
-        event,
-        Ok(net::types::ConnectivityEvent::ConnectionClosed { .. })
-    ));
-
     let remote_addr = pm1.peer_connectivity_handle.local_addresses()[0].clone();
 
-    tokio::spawn(async move {
-        loop {
-            let _ = pm1.peer_connectivity_handle.poll_next().await.unwrap();
-        }
-    });
-
-    pm2.peer_connectivity_handle.connect(remote_addr.clone()).await.unwrap();
-    if let Ok(net::types::ConnectivityEvent::ConnectionError { address, error }) =
-        pm2.peer_connectivity_handle.poll_next().await
-    {
-        assert_eq!(remote_addr, address);
-        assert!(matches!(
-            error,
-            P2pError::PeerError(PeerError::BannedPeer(_))
-        ));
+    assert_eq!(pm2.adjust_peer_score(peer_id, 10).await, Ok(()));
+    assert!(!pm2.peerdb.is_address_banned(&remote_addr.as_bannable()).unwrap());
+
+    assert_eq!(pm2.adjust_peer_score(peer_id, 90).await, Ok(()));
+    assert!(pm2.peerdb.is_address_banned(&remote_addr.as_bannable()).unwrap());
+
+    let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
+    match &event {
+        Ok(net::types::ConnectivityEvent::ConnectionClosed { .. }) => {}
+        _ => panic!("unexpected event: {event:?}"),
+    }
+    pm2.handle_connectivity_event_result(event).await.unwrap();
+
+    let (tx, rx) = oneshot::channel();
+    pm2.connect(remote_addr, Some(tx)).await.unwrap();
+    let res = rx.await.unwrap();
+    match res {
+        Err(P2pError::PeerError(PeerError::BannedAddress(_))) => {}
+        _ => panic!("unexpected result: {res:?}"),
     }
 }
 
@@ -241,7 +237,7 @@ where
         )
         .await;
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
-    assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
+    assert!(!peer_manager.is_peer_connected(&peer_id));
 
     // invalid version
     let res = peer_manager
@@ -261,7 +257,7 @@ where
         )
         .await;
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
-    assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
+    assert!(!peer_manager.is_peer_connected(&peer_id));
 
     // valid connection
     let address = B::new();
@@ -283,7 +279,7 @@ where
         .await;
     assert!(res.is_ok());
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
-    assert!(peer_manager.peerdb.is_active_peer(&peer_id));
+    assert!(peer_manager.is_peer_connected(&peer_id));
     assert!(!peer_manager.peerdb.is_address_banned(&address.as_bannable()).unwrap());
 }
 
@@ -345,7 +341,7 @@ where
         )
         .await;
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
-    assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
+    assert!(!peer_manager.is_peer_connected(&peer_id));
 
     // invalid version
     let res = peer_manager
@@ -364,7 +360,7 @@ where
         )
         .await;
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
-    assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
+    assert!(!peer_manager.is_peer_connected(&peer_id));
 
     // valid connection
     let address = B::new();
@@ -385,7 +381,7 @@ where
         .await;
     assert!(res.is_ok());
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
-    assert!(peer_manager.peerdb.is_active_peer(&peer_id));
+    assert!(peer_manager.is_peer_connected(&peer_id));
     assert!(!peer_manager.peerdb.is_address_banned(&address.as_bannable()).unwrap());
 }
 
@@ -452,10 +448,10 @@ where
     tokio::spawn(async move { pm1.run().await });
 
     let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
-    if let Ok(net::types::ConnectivityEvent::ConnectionClosed { peer_id }) = event {
-        assert_eq!(peer_id, peer_info.peer_id);
-    } else {
-        panic!("invalid event received");
+    match event {
+        Ok(net::types::ConnectivityEvent::ConnectionClosed { peer_id })
+            if peer_id == peer_info.peer_id => {}
+        _ => panic!("unexpected event: {event:?}"),
     }
 }
 
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index 925addf65..150aace80 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -60,7 +60,7 @@ async fn test_peer_manager_connect<T: NetworkingService>(
     let config = Arc::new(config::create_mainnet());
     let mut peer_manager = make_peer_manager::<T>(transport, bind_addr, config).await;
 
-    peer_manager.connect(remote_addr).await.unwrap();
+    peer_manager.try_connect(remote_addr).await.unwrap();
 
     assert!(matches!(
         peer_manager.peer_connectivity_handle.poll_next().await,
@@ -126,7 +126,7 @@ where
     pm1.peerdb.peer_discovered(&addr).unwrap();
     pm1.heartbeat().await.unwrap();
 
-    assert_eq!(pm1.pending.len(), 1);
+    assert_eq!(pm1.pending_connects.len(), 1);
     assert!(std::matches!(
         pm1.peer_connectivity_handle.poll_next().await,
         Ok(net::types::ConnectivityEvent::OutboundAccepted { .. })
@@ -418,11 +418,11 @@ where
     let mut pm1 = make_peer_manager::<T>(A::make_transport(), addr1, Arc::clone(&config)).await;
     let mut pm2 = make_peer_manager::<T>(A::make_transport(), addr2, Arc::clone(&config)).await;
 
-    peers.into_iter().for_each(|peer| {
-        pm1.peerdb.peer_connected(peer.0, Role::Inbound, peer.1);
-    });
+    for peer in peers.into_iter() {
+        pm1.accept_connection(peer.0, Role::Inbound, peer.1, None).await.unwrap();
+    }
     assert_eq!(
-        pm1.peerdb.active_peer_count(),
+        pm1.active_peer_count(),
         peer_manager::MAX_ACTIVE_CONNECTIONS
     );
 
@@ -678,6 +678,8 @@ where
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
+        ping_check_period: Default::default(),
+        ping_timeout: Default::default(),
         node_type: Default::default(),
         allow_discover_private_ips: Default::default(),
     });
@@ -703,6 +705,8 @@ where
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
+        ping_check_period: Default::default(),
+        ping_timeout: Default::default(),
         node_type: Default::default(),
         allow_discover_private_ips: Default::default(),
     });
@@ -767,6 +771,8 @@ where
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
+        ping_check_period: Default::default(),
+        ping_timeout: Default::default(),
         node_type: Default::default(),
         allow_discover_private_ips: true.into(),
     });
@@ -793,6 +799,8 @@ where
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
+        ping_check_period: Default::default(),
+        ping_timeout: Default::default(),
         node_type: Default::default(),
         allow_discover_private_ips: true.into(),
     });
@@ -812,6 +820,8 @@ where
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
+        ping_check_period: Default::default(),
+        ping_timeout: Default::default(),
         node_type: Default::default(),
         allow_discover_private_ips: true.into(),
     });
diff --git a/p2p/src/peer_manager/tests/mod.rs b/p2p/src/peer_manager/tests/mod.rs
index 11095b071..66a966ed9 100644
--- a/p2p/src/peer_manager/tests/mod.rs
+++ b/p2p/src/peer_manager/tests/mod.rs
@@ -15,6 +15,7 @@
 
 mod ban;
 mod connections;
+mod ping;
 
 use std::{sync::Arc, time::Duration};
 
diff --git a/p2p/src/peer_manager/tests/ping.rs b/p2p/src/peer_manager/tests/ping.rs
new file mode 100644
index 000000000..6091af9e1
--- /dev/null
+++ b/p2p/src/peer_manager/tests/ping.rs
@@ -0,0 +1,135 @@
+// Copyright (c) 2023 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
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// 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.
+
+use std::sync::Arc;
+
+use common::chain::config;
+
+use crate::{
+    config::P2pConfig,
+    event::PeerManagerEvent,
+    message::{PeerManagerResponse, PingRequest, PingResponse, Request},
+    net::{
+        default_backend::{
+            transport::TcpTransportSocket,
+            types::{Command, ConnectivityEvent, PeerId},
+            ConnectivityHandle, DefaultNetworkingService,
+        },
+        types::PeerInfo,
+    },
+    peer_manager::PeerManager,
+    testing_utils::{peerdb_inmemory_store, P2pTestTimeGetter},
+};
+
+#[tokio::test]
+async fn ping_timeout() {
+    type TestNetworkingService = DefaultNetworkingService<TcpTransportSocket>;
+
+    let chain_config = Arc::new(config::create_mainnet());
+    let p2p_config: Arc<P2pConfig> = Arc::new(Default::default());
+    let ping_check_period = *p2p_config.ping_check_period;
+    let ping_timeout = *p2p_config.ping_timeout;
+
+    let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::channel(10);
+    let (conn_tx, conn_rx) = tokio::sync::mpsc::channel(10);
+    let (_peer_tx, peer_rx) =
+        tokio::sync::mpsc::unbounded_channel::<PeerManagerEvent<TestNetworkingService>>();
+    let time_getter = P2pTestTimeGetter::new();
+    let (sync_tx, _sync_rx) = tokio::sync::mpsc::unbounded_channel();
+    let connectivity_handle = ConnectivityHandle::<TestNetworkingService, TcpTransportSocket>::new(
+        vec![],
+        cmd_tx,
+        conn_rx,
+    );
+
+    let mut peer_manager = PeerManager::new(
+        Arc::clone(&chain_config),
+        p2p_config,
+        connectivity_handle,
+        peer_rx,
+        sync_tx,
+        time_getter.get_time_getter(),
+        peerdb_inmemory_store(),
+    )
+    .unwrap();
+
+    tokio::spawn(async move {
+        let _ = peer_manager.run().await;
+    });
+
+    // Notify about new inbound connection
+    conn_tx
+        .send(ConnectivityEvent::InboundAccepted {
+            address: "123.123.123.123:12345".parse().unwrap(),
+            peer_info: PeerInfo {
+                peer_id: PeerId::new(),
+                network: *chain_config.magic_bytes(),
+                version: *chain_config.version(),
+                agent: None,
+                subscriptions: Default::default(),
+            },
+            receiver_address: None,
+        })
+        .await
+        .unwrap();
+
+    // Receive ping requests and send responses normally
+    for _ in 0..30 {
+        time_getter.advance_time(ping_check_period).await;
+
+        let event = cmd_rx.recv().await.unwrap();
+        match event {
+            Command::SendRequest {
+                peer_id,
+                request_id,
+                message: Request::PingRequest(PingRequest { nonce }),
+            } => {
+                conn_tx
+                    .send(ConnectivityEvent::Response {
+                        peer_id,
+                        request_id,
+                        response: PeerManagerResponse::PingResponse(PingResponse { nonce }),
+                    })
+                    .await
+                    .unwrap();
+            }
+            _ => panic!("unexpected event: {event:?}"),
+        }
+    }
+
+    // Receive one more ping request but do not send a ping response
+    time_getter.advance_time(ping_check_period).await;
+    let event = cmd_rx.recv().await.unwrap();
+    match event {
+        Command::SendRequest {
+            peer_id: _,
+            request_id: _,
+            message: Request::PingRequest(PingRequest { nonce: _ }),
+        } => {}
+        _ => panic!("unexpected event: {event:?}"),
+    }
+
+    time_getter.advance_time(ping_timeout).await;
+
+    // PeerManager should ask backend to close connection
+    let event = cmd_rx.recv().await.unwrap();
+    match event {
+        Command::Disconnect { peer_id, response } => {
+            response.send(Ok(())).unwrap();
+            conn_tx.send(ConnectivityEvent::ConnectionClosed { peer_id }).await.unwrap();
+        }
+        _ => panic!("unexpected event: {event:?}"),
+    }
+}
diff --git a/p2p/src/sync/tests/mod.rs b/p2p/src/sync/tests/mod.rs
index 76890f71d..eb48dd442 100644
--- a/p2p/src/sync/tests/mod.rs
+++ b/p2p/src/sync/tests/mod.rs
@@ -75,6 +75,8 @@ where
         ban_threshold: 100.into(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
+        ping_check_period: Default::default(),
+        ping_timeout: Default::default(),
         node_type: NodeType::Full.into(),
         allow_discover_private_ips: Default::default(),
     });
diff --git a/p2p/src/testing_utils.rs b/p2p/src/testing_utils.rs
index deec1b049..93a32c0d3 100644
--- a/p2p/src/testing_utils.rs
+++ b/p2p/src/testing_utils.rs
@@ -254,10 +254,8 @@ impl P2pTestTimeGetter {
 
     pub async fn advance_time(&self, duration: Duration) {
         tokio::time::pause();
-        self.current_time_millis.store(
-            self.current_time_millis.load(Ordering::SeqCst) + duration.as_millis() as u64,
-            Ordering::SeqCst,
-        );
+        self.current_time_millis
+            .fetch_add(duration.as_millis() as u64, Ordering::SeqCst);
         tokio::time::advance(duration).await;
         tokio::time::resume();
     }