Repository: mintlayer/mintlayer-core
Pull request #725 (2023-02-24 14:19:19)

`PeerManager` updates
I made some updates to `PeerManager` and `PeerDb` to implement some missing basic parts.
`PeerManager` will now load IPs from DNS seeds on first start and gracefully reconnect to peers.
Added two separate inbound and outbound peer limits (inbound limit is configurable).
I also removed error reporting from `PeerManager` functions where possible to simplify things.
This PR depends on https://github.com/mintlayer/mintlayer-core/pull/727.


Patch:
diff --git a/.github/scripts/codecheck.py b/.github/scripts/codecheck.py
index afa8ab625..2de423409 100755
--- a/.github/scripts/codecheck.py
+++ b/.github/scripts/codecheck.py
@@ -140,9 +140,9 @@ def check_todos():
 
         with open(path) as file:
             file_data = file.read()
-            if 'TODO(PR)' in file_data or 'FIXME' in file_data:
+            if 'TODO(PR)' in file_data or 'FIXME' in file_data or 'todo!()' in file_data:
                 ok = False
-                print("{}: Found TODO(PR) or FIXME instances".format(path))
+                print("{}: Found TODO(PR) or FIXME or todo!() instances".format(path))
 
     print()
     return ok
diff --git a/dns_server/src/crawler_p2p/crawler_manager/mod.rs b/dns_server/src/crawler_p2p/crawler_manager/mod.rs
index 66641d753..067bc6aa3 100644
--- a/dns_server/src/crawler_p2p/crawler_manager/mod.rs
+++ b/dns_server/src/crawler_p2p/crawler_manager/mod.rs
@@ -175,7 +175,6 @@ where
                     .expect("send_message must succeed");
             }
             PeerManagerMessage::AddrListResponse(_) => {}
-            PeerManagerMessage::AnnounceAddrResponse(_) => {}
             PeerManagerMessage::PingResponse(_) => {}
         }
     }
diff --git a/dns_server/src/main.rs b/dns_server/src/main.rs
index 473b14d26..52c064ebc 100644
--- a/dns_server/src/main.rs
+++ b/dns_server/src/main.rs
@@ -42,6 +42,7 @@ async fn run(config: Arc<DnsServerConfig>) -> Result<void::Void, error::DnsServe
     let p2p_config = Arc::new(P2pConfig {
         bind_addresses: Vec::new(),
         added_nodes: Vec::new(),
+        max_inbound_connections: Default::default(),
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
diff --git a/node/src/config_files/mod.rs b/node/src/config_files/mod.rs
index f9574514b..0433e9651 100644
--- a/node/src/config_files/mod.rs
+++ b/node/src/config_files/mod.rs
@@ -127,6 +127,7 @@ fn p2p_config(config: P2pConfigFile, options: &RunOptions) -> P2pConfigFile {
     let P2pConfigFile {
         bind_addresses,
         added_nodes,
+        max_inbound_connections,
         ban_threshold,
         ban_duration,
         outbound_connection_timeout,
@@ -137,6 +138,7 @@ fn p2p_config(config: P2pConfigFile, options: &RunOptions) -> P2pConfigFile {
 
     let bind_addresses = options.p2p_addr.clone().or(bind_addresses);
     let added_nodes = options.p2p_add_node.clone().or(added_nodes);
+    let max_inbound_connections = options.p2p_max_inbound_connections.or(max_inbound_connections);
     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);
@@ -147,6 +149,7 @@ fn p2p_config(config: P2pConfigFile, options: &RunOptions) -> P2pConfigFile {
     P2pConfigFile {
         bind_addresses,
         added_nodes,
+        max_inbound_connections,
         ban_threshold,
         ban_duration,
         outbound_connection_timeout,
diff --git a/node/src/config_files/p2p.rs b/node/src/config_files/p2p.rs
index 11bf26a53..8cd393d47 100644
--- a/node/src/config_files/p2p.rs
+++ b/node/src/config_files/p2p.rs
@@ -56,6 +56,8 @@ pub struct P2pConfigFile {
     pub bind_addresses: Option<Vec<String>>,
     /// Optional list of initial node addresses to connect.
     pub added_nodes: Option<Vec<String>>,
+    /// Maximum allowed number of inbound connections.
+    pub max_inbound_connections: Option<usize>,
     /// The score threshold after which a peer is banned.
     pub ban_threshold: Option<u32>,
     /// Duration of bans in seconds.
@@ -75,6 +77,7 @@ impl From<P2pConfigFile> for P2pConfig {
         P2pConfig {
             bind_addresses: c.bind_addresses.clone().unwrap_or_default(),
             added_nodes: c.added_nodes.clone().unwrap_or_default(),
+            max_inbound_connections: c.max_inbound_connections.into(),
             ban_threshold: c.ban_threshold.into(),
             ban_duration: c.ban_duration.map(Duration::from_secs).into(),
             outbound_connection_timeout: c
diff --git a/node/src/options.rs b/node/src/options.rs
index 12f169a9f..93b9d0f19 100644
--- a/node/src/options.rs
+++ b/node/src/options.rs
@@ -81,6 +81,10 @@ pub struct RunOptions {
     #[clap(long, value_name = "NODE")]
     pub p2p_add_node: Option<Vec<String>>,
 
+    /// Maximum allowed number of inbound connections.
+    #[clap(long)]
+    pub p2p_max_inbound_connections: Option<usize>,
+
     /// The p2p score threshold after which a peer is baned.
     #[clap(long)]
     pub p2p_ban_threshold: Option<u32>,
diff --git a/p2p/src/config.rs b/p2p/src/config.rs
index f05a3d09a..3e8af024d 100644
--- a/p2p/src/config.rs
+++ b/p2p/src/config.rs
@@ -19,6 +19,7 @@ use utils::make_config_setting;
 
 use crate::net::types::PubSubTopic;
 
+make_config_setting!(MaxInboundConnections, usize, 128);
 make_config_setting!(BanThreshold, u32, 100);
 make_config_setting!(BanDuration, Duration, Duration::from_secs(60 * 60 * 24));
 make_config_setting!(OutboundConnectionTimeout, Duration, Duration::from_secs(10));
@@ -37,7 +38,7 @@ pub enum NodeType {
     Full,
     /// A node that only download blocks, but ignores transactions.
     BlocksOnly,
-    /// A node that only is interested in network address announcements
+    /// A node interested only in network address announcements.
     DnsServer,
     /// A node that doesn't subscribe to any events.
     ///
@@ -67,6 +68,8 @@ pub struct P2pConfig {
     pub bind_addresses: Vec<String>,
     /// Optional list of initial node addresses, could be used to specify boot nodes for example.
     pub added_nodes: Vec<String>,
+    /// Maximum allowed number of inbound connections.
+    pub max_inbound_connections: MaxInboundConnections,
     /// The score threshold after which a peer is banned.
     pub ban_threshold: BanThreshold,
     /// Duration of bans in seconds.
diff --git a/p2p/src/error.rs b/p2p/src/error.rs
index 1e266bfd1..be1100dff 100644
--- a/p2p/src/error.rs
+++ b/p2p/src/error.rs
@@ -43,8 +43,6 @@ pub enum ProtocolError {
     DisconnectedHeaders,
     #[error("Received a message ({0}) that wasn't expected")]
     UnexpectedMessage(&'static str),
-    #[error("Unknown request identifier")]
-    UnknownRequestId,
 }
 
 /// Peer state errors (Errors either for an individual peer or for the [`PeerManager`](crate::peer_manager::PeerManager))
@@ -82,8 +80,6 @@ pub enum DialError {
     ConnectionRefusedOrTimedOut,
     #[error("I/O error: `{0:?}`")]
     IoError(std::io::ErrorKind),
-    #[error("Accept failed")]
-    AcceptFailed,
 }
 
 /// Conversion errors
@@ -119,6 +115,8 @@ pub enum P2pError {
     NoiseHandshakeError(String),
     #[error("The configuration value is invalid: {0}")]
     InvalidConfigurationValue(String),
+    #[error("The storage state is invalid: {0}")]
+    InvalidStorageState(String),
 }
 
 impl From<std::io::Error> for P2pError {
@@ -172,6 +170,7 @@ impl BanScore for P2pError {
             // Could be a noise protocol violation but also a network error, do not ban peer
             P2pError::NoiseHandshakeError(_) => 0,
             P2pError::InvalidConfigurationValue(_) => 0,
+            P2pError::InvalidStorageState(_) => 0,
         }
     }
 }
@@ -189,7 +188,6 @@ impl BanScore for ProtocolError {
             ProtocolError::UnknownBlockRequested => 20,
             ProtocolError::DisconnectedHeaders => 20,
             ProtocolError::UnexpectedMessage(_) => 20,
-            ProtocolError::UnknownRequestId => 20,
         }
     }
 }
diff --git a/p2p/src/message.rs b/p2p/src/message.rs
index 75463210b..0c57f15c1 100644
--- a/p2p/src/message.rs
+++ b/p2p/src/message.rs
@@ -36,7 +36,6 @@ pub enum PeerManagerMessage {
     AnnounceAddrRequest(AnnounceAddrRequest),
     PingRequest(PingRequest),
     AddrListResponse(AddrListResponse),
-    AnnounceAddrResponse(AnnounceAddrResponse),
     PingResponse(PingResponse),
 }
 
@@ -131,9 +130,6 @@ impl BlockResponse {
     }
 }
 
-#[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
-pub struct AnnounceAddrResponse {}
-
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub struct AddrListResponse {
     pub addresses: Vec<PeerAddress>,
diff --git a/p2p/src/net/default_backend/backend.rs b/p2p/src/net/default_backend/backend.rs
index c6ad386f9..15bb47e75 100644
--- a/p2p/src/net/default_backend/backend.rs
+++ b/p2p/src/net/default_backend/backend.rs
@@ -263,14 +263,20 @@ where
                 },
                 // Accept a new peer connection.
                 res = self.socket.accept() => {
-                    let (stream, address) = res.map_err(|_| P2pError::DialError(DialError::AcceptFailed))?;
-
-                    self.create_pending_peer(
-                        stream,
-                        PeerId::new(),
-                        PeerRole::Inbound,
-                        address,
-                    )?;
+                    match res {
+                        Ok((stream, address)) => {
+                            self.create_pending_peer(
+                                stream,
+                                PeerId::new(),
+                                PeerRole::Inbound,
+                                address,
+                            )?;
+                        },
+                        Err(err) => {
+                            // Just log the error and let the node continue working
+                            log::error!("Accepting a new connection failed unexpectedly: {err}")
+                        },
+                    }
                 }
             }
         }
@@ -508,10 +514,6 @@ where
                 peer,
                 message: PeerManagerMessage::AddrListResponse(r),
             })?,
-            Message::AnnounceAddrResponse(r) => self.conn_tx.send(ConnectivityEvent::Message {
-                peer,
-                message: PeerManagerMessage::AnnounceAddrResponse(r),
-            })?,
             Message::PingResponse(r) => self.conn_tx.send(ConnectivityEvent::Message {
                 peer,
                 message: PeerManagerMessage::PingResponse(r),
diff --git a/p2p/src/net/default_backend/mod.rs b/p2p/src/net/default_backend/mod.rs
index 281e2e050..6548fab19 100644
--- a/p2p/src/net/default_backend/mod.rs
+++ b/p2p/src/net/default_backend/mod.rs
@@ -212,272 +212,4 @@ impl<S: NetworkingService, T: TransportSocket> SyncingMessagingService<S>
 }
 
 #[cfg(test)]
-mod tests {
-    use super::{transport::NoiseTcpTransport, *};
-    use crate::error::DialError;
-    use crate::testing_utils::{TestTransportChannel, TestTransportMaker, TestTransportTcp};
-    use crate::{
-        net::default_backend::transport::{MpscChannelTransport, TcpTransportSocket},
-        testing_utils::TestTransportNoise,
-    };
-    use common::primitives::semver::SemVer;
-    use std::fmt::Debug;
-
-    async fn connect_to_remote<A, T>()
-    where
-        A: TestTransportMaker<Transport = T, Address = T::Address>,
-        T: TransportSocket + Debug,
-    {
-        let config = Arc::new(common::chain::config::create_mainnet());
-        let p2p_config: Arc<config::P2pConfig> = Arc::new(Default::default());
-
-        let (mut conn1, _) = DefaultNetworkingService::<T>::start(
-            A::make_transport(),
-            vec![A::make_address()],
-            Arc::clone(&config),
-            Arc::clone(&p2p_config),
-        )
-        .await
-        .unwrap();
-
-        let (conn2, _) = DefaultNetworkingService::<T>::start(
-            A::make_transport(),
-            vec![A::make_address()],
-            Arc::clone(&config),
-            Arc::clone(&p2p_config),
-        )
-        .await
-        .unwrap();
-
-        let addr = conn2.local_addresses();
-        assert_eq!(conn1.connect(addr[0].clone()), Ok(()));
-
-        if let Ok(ConnectivityEvent::OutboundAccepted {
-            address,
-            peer_info,
-            receiver_address: _,
-        }) = conn1.poll_next().await
-        {
-            assert_eq!(address, conn2.local_addresses()[0]);
-            assert_eq!(peer_info.network, *config.magic_bytes());
-            assert_eq!(peer_info.version, SemVer::new(0, 1, 0));
-            assert_eq!(peer_info.agent, None);
-            assert_eq!(
-                peer_info.subscriptions,
-                [PubSubTopic::Blocks, PubSubTopic::Transactions, PubSubTopic::PeerAddresses]
-                    .into_iter()
-                    .collect()
-            );
-        } else {
-            panic!("invalid event received");
-        }
-    }
-
-    #[tokio::test]
-    async fn connect_to_remote_tcp() {
-        connect_to_remote::<TestTransportTcp, TcpTransportSocket>().await;
-    }
-
-    #[tokio::test]
-    async fn connect_to_remote_channels() {
-        connect_to_remote::<TestTransportChannel, MpscChannelTransport>().await;
-    }
-
-    #[tokio::test]
-    async fn connect_to_remote_noise() {
-        connect_to_remote::<TestTransportNoise, NoiseTcpTransport>().await;
-    }
-
-    async fn accept_incoming<A, T>()
-    where
-        A: TestTransportMaker<Transport = T, Address = T::Address>,
-        T: TransportSocket,
-    {
-        let config = Arc::new(common::chain::config::create_mainnet());
-        let p2p_config: Arc<config::P2pConfig> = Arc::new(Default::default());
-
-        let (mut conn1, _) = DefaultNetworkingService::<T>::start(
-            A::make_transport(),
-            vec![A::make_address()],
-            Arc::clone(&config),
-            Arc::clone(&p2p_config),
-        )
-        .await
-        .unwrap();
-
-        let (mut conn2, _) = DefaultNetworkingService::<T>::start(
-            A::make_transport(),
-            vec![A::make_address()],
-            Arc::clone(&config),
-            Arc::clone(&p2p_config),
-        )
-        .await
-        .unwrap();
-
-        let bind_address = conn2.local_addresses();
-        conn1.connect(bind_address[0].clone()).unwrap();
-        let res2 = conn2.poll_next().await;
-        match res2.unwrap() {
-            ConnectivityEvent::InboundAccepted {
-                address: _,
-                peer_info,
-                receiver_address: _,
-            } => {
-                assert_eq!(peer_info.network, *config.magic_bytes());
-                assert_eq!(
-                    peer_info.version,
-                    common::primitives::semver::SemVer::new(0, 1, 0),
-                );
-                assert_eq!(peer_info.agent, None);
-            }
-            _ => panic!("invalid event received, expected incoming connection"),
-        }
-    }
-
-    #[tokio::test]
-    async fn accept_incoming_tcp() {
-        accept_incoming::<TestTransportTcp, TcpTransportSocket>().await;
-    }
-
-    #[tokio::test]
-    async fn accept_incoming_channels() {
-        accept_incoming::<TestTransportChannel, MpscChannelTransport>().await;
-    }
-
-    #[tokio::test]
-    async fn accept_incoming_noise() {
-        accept_incoming::<TestTransportNoise, NoiseTcpTransport>().await;
-    }
-
-    async fn disconnect<A, T>()
-    where
-        A: TestTransportMaker<Transport = T, Address = T::Address>,
-        T: TransportSocket,
-    {
-        let config = Arc::new(common::chain::config::create_mainnet());
-        let p2p_config: Arc<config::P2pConfig> = Arc::new(Default::default());
-
-        let (mut conn1, _) = DefaultNetworkingService::<T>::start(
-            A::make_transport(),
-            vec![A::make_address()],
-            Arc::clone(&config),
-            Arc::clone(&p2p_config),
-        )
-        .await
-        .unwrap();
-        let (mut conn2, _) = DefaultNetworkingService::<T>::start(
-            A::make_transport(),
-            vec![A::make_address()],
-            config,
-            p2p_config,
-        )
-        .await
-        .unwrap();
-
-        conn1.connect(conn2.local_addresses()[0].clone()).unwrap();
-        let res2 = conn2.poll_next().await;
-
-        match res2.unwrap() {
-            ConnectivityEvent::InboundAccepted {
-                address: _,
-                peer_info,
-                receiver_address: _,
-            } => {
-                assert_eq!(conn2.disconnect(peer_info.peer_id), Ok(()));
-            }
-            _ => panic!("invalid event received, expected incoming connection"),
-        }
-    }
-
-    #[tokio::test]
-    async fn disconnect_tcp() {
-        disconnect::<TestTransportTcp, TcpTransportSocket>().await;
-    }
-
-    #[tokio::test]
-    async fn disconnect_channels() {
-        disconnect::<TestTransportChannel, MpscChannelTransport>().await;
-    }
-
-    #[tokio::test]
-    async fn disconnect_noise() {
-        disconnect::<TestTransportNoise, NoiseTcpTransport>().await;
-    }
-
-    async fn self_connect<A, T>()
-    where
-        A: TestTransportMaker<Transport = T, Address = T::Address>,
-        T: TransportSocket + Debug,
-    {
-        let config = Arc::new(common::chain::config::create_mainnet());
-        let p2p_config: Arc<config::P2pConfig> = Arc::new(Default::default());
-
-        let (mut conn1, _) = DefaultNetworkingService::<T>::start(
-            A::make_transport(),
-            vec![A::make_address()],
-            Arc::clone(&config),
-            Arc::clone(&p2p_config),
-        )
-        .await
-        .unwrap();
-
-        let (conn2, _) = DefaultNetworkingService::<T>::start(
-            A::make_transport(),
-            vec![A::make_address()],
-            Arc::clone(&config),
-            Arc::clone(&p2p_config),
-        )
-        .await
-        .unwrap();
-
-        // Try connect to self
-        let addr = conn1.local_addresses();
-        assert_eq!(conn1.connect(addr[0].clone()), Ok(()));
-
-        // ConnectionError should be reported
-        if let Ok(ConnectivityEvent::ConnectionError { address, error }) = conn1.poll_next().await {
-            assert_eq!(address, conn1.local_addresses()[0]);
-            assert_eq!(error, P2pError::DialError(DialError::AttemptToDialSelf));
-        } else {
-            panic!("invalid event received");
-        }
-
-        // Check that we can still connect normally after
-        let addr = conn2.local_addresses();
-        assert_eq!(conn1.connect(addr[0].clone()), Ok(()));
-        if let Ok(ConnectivityEvent::OutboundAccepted {
-            address,
-            peer_info,
-            receiver_address: _,
-        }) = conn1.poll_next().await
-        {
-            assert_eq!(address, conn2.local_addresses()[0]);
-            assert_eq!(peer_info.network, *config.magic_bytes());
-            assert_eq!(peer_info.version, SemVer::new(0, 1, 0));
-            assert_eq!(peer_info.agent, None);
-            assert_eq!(
-                peer_info.subscriptions,
-                [PubSubTopic::Blocks, PubSubTopic::Transactions, PubSubTopic::PeerAddresses]
-                    .into_iter()
-                    .collect()
-            );
-        } else {
-            panic!("invalid event received");
-        }
-    }
-
-    #[tokio::test]
-    async fn self_connect_tcp() {
-        self_connect::<TestTransportTcp, TcpTransportSocket>().await;
-    }
-
-    #[tokio::test]
-    async fn self_connect_channels() {
-        self_connect::<TestTransportChannel, MpscChannelTransport>().await;
-    }
-
-    #[tokio::test]
-    async fn self_connect_noise() {
-        self_connect::<TestTransportNoise, NoiseTcpTransport>().await;
-    }
-}
+mod tests;
diff --git a/p2p/src/net/default_backend/peer.rs b/p2p/src/net/default_backend/peer.rs
index 10499bc7b..8368486be 100644
--- a/p2p/src/net/default_backend/peer.rs
+++ b/p2p/src/net/default_backend/peer.rs
@@ -222,8 +222,7 @@ where
                                 types::PeerEvent::MessageReceived {
                                     message
                                 },
-                            ))
-                            .map_err(P2pError::from)?;
+                            ))?;
                     }
                 }
             }
diff --git a/p2p/src/net/default_backend/types.rs b/p2p/src/net/default_backend/types.rs
index 92162b939..8a5712f66 100644
--- a/p2p/src/net/default_backend/types.rs
+++ b/p2p/src/net/default_backend/types.rs
@@ -20,9 +20,9 @@ use serialization::{Decode, Encode};
 
 use crate::{
     message::{
-        AddrListRequest, AddrListResponse, AnnounceAddrRequest, AnnounceAddrResponse, Announcement,
-        BlockListRequest, BlockResponse, HeaderListRequest, HeaderListResponse, PeerManagerMessage,
-        PingRequest, PingResponse, SyncMessage,
+        AddrListRequest, AddrListResponse, AnnounceAddrRequest, Announcement, BlockListRequest,
+        BlockResponse, HeaderListRequest, HeaderListResponse, PeerManagerMessage, PingRequest,
+        PingResponse, SyncMessage,
     },
     net::types::PubSubTopic,
     types::{peer_address::PeerAddress, peer_id::PeerId},
@@ -114,7 +114,6 @@ pub enum Message {
     HeaderListResponse(HeaderListResponse),
     BlockResponse(BlockResponse),
     AddrListResponse(AddrListResponse),
-    AnnounceAddrResponse(AnnounceAddrResponse),
     PingResponse(PingResponse),
     Announcement(Box<Announcement>),
 }
@@ -126,7 +125,6 @@ impl From<PeerManagerMessage> for Message {
             PeerManagerMessage::AnnounceAddrRequest(r) => Message::AnnounceAddrRequest(r),
             PeerManagerMessage::PingRequest(r) => Message::PingRequest(r),
             PeerManagerMessage::AddrListResponse(r) => Message::AddrListResponse(r),
-            PeerManagerMessage::AnnounceAddrResponse(r) => Message::AnnounceAddrResponse(r),
             PeerManagerMessage::PingResponse(r) => Message::PingResponse(r),
         }
     }
diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs
index eee5c9c3b..e3116ebb0 100644
--- a/p2p/src/peer_manager/mod.rs
+++ b/p2p/src/peer_manager/mod.rs
@@ -14,10 +14,6 @@
 // limitations under the License.
 
 //! Peer manager
-//!
-//! TODO
-//!
-//!
 
 pub mod global_ip;
 pub mod peer_context;
@@ -47,7 +43,6 @@ use crate::{
         PingResponse,
     },
     net::{
-        self,
         default_backend::transport::TransportAddress,
         types::{ConnectivityEvent, Role},
         types::{PeerInfo, PubSubTopic},
@@ -65,8 +60,9 @@ use self::{
     peerdb::storage::PeerDbStorage,
 };
 
-/// Maximum number of connections the [`PeerManager`] is allowed to have open
-const MAX_ACTIVE_CONNECTIONS: usize = 128;
+/// Maximum number of outbound connections the [`PeerManager`] is allowed to have open.
+/// This value is constant because users should not change this.
+const MAX_OUTBOUND_CONNECTIONS: usize = 8;
 
 /// Lower bound for how often [`PeerManager::heartbeat()`] is called
 const PEER_MGR_HEARTBEAT_INTERVAL_MIN: Duration = Duration::from_secs(5);
@@ -79,6 +75,12 @@ const MAX_ADDRESS_COUNT: usize = 1000;
 /// To how many peers resend received address
 const PEER_ADDRESS_RESEND_COUNT: usize = 2;
 
+/// Hardcoded seed DNS hostnames
+// TODO: Replace with actual values
+const DNS_SEEDS: [&str; 2] = ["seed.mintlayer.org", "seed2.mintlayer.org"];
+/// Maximum number of records accepted in a single DNS server response
+const MAX_DNS_RECORDS: usize = 10;
+
 pub struct PeerManager<T, S>
 where
     T: NetworkingService,
@@ -99,19 +101,17 @@ where
     tx_sync: mpsc::UnboundedSender<SyncControlEvent>,
 
     /// Hashmap of pending outbound connections
-    pending_connects: HashMap<T::Address, Option<oneshot_nofail::Sender<crate::Result<()>>>>,
+    pending_outbound_connects:
+        HashMap<T::Address, Option<oneshot_nofail::Sender<crate::Result<()>>>>,
 
     /// Hashmap of pending disconnect requests
     pending_disconnects: HashMap<PeerId, Option<oneshot_nofail::Sender<crate::Result<()>>>>,
 
-    /// Set of active peers
+    /// Map of all connected peers
     peers: BTreeMap<PeerId, PeerContext<T::Address>>,
 
     /// Peer database
-    peerdb: peerdb::PeerDb<T, S>,
-
-    /// Last time when heartbeat was called
-    last_heartbeat: Instant,
+    peerdb: peerdb::PeerDb<T::Address, T::BannableAddress, S>,
 
     /// List of connected peers that subscribed to PeerAddresses topic
     subscribed_to_peer_addresses: BTreeSet<PeerId>,
@@ -133,22 +133,20 @@ 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::InvalidConfigurationValue("ping timeout can't be 0".into())
         );
-        Ok(Self {
+        Ok(PeerManager {
+            chain_config,
+            p2p_config,
             peer_connectivity_handle: handle,
             rx_peer_manager,
             tx_sync,
-            peerdb,
-            peers: BTreeMap::new(),
-            pending_connects: HashMap::new(),
+            pending_outbound_connects: HashMap::new(),
             pending_disconnects: HashMap::new(),
-            chain_config,
-            p2p_config,
-            last_heartbeat: now,
+            peers: BTreeMap::new(),
+            peerdb,
             subscribed_to_peer_addresses: BTreeSet::new(),
         })
     }
@@ -161,6 +159,8 @@ where
         version == self.chain_config.version()
     }
 
+    /// Verify that the peer address has a public routable IP and any valid (non-zero) port.
+    /// Private and local IPs are allowed if `allow_discover_private_ips` is true.
     fn is_peer_address_valid(&self, address: &PeerAddress) -> bool {
         <T::Address as TransportAddress>::from_peer_address(
             address,
@@ -174,13 +174,9 @@ where
     /// *receiver_address* is this host socket address as seen and reported by remote peer.
     /// This should work for hosts with public IPs and for hosts behind NAT with port forwarding (same port is assumed).
     /// This won't work for majority of nodes but that should be accepted.
-    fn handle_outbound_receiver_address(
-        &mut self,
-        peer_id: PeerId,
-        receiver_address: PeerAddress,
-    ) -> crate::Result<()> {
+    fn handle_outbound_receiver_address(&mut self, peer_id: PeerId, receiver_address: PeerAddress) {
         if !self.subscribed_to_peer_addresses.contains(&peer_id) {
-            return Ok(());
+            return;
         }
 
         // Take IP and use port numbers from all listening sockets (with same IP version)
@@ -215,167 +211,24 @@ where
             .collect::<Vec<_>>();
 
         for address in discovered_own_addresses {
-            self.announce_address(peer_id, address)?;
+            self.announce_address(peer_id, address);
         }
-
-        Ok(())
     }
 
     /// Send address announcement to the selected peer (if the address is new)
-    fn announce_address(&mut self, peer_id: PeerId, address: T::Address) -> crate::Result<()> {
+    /// `peer_id` must be from the connected peer.
+    fn announce_address(&mut self, peer_id: PeerId, address: T::Address) {
         let peer = self.peers.get_mut(&peer_id).expect("peer must be known");
         if !peer.announced_addresses.contains(&address) {
-            self.peer_connectivity_handle.send_message(
+            Self::send_peer_message(
+                &mut self.peer_connectivity_handle,
                 peer_id,
                 PeerManagerMessage::AnnounceAddrRequest(AnnounceAddrRequest {
                     address: address.as_peer_address(),
                 }),
-            )?;
-            peer.announced_addresses.insert(address);
-        }
-        Ok(())
-    }
-
-    /// Handle connection established event
-    ///
-    /// The event is received from the networking backend and it's either a result of an incoming
-    /// connection from a remote peer or a response to an outbound connection that was initiated
-    /// by the node as result of the peer manager maintenance.
-    fn accept_connection(
-        &mut self,
-        address: T::Address,
-        role: Role,
-        info: PeerInfo,
-        receiver_address: Option<PeerAddress>,
-    ) -> crate::Result<()> {
-        let peer_id = info.peer_id;
-
-        ensure!(
-            info.is_compatible(&self.chain_config),
-            P2pError::ProtocolError(ProtocolError::DifferentNetwork(
-                *self.chain_config.magic_bytes(),
-                info.network,
-            ))
-        );
-        ensure!(
-            self.validate_version(&info.version),
-            P2pError::ProtocolError(ProtocolError::InvalidVersion(
-                *self.chain_config.version(),
-                info.version
-            ))
-        );
-        ensure!(
-            !self.is_peer_connected(info.peer_id),
-            P2pError::PeerError(PeerError::PeerAlreadyExists),
-        );
-        ensure!(
-            !self.peerdb.is_address_connected(&address),
-            P2pError::PeerError(PeerError::PeerAlreadyExists),
-        );
-
-        if role == Role::Outbound {
-            self.peer_connectivity_handle.send_message(
-                peer_id,
-                PeerManagerMessage::AddrListRequest(AddrListRequest {}),
-            )?;
-        }
-
-        log::info!(
-            "peer connected, peer_id: {}, address: {address:?}, {:?}",
-            info.peer_id,
-            role
-        );
-
-        if info.subscriptions.contains(&PubSubTopic::PeerAddresses) {
-            self.subscribed_to_peer_addresses.insert(peer_id);
-        }
-
-        let old_value = self.peers.insert(
-            info.peer_id,
-            PeerContext {
-                info,
-                address: address.clone(),
-                role,
-                score: 0,
-                sent_ping: None,
-                announced_addresses: HashSet::new(),
-            },
-        );
-        assert!(old_value.is_none());
-
-        self.peerdb.peer_connected(address);
-
-        if let (Some(receiver_address), Role::Outbound) = (receiver_address, role) {
-            self.handle_outbound_receiver_address(peer_id, receiver_address)?;
-        }
-
-        self.tx_sync.send(SyncControlEvent::Connected(peer_id)).map_err(P2pError::from)
-    }
-
-    /// Validate inbound peer connection
-    ///
-    /// As inbound connections are received, they are pre-validated before passed
-    /// on to the generic connection validator as these connections haven't gone
-    /// through the same validation as outbound connections.
-    ///
-    /// This function verifies that neither address the nor the peer ID are on the
-    /// list of banned IPs/peer IDs. It also checks that the maximum number of
-    /// connections `PeerManager` is configured to have has not been reached.
-    fn accept_inbound_connection(
-        &mut self,
-        address: T::Address,
-        info: net::types::PeerInfo,
-        receiver_address: Option<PeerAddress>,
-    ) -> crate::Result<()> {
-        log::debug!("validate inbound connection, inbound address {address:?}");
-
-        ensure!(
-            !self.is_peer_connected(info.peer_id),
-            P2pError::PeerError(PeerError::PeerAlreadyExists),
-        );
-
-        let bannable_address = address.as_bannable();
-        ensure!(
-            !self.peerdb.is_address_banned(&bannable_address)?,
-            P2pError::PeerError(PeerError::BannedAddress(address.to_string())),
-        );
-
-        // if the maximum number of connections is reached, the connection cannot be
-        // 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.active_peer_count() >= MAX_ACTIVE_CONNECTIONS {
-            return Err(P2pError::PeerError(PeerError::TooManyPeers));
-        }
-
-        self.accept_connection(address, Role::Inbound, info, receiver_address)
-    }
-
-    /// The connection to a remote peer is reported as closed.
-    ///
-    /// 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: PeerId) -> crate::Result<()> {
-        // The backend is always sending ConnectionClosed event when somebody disconnects, ensure that the peer is active
-        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))?;
-
-            if let Some(Some(response)) = self.pending_disconnects.remove(&peer_id) {
-                response.send(Ok(()));
-            }
-
-            self.subscribed_to_peer_addresses.remove(&peer_id);
-
-            self.peerdb.peer_disconnected(peer.address);
+            peer.announced_addresses.insert(address);
         }
-
-        Ok(())
     }
 
     /// Adjust peer score
@@ -386,10 +239,10 @@ where
     ///
     /// If peer is banned, it is removed from the connected peers
     /// and its address is marked as banned.
-    fn adjust_peer_score(&mut self, peer_id: PeerId, score: u32) -> crate::Result<()> {
+    fn adjust_peer_score(&mut self, peer_id: PeerId, score: u32) {
         let peer = match self.peers.get_mut(&peer_id) {
             Some(peer) => peer,
-            None => return Ok(()),
+            None => return,
         };
 
         peer.score = peer.score.saturating_add(score);
@@ -399,79 +252,61 @@ where
         );
 
         if peer.score >= *self.p2p_config.ban_threshold {
-            self.peerdb.ban_peer(&peer.address)?;
-            self.disconnect(peer_id, None)?;
-        }
-
-        Ok(())
-    }
-
-    /// Handle outbound connection error
-    ///
-    /// The outbound connection was dialed successfully but the remote either did not respond
-    /// (at all or in time) or it didn't support the handshaking which forced the connection closed.
-    ///
-    /// If the connection was initiated by the user via RPC, inform them that the connection failed.
-    /// 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_connects.remove(&address) {
-            channel.send(Err(error));
+            self.peerdb.ban_peer(&peer.address);
+            self.disconnect(peer_id, None);
         }
-
-        self.peerdb.report_outbound_failure(address);
-        Ok(())
     }
 
-    /// Attempt to establish an outbound connection
+    /// Try to initiate a new outbound connection
     ///
     /// 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.
     fn try_connect(&mut self, address: T::Address) -> crate::Result<()> {
         ensure!(
-            !self.pending_connects.contains_key(&address),
+            !self.pending_outbound_connects.contains_key(&address),
             P2pError::PeerError(PeerError::Pending(address.to_string())),
         );
-
         ensure!(
-            !self.peerdb.is_address_connected(&address),
+            !self.is_address_connected(&address),
             P2pError::PeerError(PeerError::PeerAlreadyExists),
         );
 
         let bannable_address = address.as_bannable();
         ensure!(
-            !self.peerdb.is_address_banned(&bannable_address)?,
+            !self.peerdb.is_address_banned(&bannable_address),
             P2pError::PeerError(PeerError::BannedAddress(address.to_string())),
         );
 
-        self.peer_connectivity_handle.connect(address)
+        self.peer_connectivity_handle.connect(address)?;
+
+        Ok(())
     }
 
-    /// Establish an outbound connection
+    /// Initiate a new outbound connection or send error to `response` if it's not possible
     fn connect(
         &mut self,
         address: T::Address,
         response: Option<oneshot_nofail::Sender<crate::Result<()>>>,
-    ) -> crate::Result<()> {
+    ) {
         log::debug!("try to establish outbound connection to peer at address {address:?}");
-
         let res = self.try_connect(address.clone());
 
         match res {
             Ok(()) => {
-                self.pending_connects.insert(address, response);
+                let old_value = self.pending_outbound_connects.insert(address, response);
+                assert!(old_value.is_none());
             }
             Err(e) => {
+                log::debug!("outbound connection to {address:?} failed: {e}");
                 if let Some(response) = response {
                     response.send(Err(e));
                 }
             }
         }
-
-        Ok(())
     }
 
+    // Try to disconnect a connected peer
     fn try_disconnect(&mut self, peer_id: PeerId) -> crate::Result<()> {
         ensure!(
             !self.pending_disconnects.contains_key(&peer_id),
@@ -483,7 +318,9 @@ where
             P2pError::PeerError(PeerError::PeerDisconnected),
         );
 
-        self.peer_connectivity_handle.disconnect(peer_id)
+        self.peer_connectivity_handle.disconnect(peer_id)?;
+
+        Ok(())
     }
 
     /// Disconnect an existing connection (inbound or outbound)
@@ -495,25 +332,277 @@ where
         &mut self,
         peer_id: PeerId,
         response: Option<oneshot_nofail::Sender<crate::Result<()>>>,
-    ) -> crate::Result<()> {
+    ) {
         log::debug!("disconnect peer {peer_id}");
-
         let res = self.try_disconnect(peer_id);
 
         match res {
             Ok(()) => {
-                self.pending_disconnects.insert(peer_id, response);
+                let old_value = self.pending_disconnects.insert(peer_id, response);
+                assert!(old_value.is_none());
             }
             Err(e) => {
+                log::debug!("disconnecting new peer {peer_id} failed: {e}");
                 if let Some(response) = response {
                     response.send(Err(e));
                 }
             }
         }
+    }
+
+    /// Check if the (inbound or outbound) peer connection can be accepted.
+    ///
+    /// For example, an inbound connection will not be accepted when the limit of inbound connections is reached.
+    fn validate_connection(
+        &mut self,
+        address: &T::Address,
+        role: Role,
+        info: &PeerInfo,
+    ) -> crate::Result<()> {
+        // TODO: Allow only one connection per IP address
+        ensure!(
+            info.is_compatible(&self.chain_config),
+            P2pError::ProtocolError(ProtocolError::DifferentNetwork(
+                *self.chain_config.magic_bytes(),
+                info.network,
+            ))
+        );
+        ensure!(
+            self.validate_version(&info.version),
+            P2pError::ProtocolError(ProtocolError::InvalidVersion(
+                *self.chain_config.version(),
+                info.version
+            ))
+        );
+        ensure!(
+            !self.is_peer_connected(info.peer_id),
+            P2pError::PeerError(PeerError::PeerAlreadyExists),
+        );
+        ensure!(
+            !self.is_address_connected(address),
+            P2pError::PeerError(PeerError::PeerAlreadyExists),
+        );
+        ensure!(
+            !self.peerdb.is_address_banned(&address.as_bannable()),
+            P2pError::PeerError(PeerError::BannedAddress(address.to_string())),
+        );
+        // If the maximum number of inbound connections is reached,
+        // the new inbound connection cannot be accepted even if it's valid.
+        // Outbound peer count is not checked because the node initiates new connections
+        // only when needed or from RPC requests.
+        ensure!(
+            self.inbound_peer_count() < *self.p2p_config.max_inbound_connections
+                || role != Role::Inbound,
+            P2pError::PeerError(PeerError::TooManyPeers),
+        );
 
         Ok(())
     }
 
+    /// Try accept new connection
+    ///
+    /// The event is received from the networking backend and it's either a result of an incoming
+    /// connection from a remote peer or a response to an outbound connection that was initiated
+    /// by the node as result of the peer manager maintenance.
+    fn try_accept_connection(
+        &mut self,
+        address: T::Address,
+        role: Role,
+        info: PeerInfo,
+        receiver_address: Option<PeerAddress>,
+    ) -> crate::Result<()> {
+        let peer_id = info.peer_id;
+
+        self.validate_connection(&address, role, &info)?;
+
+        log::info!("new peer accepted, peer_id: {peer_id}, address: {address:?}, role: {role:?}",);
+
+        if info.subscriptions.contains(&PubSubTopic::PeerAddresses) {
+            self.subscribed_to_peer_addresses.insert(info.peer_id);
+        }
+
+        // For security reasons, the address list is only requested from outbound peers.
+        // (inbound peers are generally less trustworthy than outbound peers).
+        let expect_addr_list_response = role == Role::Outbound;
+
+        if expect_addr_list_response {
+            Self::send_peer_message(
+                &mut self.peer_connectivity_handle,
+                peer_id,
+                PeerManagerMessage::AddrListRequest(AddrListRequest {}),
+            );
+        }
+
+        let old_value = self.peers.insert(
+            peer_id,
+            PeerContext {
+                info,
+                address: address.clone(),
+                role,
+                score: 0,
+                sent_ping: None,
+                expect_addr_list_response,
+                announced_addresses: HashSet::new(),
+            },
+        );
+        assert!(old_value.is_none());
+
+        if role == Role::Outbound {
+            self.peerdb.outbound_peer_connected(address);
+
+            if let Some(receiver_address) = receiver_address {
+                self.handle_outbound_receiver_address(peer_id, receiver_address);
+            }
+        }
+
+        Self::send_sync_message(&self.tx_sync, SyncControlEvent::Connected(peer_id));
+
+        Ok(())
+    }
+
+    fn accept_connection(
+        &mut self,
+        address: T::Address,
+        role: Role,
+        info: PeerInfo,
+        receiver_address: Option<PeerAddress>,
+    ) {
+        let peer_id = info.peer_id;
+
+        let accept_res = self.try_accept_connection(address.clone(), role, info, receiver_address);
+
+        if let Err(accept_err) = &accept_res {
+            log::debug!("connection rejected for peer {peer_id}: {accept_err}");
+
+            // Disconnect should always succeed unless the node is shutting down.
+            // Calling expect here is fine because PeerManager will stop before the backend.
+            self.peer_connectivity_handle
+                .disconnect(peer_id)
+                .expect("disconnect failed unexpectedly");
+
+            if role == Role::Outbound {
+                self.peerdb.report_outbound_failure(address.clone(), accept_err);
+            }
+        }
+
+        if role == Role::Outbound {
+            let pending_connect = self
+                .pending_outbound_connects
+                .remove(&address)
+                .expect("pending_outbound_connects must exist (accept_connection)");
+            if let Some(channel) = pending_connect {
+                channel.send(accept_res);
+            }
+        }
+    }
+
+    /// Handle outbound connection error
+    ///
+    /// The outbound connection was dialed successfully but the remote either did not respond
+    /// (at all or in time) or it didn't support the handshaking which forced the connection closed.
+    ///
+    /// If the connection was initiated by the user via RPC, inform them that the connection failed.
+    /// 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) {
+        self.peerdb.report_outbound_failure(address.clone(), &error);
+
+        let pending_connect = self
+            .pending_outbound_connects
+            .remove(&address)
+            .expect("pending_outbound_connects must exist (handle_outbound_error)");
+        if let Some(channel) = pending_connect {
+            channel.send(Err(error));
+        }
+    }
+
+    /// The connection to a remote peer is reported as closed.
+    ///
+    /// 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: PeerId) {
+        // The peer will not be in `peers` for rejected connections
+        if let Some(peer) = self.peers.remove(&peer_id) {
+            log::info!(
+                "peer disconnected, peer_id: {}, address: {:?}",
+                peer.info.peer_id,
+                peer.address
+            );
+
+            Self::send_sync_message(&self.tx_sync, SyncControlEvent::Disconnected(peer_id));
+
+            if let Some(Some(response)) = self.pending_disconnects.remove(&peer_id) {
+                response.send(Ok(()));
+            }
+
+            self.subscribed_to_peer_addresses.remove(&peer_id);
+
+            if peer.role == Role::Outbound {
+                self.peerdb.outbound_peer_disconnected(peer.address);
+            }
+        }
+    }
+
+    fn send_peer_message(
+        peer_connectivity_handle: &mut T::ConnectivityHandle,
+        peer_id: PeerId,
+        message: PeerManagerMessage,
+    ) {
+        // `send_message` should not fail, but even if it does, the error can be ignored
+        // because sending messages over the network does not guarantee that they will be received
+        let res = peer_connectivity_handle.send_message(peer_id, message);
+        if let Err(err) = res {
+            log::error!("send_message failed unexpectedly: {err:?}");
+        }
+    }
+
+    fn send_sync_message(
+        tx_sync: &mpsc::UnboundedSender<SyncControlEvent>,
+        message: SyncControlEvent,
+    ) {
+        let res = tx_sync.send(message);
+        if res.is_err() {
+            log::error!("sending sync messages failed unexpectedly");
+        }
+    }
+
+    /// Fill PeerDb with addresses from the DNS seed servers
+    async fn reload_dns_seed(&mut self) {
+        log::debug!("Resolve DNS seed...");
+        let results = futures::future::join_all(
+            DNS_SEEDS
+                .iter()
+                .map(|host| tokio::net::lookup_host((*host, self.chain_config.p2p_port()))),
+        )
+        .await;
+
+        let mut total = 0;
+        for result in results {
+            match result {
+                Ok(list) => {
+                    list.filter_map(|addr| {
+                        TransportAddress::from_peer_address(
+                            // Convert SocketAddr to PeerAddress
+                            &addr.into(),
+                            *self.p2p_config.allow_discover_private_ips,
+                        )
+                    })
+                    // Randomize selection because records can be sorted by type (A and AAAA)
+                    .choose_multiple(&mut make_pseudo_rng(), MAX_DNS_RECORDS)
+                    .into_iter()
+                    .for_each(|addr| {
+                        total += 1;
+                        self.peerdb.peer_discovered(addr);
+                    });
+                }
+                Err(err) => {
+                    log::error!("resolve DNS seed failed: {err}");
+                }
+            }
+        }
+        log::debug!("DNS seed records found: {total}");
+    }
+
     /// Maintains the peer manager state.
     ///
     /// `PeerManager::heartbeat()` is called every time a network/control event is received
@@ -535,168 +624,173 @@ where
     /// the number of desired connections and there are available peers, the function tries to
     /// establish new connections. After that it updates the peer scores and discards any records
     /// that no longer need to be stored.
-    fn heartbeat(&mut self) -> crate::Result<()> {
-        let count = std::cmp::min(
-            self.peerdb.available_addresses_count(),
-            MAX_ACTIVE_CONNECTIONS
-                .saturating_sub(self.peerdb.available_addresses_count())
-                .saturating_sub(self.pending_connects.len()),
-        );
+    async fn heartbeat(&mut self) {
+        self.peerdb.heartbeat();
+
+        let pending_outbound = self.pending_outbound_connects.keys().cloned().collect();
+        let connected_outbound = self
+            .peers
+            .values()
+            .filter_map(|peer| {
+                if peer.role == Role::Outbound {
+                    Some(peer.address.clone())
+                } else {
+                    None
+                }
+            })
+            .collect();
 
-        let addresses = self.peerdb.random_known_addresses(count);
+        let new_addresses = self
+            .peerdb
+            .select_new_outbound_addresses(&pending_outbound, &connected_outbound);
+
+        // Try to get some records from DNS servers if there are no addresses to connect.
+        // Do this only if no peers are currently connected.
+        let new_addresses = if new_addresses.is_empty()
+            && self.peers.is_empty()
+            && self.pending_outbound_connects.is_empty()
+        {
+            self.reload_dns_seed().await;
+            self.peerdb
+                .select_new_outbound_addresses(&pending_outbound, &connected_outbound)
+        } else {
+            new_addresses
+        };
 
-        for address in addresses {
-            self.connect(address, None)?;
+        for address in new_addresses {
+            self.connect(address, None);
         }
-
-        // TODO: update peer scores
-
-        Ok(())
     }
 
-    fn handle_incoming_message(
-        &mut self,
-        peer: PeerId,
-        message: PeerManagerMessage,
-    ) -> crate::Result<()> {
+    fn handle_incoming_message(&mut self, peer: PeerId, message: PeerManagerMessage) {
         match message {
             PeerManagerMessage::AddrListRequest(_) => self.handle_add_list_request(peer),
             PeerManagerMessage::AnnounceAddrRequest(r) => {
                 self.handle_announce_addr_request(peer, r.address)
             }
             PeerManagerMessage::PingRequest(r) => self.handle_ping_request(peer, r.nonce),
-            PeerManagerMessage::AddrListResponse(r) => self.handle_add_list_response(r.addresses),
-            PeerManagerMessage::AnnounceAddrResponse(_) => Ok(()),
+            PeerManagerMessage::AddrListResponse(r) => {
+                self.handle_addr_list_response(peer, r.addresses)
+            }
             PeerManagerMessage::PingResponse(r) => self.handle_ping_response(peer, r.nonce),
         }
     }
 
-    fn handle_add_list_request(&mut self, peer: PeerId) -> crate::Result<()> {
-        let addresses = self
-            .peerdb
-            .random_known_addresses(MAX_ADDRESS_COUNT)
-            .iter()
-            .map(TransportAddress::as_peer_address)
-            .filter(|address| self.is_peer_address_valid(address))
-            .collect();
-
-        self.peer_connectivity_handle.send_message(
-            peer,
-            PeerManagerMessage::AddrListResponse(AddrListResponse { addresses }),
-        )
-    }
-
-    fn handle_announce_addr_request(
-        &mut self,
-        peer: PeerId,
-        address: PeerAddress,
-    ) -> crate::Result<()> {
+    fn handle_announce_addr_request(&mut self, peer: PeerId, address: PeerAddress) {
         // TODO: Rate limit announce address requests to prevent DoS attacks.
         // For example it's 0.1 req/sec in Bitcoin Core.
-        if let Some(address) = TransportAddress::from_peer_address(
+        if let Some(address) = <T::Address as TransportAddress>::from_peer_address(
             &address,
             *self.p2p_config.allow_discover_private_ips,
         ) {
-            self.peerdb.peer_discovered(&address)?;
-
             self.peers
                 .get_mut(&peer)
                 .expect("peer sending AnnounceAddrRequest must be known")
                 .announced_addresses
                 .insert(address.clone());
 
+            self.peerdb.peer_discovered(address.clone());
+
             let peer_ids = self
                 .subscribed_to_peer_addresses
                 .iter()
                 .cloned()
                 .choose_multiple(&mut make_pseudo_rng(), PEER_ADDRESS_RESEND_COUNT);
             for new_peer_id in peer_ids {
-                self.announce_address(new_peer_id, address.clone())?;
+                self.announce_address(new_peer_id, address.clone());
             }
         }
-        Ok(())
     }
 
-    fn handle_ping_request(&mut self, peer: PeerId, nonce: u64) -> crate::Result<()> {
-        self.peer_connectivity_handle.send_message(
+    fn handle_add_list_request(&mut self, peer: PeerId) {
+        let addresses = self
+            .peerdb
+            .known_addresses()
+            .map(TransportAddress::as_peer_address)
+            .filter(|address| self.is_peer_address_valid(address))
+            .choose_multiple(&mut make_pseudo_rng(), MAX_ADDRESS_COUNT);
+
+        assert!(addresses.len() <= MAX_ADDRESS_COUNT);
+
+        Self::send_peer_message(
+            &mut self.peer_connectivity_handle,
             peer,
-            PeerManagerMessage::PingResponse(PingResponse { nonce }),
-        )
+            PeerManagerMessage::AddrListResponse(AddrListResponse { addresses }),
+        );
     }
 
-    fn handle_add_list_response(&mut self, addresses: Vec<PeerAddress>) -> crate::Result<()> {
+    fn try_handle_addr_list_response(
+        &mut self,
+        peer_id: PeerId,
+        addresses: Vec<PeerAddress>,
+    ) -> crate::Result<()> {
+        let peer = self
+            .peers
+            .get_mut(&peer_id)
+            .ok_or(P2pError::PeerError(PeerError::PeerDoesntExist))?;
+        ensure!(
+            addresses.len() <= MAX_ADDRESS_COUNT,
+            P2pError::ProtocolError(ProtocolError::InvalidMessage)
+        );
+        ensure!(
+            peer.expect_addr_list_response,
+            P2pError::ProtocolError(ProtocolError::UnexpectedMessage("AddrListResponse"))
+        );
+
+        peer.expect_addr_list_response = false;
+
         for address in addresses {
             if let Some(address) = TransportAddress::from_peer_address(
                 &address,
                 *self.p2p_config.allow_discover_private_ips,
             ) {
-                self.peerdb.peer_discovered(&address)?;
+                self.peerdb.peer_discovered(address);
             }
         }
+
         Ok(())
     }
 
-    fn handle_ping_response(&mut self, peer: PeerId, nonce: u64) -> crate::Result<()> {
-        if let Some(peer) = self.peers.get_mut(&peer) {
-            if peer.sent_ping.as_ref().map(|sent_ping| sent_ping.nonce) == Some(nonce) {
-                // Correct reply received, clear pending request.
-                peer.sent_ping = None;
-            }
+    fn handle_addr_list_response(&mut self, peer_id: PeerId, addresses: Vec<PeerAddress>) {
+        let res = self.try_handle_addr_list_response(peer_id, addresses);
+        if let Err(err) = res {
+            log::debug!("try_handle_addr_list_response failed: {err}");
+            self.adjust_peer_score(peer_id, err.ban_score());
         }
-        Ok(())
     }
 
-    /// Handle the result of a control/network event
-    ///
-    /// Currently only subsystem/channel-related errors are considered fatal.
-    /// Other errors are logged as warnings and `Ok(())` is returned as they should
-    /// not disturb the operation of `PeerManager`.
-    ///
-    /// If an error has ban score greater than zero, the peer score is updated and connection
-    /// to that peer is possibly closed if their score crossed the ban threshold.
-    ///
-    /// # Arguments
-    /// `peer_id` - peer ID of the remote peer, if available
-    /// `result` - result of the operation that was performed
-    pub fn handle_result(
-        &mut self,
-        peer_id: Option<PeerId>,
-        result: crate::Result<()>,
-    ) -> crate::Result<()> {
-        match result {
-            Ok(_) => Ok(()),
-            Err(P2pError::ChannelClosed | P2pError::SubsystemFailure) => {
-                log::error!("connection lost with subsystem wrapper/p2p subsystem");
-                result
-            }
-            Err(err) => {
-                log::warn!("non-fatal error occurred: {err}");
-
-                if let Some(peer_id) = peer_id {
-                    log::info!("adjust peer score for peer {peer_id}, reason {err}");
-                    return self.adjust_peer_score(peer_id, err.ban_score());
-                }
+    fn handle_ping_request(&mut self, peer: PeerId, nonce: u64) {
+        Self::send_peer_message(
+            &mut self.peer_connectivity_handle,
+            peer,
+            PeerManagerMessage::PingResponse(PingResponse { nonce }),
+        );
+    }
 
-                Ok(())
+    fn handle_ping_response(&mut self, peer: PeerId, nonce: u64) {
+        if let Some(peer) = self.peers.get_mut(&peer) {
+            if peer.sent_ping.as_ref().map(|sent_ping| sent_ping.nonce) == Some(nonce) {
+                // Correct reply received, clear pending request
+                peer.sent_ping = None;
             }
         }
     }
 
     /// Handle control event.
     ///
-    /// Handle events from an outside controller (rpc, for example) that sets/gets values for PeerManager
-    fn handle_control_event(&mut self, event: PeerManagerEvent<T>) -> crate::Result<()> {
+    /// Handle events from an outside controller (rpc, for example) that sets/gets values for PeerManager.
+    fn handle_control_event(&mut self, event: PeerManagerEvent<T>) {
         match event {
             PeerManagerEvent::Connect(address, response) => {
-                self.connect(address, Some(response))?;
+                self.connect(address, Some(response));
             }
             PeerManagerEvent::Disconnect(peer_id, response) => {
-                self.disconnect(peer_id, Some(response))?;
+                self.disconnect(peer_id, Some(response));
             }
             PeerManagerEvent::AdjustPeerScore(peer_id, score, response) => {
                 log::debug!("adjust peer {peer_id} score: {score}");
-
-                response.send(self.adjust_peer_score(peer_id, score));
+                self.adjust_peer_score(peer_id, score);
+                response.send(Ok(()));
             }
             PeerManagerEvent::GetPeerCount(response) => {
                 response.send(self.active_peer_count());
@@ -715,104 +809,65 @@ where
                 response.send(peers);
             }
         }
-
-        Ok(())
     }
 
-    /// Handle connectivity event.
-    fn handle_connectivity_event_result(
-        &mut self,
-        event_res: crate::Result<ConnectivityEvent<T::Address>>,
-    ) -> crate::Result<()> {
-        match event_res {
-            Ok(event) => match event {
-                ConnectivityEvent::Message { peer, message } => {
-                    self.handle_incoming_message(peer, message)?
-                }
-                net::types::ConnectivityEvent::InboundAccepted {
-                    address,
-                    peer_info,
-                    receiver_address,
-                } => {
-                    let peer_id = peer_info.peer_id;
-
-                    match self.accept_inbound_connection(address, peer_info, receiver_address) {
-                        Ok(_) => {}
-                        Err(P2pError::ChannelClosed) => return Err(P2pError::ChannelClosed),
-                        Err(P2pError::PeerError(err)) => {
-                            log::warn!("peer error for peer {peer_id}: {err}");
-                            self.peer_connectivity_handle.disconnect(peer_id)?;
-                        }
-                        Err(P2pError::ProtocolError(err)) => {
-                            log::warn!("peer error for peer {peer_id}: {err}");
-                            self.adjust_peer_score(peer_id, err.ban_score())?;
-                        }
-                        Err(err) => {
-                            log::error!("unknown error for peer {peer_id}: {err}");
-                        }
-                    }
-                }
-                net::types::ConnectivityEvent::OutboundAccepted {
-                    address,
-                    peer_info,
-                    receiver_address,
-                } => {
-                    let peer_id = peer_info.peer_id;
-                    let res = self.accept_connection(
-                        address.clone(),
-                        Role::Outbound,
-                        peer_info,
-                        receiver_address,
-                    );
-                    self.handle_result(Some(peer_id), res)?;
-
-                    match self.pending_connects.remove(&address) {
-                        Some(Some(channel)) => {
-                            channel.send(Ok(()));
-                        }
-                        Some(None) => {}
-                        None => log::error!("connection accepted but it's not pending?"),
-                    }
-                }
-                net::types::ConnectivityEvent::ConnectionClosed { peer_id } => {
-                    let res = self.connection_closed(peer_id);
-                    self.handle_result(Some(peer_id), res)?;
-                }
-                net::types::ConnectivityEvent::ConnectionError { address, error } => {
-                    let res = self.handle_outbound_error(address, error);
-                    self.handle_result(None, res)?;
-                }
-                net::types::ConnectivityEvent::Misbehaved { peer_id, error } => {
-                    let res = self.adjust_peer_score(peer_id, error.ban_score());
-                    self.handle_result(Some(peer_id), res)?;
-                }
-            },
-            Err(err) => {
-                log::error!("failed to read network event: {err:?}");
-                self.handle_result(None, Err(err))?;
+    /// Handle connectivity event
+    fn handle_connectivity_event(&mut self, event: ConnectivityEvent<T::Address>) {
+        match event {
+            ConnectivityEvent::Message { peer, message } => {
+                self.handle_incoming_message(peer, message);
+            }
+            ConnectivityEvent::InboundAccepted {
+                address,
+                peer_info,
+                receiver_address,
+            } => {
+                self.accept_connection(address, Role::Inbound, peer_info, receiver_address);
+            }
+            ConnectivityEvent::OutboundAccepted {
+                address,
+                peer_info,
+                receiver_address,
+            } => {
+                self.accept_connection(address, Role::Outbound, peer_info, receiver_address);
+            }
+            ConnectivityEvent::ConnectionClosed { peer_id } => {
+                self.connection_closed(peer_id);
+            }
+            ConnectivityEvent::ConnectionError { address, error } => {
+                self.handle_outbound_error(address, error);
+            }
+            ConnectivityEvent::Misbehaved { peer_id, error } => {
+                self.adjust_peer_score(peer_id, error.ban_score());
             }
         }
-
-        Ok(())
     }
 
     /// Get the number of active peers
-    pub fn active_peer_count(&self) -> usize {
+    fn active_peer_count(&self) -> usize {
         self.peers.len()
     }
 
     /// Returns short info about all connected peers
-    pub fn get_connected_peers(&self) -> Vec<ConnectedPeer> {
+    fn get_connected_peers(&self) -> Vec<ConnectedPeer> {
         self.peers.values().map(Into::into).collect()
     }
 
     /// Checks if the peer is in active state
-    pub fn is_peer_connected(&self, peer_id: PeerId) -> bool {
+    fn is_peer_connected(&self, peer_id: PeerId) -> bool {
         self.peers.get(&peer_id).is_some()
     }
 
+    fn is_address_connected(&self, address: &T::Address) -> bool {
+        self.peers.values().any(|peer| peer.address == *address)
+    }
+
+    fn inbound_peer_count(&self) -> usize {
+        self.peers.values().filter(|peer| peer.role == Role::Inbound).count()
+    }
+
     /// Sends ping requests and disconnects peers that do not respond in time
-    fn ping_check(&mut self) -> crate::Result<()> {
+    fn ping_check(&mut self) {
         let now = Instant::now();
         let mut dead_peers = Vec::new();
         for (peer_id, peer) in self.peers.iter_mut() {
@@ -828,10 +883,11 @@ where
                 }
                 None => {
                     let nonce = make_pseudo_rng().gen();
-                    self.peer_connectivity_handle.send_message(
+                    Self::send_peer_message(
+                        &mut self.peer_connectivity_handle,
                         *peer_id,
                         PeerManagerMessage::PingRequest(PingRequest { nonce }),
-                    )?;
+                    );
                     peer.sent_ping = Some(SentPing {
                         nonce,
                         timestamp: now,
@@ -841,10 +897,8 @@ where
         }
 
         for peer_id in dead_peers {
-            self.disconnect(peer_id, None)?;
+            self.disconnect(peer_id, None);
         }
-
-        Ok(())
     }
 
     /// Runs the `PeerManager` event loop.
@@ -871,25 +925,30 @@ where
             tokio::time::interval(Duration::MAX)
         };
 
+        // Run heartbeat right away to start outbound connections
+        self.heartbeat().await;
+        // Last time when heartbeat was called
+        let mut last_heartbeat = Instant::now();
+
         loop {
             tokio::select! {
-                event = self.rx_peer_manager.recv() => {
-                    self.handle_control_event(event.ok_or(P2pError::ChannelClosed)?)?;
+                event_res = self.rx_peer_manager.recv() => {
+                    self.handle_control_event(event_res.ok_or(P2pError::ChannelClosed)?);
                 },
                 event_res = self.peer_connectivity_handle.poll_next() => {
-                    self.handle_connectivity_event_result(event_res)?;
+                    self.handle_connectivity_event(event_res?);
                 },
                 _event = ping_check_interval.tick(), if ping_check_enabled => {
-                    self.ping_check()?;
+                    self.ping_check();
                 }
                 _event = tokio::time::sleep(PEER_MGR_HEARTBEAT_INTERVAL_MAX) => {}
             }
 
-            // finally update peer manager state
-            let now = tokio::time::Instant::now();
-            if now.duration_since(self.last_heartbeat) > PEER_MGR_HEARTBEAT_INTERVAL_MIN {
-                self.heartbeat()?;
-                self.last_heartbeat = now;
+            // Finally, update the peer manager state
+            let now = Instant::now();
+            if now.duration_since(last_heartbeat) >= PEER_MGR_HEARTBEAT_INTERVAL_MIN {
+                self.heartbeat().await;
+                last_heartbeat = now;
             }
         }
     }
diff --git a/p2p/src/peer_manager/peer_context.rs b/p2p/src/peer_manager/peer_context.rs
index fce8fb862..59e8838cc 100644
--- a/p2p/src/peer_manager/peer_context.rs
+++ b/p2p/src/peer_manager/peer_context.rs
@@ -45,6 +45,11 @@ pub struct PeerContext<A> {
     /// Sent ping details
     pub sent_ping: Option<SentPing>,
 
+    /// Whether an address list request was sent and no response was received.
+    ///
+    /// It is used to score peers that send unsolicited address list responses.
+    pub expect_addr_list_response: bool,
+
     /// All addresses that were announced to or from some peer.
     /// Used to prevent infinity loops while broadcasting addresses.
     // TODO: Use bloom filter (like it's done in Bitcoin Core).
diff --git a/p2p/src/peer_manager/peerdb/address_data/mod.rs b/p2p/src/peer_manager/peerdb/address_data/mod.rs
new file mode 100644
index 000000000..6b0b4d761
--- /dev/null
+++ b/p2p/src/peer_manager/peerdb/address_data/mod.rs
@@ -0,0 +1,200 @@
+// 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::time::Duration;
+
+use tokio::time::Instant;
+use utils::const_value::ConstValue;
+
+/// When the node drops the unreachable node address. Used for negative caching.
+pub const PURGE_UNREACHABLE_TIME: Duration = Duration::from_secs(3600);
+
+/// When the server drops the unreachable node address that was once reachable. This should take about a month.
+/// Such a long time is useful if the node itself has prolonged connectivity problems.
+pub const PURGE_REACHABLE_FAIL_COUNT: u32 = 35;
+
+pub enum AddressState {
+    Connected {},
+
+    Disconnected {
+        /// Whether the address was reachable at least once.
+        /// Addresses that were once reachable are stored in the DB.
+        was_reachable: bool,
+
+        /// The number of consecutive failed connection attempts.
+        /// New connection attempts are made after a progressive backoff time.
+        fail_count: u32,
+
+        /// Last time the peer disconnected
+        disconnected_at: Instant,
+    },
+
+    Unreachable {
+        /// At which time the address would be removed from memory
+        erase_after: Instant,
+    },
+}
+
+pub enum AddressStateTransitionTo {
+    Connected,
+    Disconnected,
+    ConnectionFailed,
+}
+
+pub struct AddressData {
+    state: AddressState,
+
+    user_added: ConstValue<bool>,
+}
+
+impl AddressData {
+    pub fn new(was_reachable: bool, user_added: bool, now: Instant) -> Self {
+        AddressData {
+            state: AddressState::Disconnected {
+                was_reachable,
+                fail_count: 0,
+                disconnected_at: now,
+            },
+            user_added: user_added.into(),
+        }
+    }
+
+    /// Returns true when it is time to attempt a new outbound connection
+    pub fn connect_now(&self, now: Instant) -> bool {
+        match self.state {
+            AddressState::Connected {} => false,
+
+            AddressState::Disconnected {
+                fail_count,
+                disconnected_at,
+                was_reachable,
+            } => {
+                let age = now.duration_since(disconnected_at);
+                if *self.user_added {
+                    // Try to connect to the user added nodes more often
+                    let age = now - disconnected_at;
+                    match fail_count {
+                        0 => true,
+                        1 => age > Duration::from_secs(10),
+                        2 => age > Duration::from_secs(60),
+                        3 => age > Duration::from_secs(180),
+                        _ => age > Duration::from_secs(360),
+                    }
+                } else if was_reachable {
+                    match fail_count {
+                        0 => true,
+                        1 => age > Duration::from_secs(60),
+                        2 => age > Duration::from_secs(360),
+                        3 => age > Duration::from_secs(3600),
+                        4 => age > Duration::from_secs(3 * 3600),
+                        5 => age > Duration::from_secs(6 * 3600),
+                        6 => age > Duration::from_secs(12 * 3600),
+                        _ => age > Duration::from_secs(24 * 3600),
+                    }
+                } else {
+                    fail_count == 0
+                }
+            }
+
+            AddressState::Unreachable { erase_after: _ } => false,
+        }
+    }
+
+    /// Returns true if the address should be kept in memory
+    pub fn retain(&self, now: Instant) -> bool {
+        match self.state {
+            AddressState::Connected {} => true,
+            AddressState::Disconnected {
+                was_reachable: _,
+                fail_count: _,
+                disconnected_at: _,
+            } => true,
+            AddressState::Unreachable { erase_after } => erase_after < now,
+        }
+    }
+
+    /// Returns true if the address should be stored in the DB
+    pub fn is_persistent(&self) -> bool {
+        match self.state {
+            AddressState::Connected {} => true,
+            AddressState::Disconnected {
+                fail_count: _,
+                disconnected_at: _,
+                was_reachable,
+            } => was_reachable,
+            AddressState::Unreachable { erase_after: _ } => false,
+        }
+    }
+
+    pub fn transition_to(&mut self, transition: AddressStateTransitionTo, now: Instant) {
+        self.state = match transition {
+            AddressStateTransitionTo::Connected => match self.state {
+                AddressState::Connected {} => unreachable!(),
+                AddressState::Disconnected {
+                    fail_count: _,
+                    disconnected_at: _,
+                    was_reachable: _,
+                } => AddressState::Connected {},
+                AddressState::Unreachable { erase_after: _ } => unreachable!(),
+            },
+
+            AddressStateTransitionTo::Disconnected => match self.state {
+                AddressState::Connected {} => AddressState::Disconnected {
+                    fail_count: 0,
+                    disconnected_at: now,
+                    was_reachable: true,
+                },
+                AddressState::Disconnected {
+                    fail_count: _,
+                    disconnected_at: _,
+                    was_reachable: _,
+                } => unreachable!(),
+                AddressState::Unreachable { erase_after: _ } => unreachable!(),
+            },
+
+            AddressStateTransitionTo::ConnectionFailed => match self.state {
+                AddressState::Connected {} => unreachable!(),
+                AddressState::Disconnected {
+                    fail_count,
+                    disconnected_at: _,
+                    was_reachable,
+                } => {
+                    if *self.user_added {
+                        AddressState::Disconnected {
+                            fail_count: fail_count + 1,
+                            disconnected_at: now,
+                            was_reachable,
+                        }
+                    } else if !was_reachable {
+                        AddressState::Unreachable {
+                            erase_after: now + PURGE_UNREACHABLE_TIME,
+                        }
+                    } else if fail_count + 1 >= PURGE_REACHABLE_FAIL_COUNT {
+                        AddressState::Unreachable { erase_after: now }
+                    } else {
+                        AddressState::Disconnected {
+                            fail_count: fail_count + 1,
+                            disconnected_at: now,
+                            was_reachable,
+                        }
+                    }
+                }
+                AddressState::Unreachable { erase_after: _ } => {
+                    unreachable!()
+                }
+            },
+        }
+    }
+}
diff --git a/p2p/src/peer_manager/peerdb/mod.rs b/p2p/src/peer_manager/peerdb/mod.rs
index a433f98ba..f3131078a 100644
--- a/p2p/src/peer_manager/peerdb/mod.rs
+++ b/p2p/src/peer_manager/peerdb/mod.rs
@@ -16,8 +16,7 @@
 //! Peer database
 //!
 //! The peer database stores:
-//! - connected peers
-//! - available (discovered) addresses
+//! - all outbound peer addresses
 //! - banned addresses
 //!
 //! Connected peers are those peers that the [`crate::peer_manager::PeerManager`] has an active
@@ -25,188 +24,227 @@
 //! used by [`crate::peer_manager::PeerManager::heartbeat()`] to establish new outbound connections
 //! if the actual number of active connection is less than the desired number of connections.
 
+pub mod address_data;
 pub mod storage;
 pub mod storage_impl;
+mod storage_load;
 
 use std::{
-    collections::{BTreeMap, BTreeSet},
+    collections::{btree_map::Entry, BTreeMap, BTreeSet},
+    str::FromStr,
     sync::Arc,
     time::Duration,
 };
 
 use common::time_getter::TimeGetter;
-use crypto::random::{make_pseudo_rng, SliceRandom};
+use crypto::random::{make_pseudo_rng, seq::IteratorRandom};
+use logging::log;
+use tokio::time::Instant;
 
-use crate::{
-    config,
-    error::{ConversionError, P2pError},
-    net::{AsBannableAddress, NetworkingService},
-};
+use crate::{config, error::P2pError, net::AsBannableAddress};
 
-use self::storage::{
-    PeerDbStorage, PeerDbStorageRead, PeerDbStorageWrite, PeerDbTransactionRo, PeerDbTransactionRw,
+use self::{
+    address_data::{AddressData, AddressStateTransitionTo},
+    storage::{PeerDbStorage, PeerDbStorageWrite},
+    storage_load::LoadedStorage,
 };
 
-pub struct PeerDb<T: NetworkingService, S> {
+use super::MAX_OUTBOUND_CONNECTIONS;
+
+pub struct PeerDb<A, B, S> {
     /// P2P configuration
     p2p_config: Arc<config::P2pConfig>,
 
-    /// Set of currently connected addresses
-    connected_addresses: BTreeSet<T::Address>,
-
-    /// Set of all known addresses
-    known_addresses: BTreeSet<T::Address>,
+    /// Map of all outbound peer addresses
+    addresses: BTreeMap<A, AddressData>,
 
     /// Banned addresses along with the duration of the ban.
     ///
     /// The duration represents the `UNIX_EPOCH + duration` time point, so the ban should end
     /// when `current_time > ban_duration`.
-    banned_addresses: BTreeMap<T::BannableAddress, Duration>,
+    banned_addresses: BTreeMap<B, Duration>,
 
     time_getter: TimeGetter,
 
     storage: S,
 }
 
-impl<T: NetworkingService, S: PeerDbStorage> PeerDb<T, S> {
+impl<A, B, S> PeerDb<A, B, S>
+where
+    A: Ord + FromStr + ToString + Clone + AsBannableAddress<BannableAddress = B>,
+    B: Ord + FromStr + ToString,
+    S: PeerDbStorage,
+{
     pub fn new(
         p2p_config: Arc<config::P2pConfig>,
         time_getter: TimeGetter,
         storage: S,
     ) -> crate::Result<Self> {
+        // Node won't start if DB loading fails!
+        let loaded_storage = LoadedStorage::<A, B>::load_storage(&storage)?;
+
         let added_nodes = p2p_config
             .added_nodes
             .iter()
             .map(|addr| {
-                addr.parse::<T::Address>().map_err(|_err| {
-                    P2pError::ConversionError(ConversionError::InvalidAddress(addr.clone()))
+                addr.parse::<A>().map_err(|_err| {
+                    P2pError::InvalidConfigurationValue(format!("Invalid address: {addr}"))
                 })
             })
-            .collect::<Result<Vec<_>, _>>()?;
+            .collect::<Result<BTreeSet<_>, _>>()?;
 
-        // Node won't start if DB loading fails!
-        let tx = storage.transaction_ro()?;
-        let stored_known_addresses = tx.get_known_addresses()?;
-        let stored_banned_addresses = tx.get_banned_addresses()?;
-        tx.close();
-
-        let stored_known_addresses_iter =
-            stored_known_addresses.iter().filter_map(|address| address.parse().ok());
-        // TODO: We need to handle added nodes differently from ordinary nodes.
-        // There are peers that we want to persistently have, and others that we want to just give a "shot" at connecting at.
-        let known_addresses = stored_known_addresses_iter.chain(added_nodes.into_iter()).collect();
-
-        let banned_addresses = stored_banned_addresses
-            .iter()
-            .filter_map(|(address, duration)| {
-                address.parse().ok().map(|address| (address, *duration))
+        let now = Instant::now();
+        let addresses = loaded_storage
+            .known_addresses
+            .union(&added_nodes)
+            .map(|addr| {
+                (
+                    addr.clone(),
+                    AddressData::new(
+                        loaded_storage.known_addresses.contains(addr),
+                        added_nodes.contains(addr),
+                        now,
+                    ),
+                )
             })
             .collect();
 
         Ok(Self {
-            connected_addresses: Default::default(),
-            known_addresses,
-            banned_addresses,
+            addresses,
+            banned_addresses: loaded_storage.banned_addresses,
             p2p_config,
             time_getter,
             storage,
         })
     }
 
-    /// Get the number of idle (available) addresses
-    pub fn available_addresses_count(&self) -> usize {
-        self.known_addresses.len()
-    }
-
-    /// Checks if the given address is already connected.
-    pub fn is_address_connected(&self, address: &T::Address) -> bool {
-        self.connected_addresses.contains(address)
-    }
-
-    /// Selects requested count of peer addresses from the DB randomly.
+    /// Iterator of all known addresses.
     ///
     /// Result could be shared with remote peers over network.
-    pub fn random_known_addresses(&self, count: usize) -> Vec<T::Address> {
-        // TODO: Use something more efficient (without iterating over the all addresses first)
-        let all_addresses = self.known_addresses.iter().cloned().collect::<Vec<_>>();
-        all_addresses
-            .choose_multiple(&mut make_pseudo_rng(), count)
-            .cloned()
-            .collect::<Vec<_>>()
+    pub fn known_addresses(&self) -> impl Iterator<Item = &A> {
+        self.addresses.keys()
     }
 
-    /// 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) {
-            // Check if the ban has expired.
-            let now = self.time_getter.get_time();
-            if now > *banned_till {
-                self.banned_addresses.remove(address);
-                let mut tx = self.storage.transaction_rw()?;
-                tx.del_banned_address(&address.to_string())?;
-                tx.commit()?;
-            } else {
-                return Ok(true);
-            }
-        }
-
-        Ok(false)
+    /// Selects peer addresses for outbound connections
+    pub fn select_new_outbound_addresses(
+        &self,
+        pending_outbound: &BTreeSet<A>,
+        connected_outbound: &BTreeSet<A>,
+    ) -> Vec<A> {
+        let count = MAX_OUTBOUND_CONNECTIONS
+            .saturating_sub(pending_outbound.len())
+            .saturating_sub(connected_outbound.len());
+
+        // TODO: Ignore banned addresses
+        // TODO: Allow only one connection per IP address
+        // TODO: Always try to connect to user-added addresses without considering `MAX_OUTBOUND_CONNECTIONS`
+        self.addresses
+            .iter()
+            .filter(|(addr, address_data)| {
+                address_data.connect_now(Instant::now())
+                    && !pending_outbound.contains(addr)
+                    && !connected_outbound.contains(addr)
+            })
+            .map(|(addr, _address_data)| addr.clone())
+            .choose_multiple(&mut make_pseudo_rng(), count)
     }
 
-    /// 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> {
-        self.random_known_addresses(1).into_iter().next()
+    /// Perform the PeerDb maintenance
+    pub fn heartbeat(&mut self) {
+        let now = Instant::now();
+        self.addresses.retain(|_addr, address_data| address_data.retain(now));
     }
 
     /// Add new peer addresses
-    pub fn peer_discovered(&mut self, address: &T::Address) -> crate::Result<()> {
-        self.known_addresses.insert(address.clone());
-
-        let mut tx = self.storage.transaction_rw()?;
-        tx.add_known_address(&address.to_string())?;
-        tx.commit()?;
-        Ok(())
+    pub fn peer_discovered(&mut self, address: A) {
+        if let Entry::Vacant(entry) = self.addresses.entry(address.clone()) {
+            log::debug!("new address discovered: {}", address.to_string());
+            entry.insert(AddressData::new(false, false, Instant::now()));
+        }
     }
 
     /// Report outbound connection failure
     ///
     /// When [`crate::peer_manager::PeerManager::heartbeat()`] has initiated an outbound connection
     /// and the connection is refused, it's reported back to the `PeerDb` so it marks the address as unreachable.
-    pub fn report_outbound_failure(&mut self, _address: T::Address) {
-        // TODO: implement
+    pub fn report_outbound_failure(&mut self, address: A, _error: &P2pError) {
+        self.change_address_state(address, AddressStateTransitionTo::ConnectionFailed);
     }
 
     /// Mark peer as connected
     ///
     /// 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) {
-        let is_inserted = self.connected_addresses.insert(address);
-        assert!(is_inserted);
+    pub fn outbound_peer_connected(&mut self, address: A) {
+        self.change_address_state(address, AddressStateTransitionTo::Connected);
     }
 
     /// Handle peer disconnection event
     ///
     /// Close the connection to an active peer.
-    pub fn peer_disconnected(&mut self, address: T::Address) {
-        let is_removed = self.connected_addresses.remove(&address);
-        assert!(is_removed);
+    pub fn outbound_peer_disconnected(&mut self, address: A) {
+        self.change_address_state(address, AddressStateTransitionTo::Disconnected);
+    }
+
+    pub fn change_address_state(&mut self, address: A, transition: AddressStateTransitionTo) {
+        if let Some(address_data) = self.addresses.get_mut(&address) {
+            let is_persistent_old = address_data.is_persistent();
+
+            address_data.transition_to(transition, Instant::now());
+
+            let is_persistent_new = address_data.is_persistent();
+
+            match (is_persistent_old, is_persistent_new) {
+                (false, true) => {
+                    storage::update_db(&self.storage, |tx| {
+                        tx.add_known_address(&address.to_string())
+                    })
+                    .expect("adding address expected to succeed (peer_connected)");
+                }
+                (true, false) => {
+                    storage::update_db(&self.storage, |tx| {
+                        tx.del_known_address(&address.to_string())
+                    })
+                    .expect("adding address expected to succeed (peer_connected)");
+                }
+                _ => {}
+            }
+        }
     }
 
-    /// Changes the peer state to `Peer::Banned` and bans it for 24 hours.
-    pub fn ban_peer(&mut self, address: &T::Address) -> crate::Result<()> {
+    /// Checks if the given address is banned
+    pub fn is_address_banned(&mut self, address: &B) -> bool {
+        if let Some(banned_till) = self.banned_addresses.get(address) {
+            // Check if the ban has expired
+            let now = self.time_getter.get_time();
+            if now <= *banned_till {
+                return true;
+            }
+
+            self.banned_addresses.remove(address);
+
+            storage::update_db(&self.storage, |tx| {
+                tx.del_banned_address(&address.to_string())
+            })
+            .expect("removing banned address is expected to succeed (is_address_banned)");
+        }
+
+        false
+    }
+
+    /// Changes the address state to banned
+    pub fn ban_peer(&mut self, address: &A) {
         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(())
-    }
 
-    #[cfg(feature = "testing_utils")]
-    pub fn get_storage_mut(&mut self) -> &mut S {
-        &mut self.storage
+        storage::update_db(&self.storage, |tx| {
+            tx.add_banned_address(&bannable_address.to_string(), ban_till)
+        })
+        .expect("adding banned address is expected to succeed (ban_peer)");
+
+        self.banned_addresses.insert(bannable_address, ban_till);
     }
 }
+
+#[cfg(test)]
+mod tests;
diff --git a/p2p/src/peer_manager/peerdb/storage.rs b/p2p/src/peer_manager/peerdb/storage.rs
index 49c7721a7..5c531e9f2 100644
--- a/p2p/src/peer_manager/peerdb/storage.rs
+++ b/p2p/src/peer_manager/peerdb/storage.rs
@@ -16,19 +16,27 @@
 use std::time::Duration;
 
 pub trait PeerDbStorageRead {
-    fn get_known_addresses(&self) -> crate::Result<Vec<String>>;
+    fn get_version(&self) -> Result<Option<u32>, storage::Error>;
 
-    fn get_banned_addresses(&self) -> crate::Result<Vec<(String, Duration)>>;
+    fn get_known_addresses(&self) -> Result<Vec<String>, storage::Error>;
+
+    fn get_banned_addresses(&self) -> Result<Vec<(String, Duration)>, storage::Error>;
 }
 
 pub trait PeerDbStorageWrite {
-    fn add_known_address(&mut self, address: &str) -> crate::Result<()>;
+    fn set_version(&mut self, version: u32) -> Result<(), storage::Error>;
+
+    fn add_known_address(&mut self, address: &str) -> Result<(), storage::Error>;
 
-    fn del_known_address(&mut self, address: &str) -> crate::Result<()>;
+    fn del_known_address(&mut self, address: &str) -> Result<(), storage::Error>;
 
-    fn add_banned_address(&mut self, address: &str, duration: Duration) -> crate::Result<()>;
+    fn add_banned_address(
+        &mut self,
+        address: &str,
+        duration: Duration,
+    ) -> Result<(), storage::Error>;
 
-    fn del_banned_address(&mut self, address: &str) -> crate::Result<()>;
+    fn del_banned_address(&mut self, address: &str) -> Result<(), storage::Error>;
 }
 
 pub trait PeerDbTransactionRo: PeerDbStorageRead {
@@ -38,7 +46,7 @@ pub trait PeerDbTransactionRo: PeerDbStorageRead {
 pub trait PeerDbTransactionRw: PeerDbStorageWrite {
     fn abort(self);
 
-    fn commit(self) -> crate::Result<()>;
+    fn commit(self) -> Result<(), storage::Error>;
 }
 
 /// Support for transactions over blockchain storage
@@ -50,10 +58,39 @@ pub trait PeerDbTransactional<'t> {
     type TransactionRw: PeerDbTransactionRw + 't;
 
     /// Start a read-only transaction.
-    fn transaction_ro<'s: 't>(&'s self) -> crate::Result<Self::TransactionRo>;
+    fn transaction_ro<'s: 't>(&'s self) -> Result<Self::TransactionRo, storage::Error>;
 
     /// Start a read-write transaction.
-    fn transaction_rw<'s: 't>(&'s self) -> crate::Result<Self::TransactionRw>;
+    fn transaction_rw<'s: 't>(&'s self) -> Result<Self::TransactionRw, storage::Error>;
 }
 
 pub trait PeerDbStorage: for<'tx> PeerDbTransactional<'tx> + Send {}
+
+const MAX_RECOVERABLE_ERROR_RETRY_COUNT: u32 = 3;
+
+/// Try update storage, gracefully handle recoverable errors
+pub fn update_db<S, F>(storage: &S, f: F) -> Result<(), storage::Error>
+where
+    S: PeerDbStorage,
+    F: Fn(&mut <S as PeerDbTransactional<'_>>::TransactionRw) -> Result<(), storage::Error>,
+{
+    let mut recoverable_errors = 0;
+    loop {
+        let res = || -> Result<(), storage::Error> {
+            let mut tx = storage.transaction_rw()?;
+            f(&mut tx)?;
+            tx.commit()
+        }();
+
+        match res {
+            Ok(()) => return Ok(()),
+            err @ Err(storage::Error::Recoverable(_)) => {
+                recoverable_errors += 1;
+                if recoverable_errors >= MAX_RECOVERABLE_ERROR_RETRY_COUNT {
+                    return err;
+                }
+            }
+            err @ Err(storage::Error::Fatal(_)) => return err,
+        }
+    }
+}
diff --git a/p2p/src/peer_manager/peerdb/storage_impl.rs b/p2p/src/peer_manager/peerdb/storage_impl.rs
index 8c9ef5106..635d33452 100644
--- a/p2p/src/peer_manager/peerdb/storage_impl.rs
+++ b/p2p/src/peer_manager/peerdb/storage_impl.rs
@@ -19,10 +19,16 @@ use super::storage::{
     PeerDbStorage, PeerDbStorageRead, PeerDbStorageWrite, PeerDbTransactionRo, PeerDbTransactionRw,
     PeerDbTransactional,
 };
+use serialization::{encoded::Encoded, DecodeAll, Encode};
+
+type ValueId = u32;
 
 storage::decl_schema! {
     /// Database schema for peer db storage
     pub Schema {
+        /// Storage for individual values
+        pub DBValue: Map<ValueId, Vec<u8>>,
+
         /// Table for known addresses
         pub DBKnownAddresses: Map<String, ()>,
 
@@ -31,6 +37,8 @@ storage::decl_schema! {
     }
 }
 
+const VALUE_ID_VERSION: ValueId = 1;
+
 pub struct PeerDbStoreTxRo<'st, B: storage::Backend>(storage::TransactionRo<'st, B, Schema>);
 
 pub struct PeerDbStoreTxRw<'st, B: storage::Backend>(storage::TransactionRw<'st, B, Schema>);
@@ -39,12 +47,12 @@ impl<'tx, B: storage::Backend + 'tx> PeerDbTransactional<'tx> for PeerDbStorageI
     type TransactionRo = PeerDbStoreTxRo<'tx, B>;
     type TransactionRw = PeerDbStoreTxRw<'tx, B>;
 
-    fn transaction_ro<'st: 'tx>(&'st self) -> crate::Result<Self::TransactionRo> {
-        self.0.transaction_ro().map_err(crate::P2pError::from).map(PeerDbStoreTxRo)
+    fn transaction_ro<'st: 'tx>(&'st self) -> Result<Self::TransactionRo, storage::Error> {
+        self.0.transaction_ro().map(PeerDbStoreTxRo)
     }
 
-    fn transaction_rw<'st: 'tx>(&'st self) -> crate::Result<Self::TransactionRw> {
-        self.0.transaction_rw(None).map_err(crate::P2pError::from).map(PeerDbStoreTxRw)
+    fn transaction_rw<'st: 'tx>(&'st self) -> Result<Self::TransactionRw, storage::Error> {
+        self.0.transaction_rw(None).map(PeerDbStoreTxRw)
     }
 }
 
@@ -60,23 +68,28 @@ impl<B: storage::Backend> PeerDbStorageImpl<B> {
 }
 
 impl<'st, B: storage::Backend> PeerDbStorageWrite for PeerDbStoreTxRw<'st, B> {
-    fn add_known_address(&mut self, address: &str) -> crate::Result<()> {
-        self.0.get_mut::<DBKnownAddresses, _>().put(address, ()).map_err(Into::into)
+    fn set_version(&mut self, version: u32) -> Result<(), storage::Error> {
+        self.0.get_mut::<DBValue, _>().put(VALUE_ID_VERSION, version.encode())
     }
 
-    fn del_known_address(&mut self, address: &str) -> crate::Result<()> {
-        self.0.get_mut::<DBKnownAddresses, _>().del(address).map_err(Into::into)
+    fn add_known_address(&mut self, address: &str) -> Result<(), storage::Error> {
+        self.0.get_mut::<DBKnownAddresses, _>().put(address, ())
     }
 
-    fn add_banned_address(&mut self, address: &str, duration: Duration) -> crate::Result<()> {
-        self.0
-            .get_mut::<DBBannedAddresses, _>()
-            .put(address, duration)
-            .map_err(Into::into)
+    fn del_known_address(&mut self, address: &str) -> Result<(), storage::Error> {
+        self.0.get_mut::<DBKnownAddresses, _>().del(address)
     }
 
-    fn del_banned_address(&mut self, address: &str) -> crate::Result<()> {
-        self.0.get_mut::<DBBannedAddresses, _>().del(address).map_err(Into::into)
+    fn add_banned_address(
+        &mut self,
+        address: &str,
+        duration: Duration,
+    ) -> Result<(), storage::Error> {
+        self.0.get_mut::<DBBannedAddresses, _>().put(address, duration)
+    }
+
+    fn del_banned_address(&mut self, address: &str) -> Result<(), storage::Error> {
+        self.0.get_mut::<DBBannedAddresses, _>().del(address)
     }
 }
 
@@ -85,19 +98,27 @@ impl<'st, B: storage::Backend> PeerDbTransactionRw for PeerDbStoreTxRw<'st, B> {
         self.0.abort()
     }
 
-    fn commit(self) -> crate::Result<()> {
-        self.0.commit().map_err(Into::into)
+    fn commit(self) -> Result<(), storage::Error> {
+        self.0.commit()
     }
 }
 
 impl<'st, B: storage::Backend> PeerDbStorageRead for PeerDbStoreTxRo<'st, B> {
-    fn get_known_addresses(&self) -> crate::Result<Vec<String>> {
+    fn get_version(&self) -> Result<Option<u32>, storage::Error> {
+        let map = self.0.get::<DBValue, _>();
+        let vec_opt = map.get(VALUE_ID_VERSION)?.as_ref().map(Encoded::decode);
+        Ok(vec_opt.map(|vec| {
+            u32::decode_all(&mut vec.as_ref()).expect("db values to be encoded correctly")
+        }))
+    }
+
+    fn get_known_addresses(&self) -> Result<Vec<String>, storage::Error> {
         let map = self.0.get::<DBKnownAddresses, _>();
         let iter = map.prefix_iter_decoded(&())?;
         Ok(iter.map(|(key, _value)| key).collect::<Vec<_>>())
     }
 
-    fn get_banned_addresses(&self) -> crate::Result<Vec<(String, Duration)>> {
+    fn get_banned_addresses(&self) -> Result<Vec<(String, Duration)>, storage::Error> {
         let map = self.0.get::<DBBannedAddresses, _>();
         let iter = map.prefix_iter_decoded(&())?;
         Ok(iter.collect::<Vec<_>>())
diff --git a/p2p/src/peer_manager/peerdb/storage_load.rs b/p2p/src/peer_manager/peerdb/storage_load.rs
new file mode 100644
index 000000000..b7cb17240
--- /dev/null
+++ b/p2p/src/peer_manager/peerdb/storage_load.rs
@@ -0,0 +1,95 @@
+// 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::{
+    collections::{BTreeMap, BTreeSet},
+    str::FromStr,
+    time::Duration,
+};
+
+use crate::error::P2pError;
+
+use super::storage::{
+    PeerDbStorage, PeerDbStorageRead, PeerDbStorageWrite, PeerDbTransactionRo, PeerDbTransactionRw,
+};
+
+const STORAGE_VERSION: u32 = 1;
+
+pub struct LoadedStorage<A, B> {
+    pub known_addresses: BTreeSet<A>,
+    pub banned_addresses: BTreeMap<B, Duration>,
+}
+
+impl<A: Ord + FromStr, B: Ord + FromStr> LoadedStorage<A, B> {
+    pub fn load_storage<S: PeerDbStorage>(storage: &S) -> crate::Result<LoadedStorage<A, B>> {
+        let tx = storage.transaction_ro()?;
+        let version = tx.get_version()?;
+        tx.close();
+
+        match version {
+            None => Self::init_storage(storage),
+            Some(STORAGE_VERSION) => Self::load_storage_v1(storage),
+            Some(version) => Err(P2pError::InvalidStorageState(format!(
+                "Unexpected PeerDb storage version: {version}"
+            ))),
+        }
+    }
+
+    fn init_storage<S: PeerDbStorage>(storage: &S) -> crate::Result<LoadedStorage<A, B>> {
+        let mut tx = storage.transaction_rw()?;
+        tx.set_version(STORAGE_VERSION)?;
+        tx.commit()?;
+        Ok(LoadedStorage {
+            known_addresses: BTreeSet::new(),
+            banned_addresses: BTreeMap::new(),
+        })
+    }
+
+    fn load_storage_v1<S: PeerDbStorage>(storage: &S) -> crate::Result<LoadedStorage<A, B>> {
+        let tx = storage.transaction_ro()?;
+
+        // TODO: Is there a concern that the number of addresses will be so huge that it'll cause a hiccup?
+        let known_addresses = tx
+            .get_known_addresses()?
+            .iter()
+            .map(|addr| {
+                addr.parse::<A>().map_err(|_err| {
+                    P2pError::InvalidStorageState(format!(
+                        "Invalid address in PeerDb storage: {addr}"
+                    ))
+                })
+            })
+            .collect::<Result<BTreeSet<_>, _>>()?;
+
+        let banned_addresses = tx
+            .get_banned_addresses()?
+            .iter()
+            .map(|(addr, duration)| {
+                addr.parse::<B>()
+                    .map_err(|_err| {
+                        P2pError::InvalidStorageState(format!(
+                            "Invalid banned address in PeerDb storage: {addr}"
+                        ))
+                    })
+                    .map(|addr| (addr, *duration))
+            })
+            .collect::<Result<BTreeMap<_, _>, _>>()?;
+
+        Ok(LoadedStorage {
+            known_addresses,
+            banned_addresses,
+        })
+    }
+}
diff --git a/p2p/src/sync/protocol.rs b/p2p/src/sync/protocol.rs
index 36e6154df..fa2fee85b 100644
--- a/p2p/src/sync/protocol.rs
+++ b/p2p/src/sync/protocol.rs
@@ -353,7 +353,8 @@ where
             // Fatal errors, simply propagate them to stop the sync manager.
             e @ (P2pError::ChannelClosed
             | P2pError::SubsystemFailure
-            | P2pError::StorageFailure(_)) => Err(e),
+            | P2pError::StorageFailure(_)
+            | P2pError::InvalidStorageState(_)) => Err(e),
         }
     }
 }


Test patch:
diff --git a/dns_server/src/crawler_p2p/crawler_manager/tests/mock_manager.rs b/dns_server/src/crawler_p2p/crawler_manager/tests/mock_manager.rs
index 3db6f2537..e77b071d7 100644
--- a/dns_server/src/crawler_p2p/crawler_manager/tests/mock_manager.rs
+++ b/dns_server/src/crawler_p2p/crawler_manager/tests/mock_manager.rs
@@ -35,7 +35,7 @@ use p2p::{
         types::{ConnectivityEvent, PeerInfo, SyncingEvent},
         ConnectivityService, NetworkingService, SyncingMessagingService,
     },
-    testing_utils::P2pTestTimeGetter,
+    testing_utils::P2pTokioTestTimeGetter,
     types::peer_id::PeerId,
 };
 use tokio::sync::mpsc;
@@ -203,7 +203,7 @@ pub fn test_crawler(
     CrawlerManager<MockNetworkingService, DnsServerStorageImpl<storage::inmemory::InMemory>>,
     MockStateRef,
     mpsc::UnboundedReceiver<DnsServerCommand>,
-    P2pTestTimeGetter,
+    P2pTokioTestTimeGetter,
 ) {
     let (conn_tx, conn_rx) = mpsc::unbounded_channel();
     let add_node = add_node.iter().map(ToString::to_string).collect();
@@ -242,7 +242,7 @@ pub fn test_crawler(
     )
     .unwrap();
 
-    let time_getter = P2pTestTimeGetter::new();
+    let time_getter = P2pTokioTestTimeGetter::new();
 
     (crawler, state, dns_server_cmd_rx, time_getter)
 }
@@ -254,7 +254,7 @@ pub async fn advance_time(
         MockNetworkingService,
         DnsServerStorageImpl<storage::inmemory::InMemory>,
     >,
-    time_getter: &P2pTestTimeGetter,
+    time_getter: &P2pTokioTestTimeGetter,
     step: Duration,
     count: u32,
 ) {
diff --git a/node/tests/cli.rs b/node/tests/cli.rs
index 0ef3cddd5..7a9b381c8 100644
--- a/node/tests/cli.rs
+++ b/node/tests/cli.rs
@@ -99,6 +99,7 @@ fn read_config_override_values() {
     let max_orphan_blocks = 2;
     let p2p_addr = "address";
     let p2p_add_node = "add_node";
+    let p2p_max_inbound_connections = 123;
     let p2p_ban_threshold = 3;
     let p2p_timeout = NonZeroU64::new(10000).unwrap();
     let p2p_ping_check_period = 30;
@@ -117,6 +118,7 @@ fn read_config_override_values() {
         tx_index_enabled: Some(false),
         p2p_addr: Some(vec![p2p_addr.to_owned()]),
         p2p_add_node: Some(vec![p2p_add_node.to_owned()]),
+        p2p_max_inbound_connections: Some(p2p_max_inbound_connections),
         p2p_ban_threshold: Some(p2p_ban_threshold),
         p2p_outbound_connection_timeout: Some(p2p_timeout),
         p2p_ping_check_period: Some(p2p_ping_check_period),
@@ -154,6 +156,10 @@ fn read_config_override_values() {
         config.p2p.clone().unwrap().added_nodes,
         Some(vec!(p2p_add_node.to_owned()))
     );
+    assert_eq!(
+        config.p2p.clone().unwrap().max_inbound_connections,
+        Some(p2p_max_inbound_connections)
+    );
     assert_eq!(
         config.p2p.clone().unwrap().ban_threshold,
         Some(p2p_ban_threshold)
diff --git a/p2p/backend-test-suite/src/block_announcement.rs b/p2p/backend-test-suite/src/block_announcement.rs
index 3f5006400..dca77bd6d 100644
--- a/p2p/backend-test-suite/src/block_announcement.rs
+++ b/p2p/backend-test-suite/src/block_announcement.rs
@@ -134,6 +134,7 @@ where
     let p2p_config = Arc::new(P2pConfig {
         bind_addresses: Vec::new(),
         added_nodes: Vec::new(),
+        max_inbound_connections: Default::default(),
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
diff --git a/p2p/backend-test-suite/src/lib.rs b/p2p/backend-test-suite/src/lib.rs
index 183ebf19e..3bce3f506 100644
--- a/p2p/backend-test-suite/src/lib.rs
+++ b/p2p/backend-test-suite/src/lib.rs
@@ -22,7 +22,6 @@ mod utils;
 mod ban;
 mod block_announcement;
 mod connect;
-mod peer_manager_peerdb;
 
 use std::fmt::Debug;
 
@@ -60,6 +59,5 @@ where
         .chain(connect::tests::<T, N, A>())
         .chain(block_announcement::tests::<T, N, A>())
         .chain(ban::tests::<T, N, A>())
-        .chain(peer_manager_peerdb::tests::<T, N, A>())
         .collect()
 }
diff --git a/p2p/src/net/default_backend/tests.rs b/p2p/src/net/default_backend/tests.rs
new file mode 100644
index 000000000..09ee84d95
--- /dev/null
+++ b/p2p/src/net/default_backend/tests.rs
@@ -0,0 +1,282 @@
+// Copyright (c) 2021-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 super::{transport::NoiseTcpTransport, *};
+use crate::error::DialError;
+use crate::testing_utils::{TestTransportChannel, TestTransportMaker, TestTransportTcp};
+use crate::{
+    net::default_backend::transport::{MpscChannelTransport, TcpTransportSocket},
+    testing_utils::TestTransportNoise,
+};
+use common::primitives::semver::SemVer;
+use std::fmt::Debug;
+
+async fn connect_to_remote<A, T>()
+where
+    A: TestTransportMaker<Transport = T, Address = T::Address>,
+    T: TransportSocket + Debug,
+{
+    let config = Arc::new(common::chain::config::create_mainnet());
+    let p2p_config: Arc<config::P2pConfig> = Arc::new(Default::default());
+
+    let (mut conn1, _) = DefaultNetworkingService::<T>::start(
+        A::make_transport(),
+        vec![A::make_address()],
+        Arc::clone(&config),
+        Arc::clone(&p2p_config),
+    )
+    .await
+    .unwrap();
+
+    let (conn2, _) = DefaultNetworkingService::<T>::start(
+        A::make_transport(),
+        vec![A::make_address()],
+        Arc::clone(&config),
+        Arc::clone(&p2p_config),
+    )
+    .await
+    .unwrap();
+
+    let addr = conn2.local_addresses();
+    conn1.connect(addr[0].clone()).unwrap();
+
+    if let Ok(ConnectivityEvent::OutboundAccepted {
+        address,
+        peer_info,
+        receiver_address: _,
+    }) = conn1.poll_next().await
+    {
+        assert_eq!(address, conn2.local_addresses()[0]);
+        assert_eq!(peer_info.network, *config.magic_bytes());
+        assert_eq!(peer_info.version, SemVer::new(0, 1, 0));
+        assert_eq!(peer_info.agent, None);
+        assert_eq!(
+            peer_info.subscriptions,
+            [PubSubTopic::Blocks, PubSubTopic::Transactions, PubSubTopic::PeerAddresses]
+                .into_iter()
+                .collect()
+        );
+    } else {
+        panic!("invalid event received");
+    }
+}
+
+#[tokio::test]
+async fn connect_to_remote_tcp() {
+    connect_to_remote::<TestTransportTcp, TcpTransportSocket>().await;
+}
+
+#[tokio::test]
+async fn connect_to_remote_channels() {
+    connect_to_remote::<TestTransportChannel, MpscChannelTransport>().await;
+}
+
+#[tokio::test]
+async fn connect_to_remote_noise() {
+    connect_to_remote::<TestTransportNoise, NoiseTcpTransport>().await;
+}
+
+async fn accept_incoming<A, T>()
+where
+    A: TestTransportMaker<Transport = T, Address = T::Address>,
+    T: TransportSocket,
+{
+    let config = Arc::new(common::chain::config::create_mainnet());
+    let p2p_config: Arc<config::P2pConfig> = Arc::new(Default::default());
+
+    let (mut conn1, _) = DefaultNetworkingService::<T>::start(
+        A::make_transport(),
+        vec![A::make_address()],
+        Arc::clone(&config),
+        Arc::clone(&p2p_config),
+    )
+    .await
+    .unwrap();
+
+    let (mut conn2, _) = DefaultNetworkingService::<T>::start(
+        A::make_transport(),
+        vec![A::make_address()],
+        Arc::clone(&config),
+        Arc::clone(&p2p_config),
+    )
+    .await
+    .unwrap();
+
+    let bind_address = conn2.local_addresses();
+    conn1.connect(bind_address[0].clone()).unwrap();
+    let res2 = conn2.poll_next().await;
+    match res2.unwrap() {
+        ConnectivityEvent::InboundAccepted {
+            address: _,
+            peer_info,
+            receiver_address: _,
+        } => {
+            assert_eq!(peer_info.network, *config.magic_bytes());
+            assert_eq!(
+                peer_info.version,
+                common::primitives::semver::SemVer::new(0, 1, 0),
+            );
+            assert_eq!(peer_info.agent, None);
+        }
+        _ => panic!("invalid event received, expected incoming connection"),
+    }
+}
+
+#[tokio::test]
+async fn accept_incoming_tcp() {
+    accept_incoming::<TestTransportTcp, TcpTransportSocket>().await;
+}
+
+#[tokio::test]
+async fn accept_incoming_channels() {
+    accept_incoming::<TestTransportChannel, MpscChannelTransport>().await;
+}
+
+#[tokio::test]
+async fn accept_incoming_noise() {
+    accept_incoming::<TestTransportNoise, NoiseTcpTransport>().await;
+}
+
+async fn disconnect<A, T>()
+where
+    A: TestTransportMaker<Transport = T, Address = T::Address>,
+    T: TransportSocket,
+{
+    let config = Arc::new(common::chain::config::create_mainnet());
+    let p2p_config: Arc<config::P2pConfig> = Arc::new(Default::default());
+
+    let (mut conn1, _) = DefaultNetworkingService::<T>::start(
+        A::make_transport(),
+        vec![A::make_address()],
+        Arc::clone(&config),
+        Arc::clone(&p2p_config),
+    )
+    .await
+    .unwrap();
+    let (mut conn2, _) = DefaultNetworkingService::<T>::start(
+        A::make_transport(),
+        vec![A::make_address()],
+        config,
+        p2p_config,
+    )
+    .await
+    .unwrap();
+
+    conn1.connect(conn2.local_addresses()[0].clone()).unwrap();
+    let res2 = conn2.poll_next().await;
+
+    match res2.unwrap() {
+        ConnectivityEvent::InboundAccepted {
+            address: _,
+            peer_info,
+            receiver_address: _,
+        } => {
+            conn2.disconnect(peer_info.peer_id).unwrap();
+        }
+        _ => panic!("invalid event received, expected incoming connection"),
+    }
+}
+
+#[tokio::test]
+async fn disconnect_tcp() {
+    disconnect::<TestTransportTcp, TcpTransportSocket>().await;
+}
+
+#[tokio::test]
+async fn disconnect_channels() {
+    disconnect::<TestTransportChannel, MpscChannelTransport>().await;
+}
+
+#[tokio::test]
+async fn disconnect_noise() {
+    disconnect::<TestTransportNoise, NoiseTcpTransport>().await;
+}
+
+async fn self_connect<A, T>()
+where
+    A: TestTransportMaker<Transport = T, Address = T::Address>,
+    T: TransportSocket + Debug,
+{
+    let config = Arc::new(common::chain::config::create_mainnet());
+    let p2p_config: Arc<config::P2pConfig> = Arc::new(Default::default());
+
+    let (mut conn1, _) = DefaultNetworkingService::<T>::start(
+        A::make_transport(),
+        vec![A::make_address()],
+        Arc::clone(&config),
+        Arc::clone(&p2p_config),
+    )
+    .await
+    .unwrap();
+
+    let (conn2, _) = DefaultNetworkingService::<T>::start(
+        A::make_transport(),
+        vec![A::make_address()],
+        Arc::clone(&config),
+        Arc::clone(&p2p_config),
+    )
+    .await
+    .unwrap();
+
+    // Try connect to self
+    let addr = conn1.local_addresses();
+    conn1.connect(addr[0].clone()).unwrap();
+
+    // ConnectionError should be reported
+    if let Ok(ConnectivityEvent::ConnectionError { address, error }) = conn1.poll_next().await {
+        assert_eq!(address, conn1.local_addresses()[0]);
+        assert_eq!(error, P2pError::DialError(DialError::AttemptToDialSelf));
+    } else {
+        panic!("invalid event received");
+    }
+
+    // Check that we can still connect normally after
+    let addr = conn2.local_addresses();
+    conn1.connect(addr[0].clone()).unwrap();
+    if let Ok(ConnectivityEvent::OutboundAccepted {
+        address,
+        peer_info,
+        receiver_address: _,
+    }) = conn1.poll_next().await
+    {
+        assert_eq!(address, conn2.local_addresses()[0]);
+        assert_eq!(peer_info.network, *config.magic_bytes());
+        assert_eq!(peer_info.version, SemVer::new(0, 1, 0));
+        assert_eq!(peer_info.agent, None);
+        assert_eq!(
+            peer_info.subscriptions,
+            [PubSubTopic::Blocks, PubSubTopic::Transactions, PubSubTopic::PeerAddresses]
+                .into_iter()
+                .collect()
+        );
+    } else {
+        panic!("invalid event received");
+    }
+}
+
+#[tokio::test]
+async fn self_connect_tcp() {
+    self_connect::<TestTransportTcp, TcpTransportSocket>().await;
+}
+
+#[tokio::test]
+async fn self_connect_channels() {
+    self_connect::<TestTransportChannel, MpscChannelTransport>().await;
+}
+
+#[tokio::test]
+async fn self_connect_noise() {
+    self_connect::<TestTransportNoise, NoiseTcpTransport>().await;
+}
diff --git a/p2p/src/net/default_backend/transport/impls/stream_adapter/wrapped_transport/tests.rs b/p2p/src/net/default_backend/transport/impls/stream_adapter/wrapped_transport/tests.rs
index 9af1400ba..fc58661d5 100644
--- a/p2p/src/net/default_backend/transport/impls/stream_adapter/wrapped_transport/tests.rs
+++ b/p2p/src/net/default_backend/transport/impls/stream_adapter/wrapped_transport/tests.rs
@@ -26,7 +26,7 @@ use tokio::{
 };
 
 use crate::testing_utils::{
-    P2pTestTimeGetter, TestTransportChannel, TestTransportMaker, TestTransportTcp,
+    P2pTokioTestTimeGetter, TestTransportChannel, TestTransportMaker, TestTransportTcp,
 };
 use common::{
     chain::Block,
@@ -266,7 +266,7 @@ async fn pending_handshakes() {
 
 #[tokio::test]
 async fn handshake_timeout() {
-    let time_getter = P2pTestTimeGetter::new();
+    let time_getter = P2pTokioTestTimeGetter::new();
     let transport = WrappedTransportSocket::<NoiseEncryptionAdapter, TcpTransportSocket>::new(
         NoiseEncryptionAdapter::gen_new(),
         TcpTransportSocket::new(),
diff --git a/p2p/backend-test-suite/src/peer_manager_peerdb.rs b/p2p/src/peer_manager/peerdb/tests.rs
similarity index 56%
rename from p2p/backend-test-suite/src/peer_manager_peerdb.rs
rename to p2p/src/peer_manager/peerdb/tests.rs
index 3b6368aee..8eb7bbed5 100644
--- a/p2p/backend-test-suite/src/peer_manager_peerdb.rs
+++ b/p2p/src/peer_manager/peerdb/tests.rs
@@ -1,4 +1,4 @@
-// Copyright (c) 2022 RBB S.r.l
+// Copyright (c) 2022-2023 RBB S.r.l
 // opensource@mintlayer.org
 // SPDX-License-Identifier: MIT
 // Licensed under the MIT License;
@@ -13,34 +13,28 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Test functions need to be marked async because of the `tests!` macro.
-#![allow(clippy::unused_async)]
-
 use std::{sync::Arc, time::Duration};
 
-use p2p::{
+use crate::{
     config::P2pConfig,
-    net::{AsBannableAddress, NetworkingService},
-    peer_manager::peerdb::{
-        storage::{PeerDbStorageRead, PeerDbTransactional},
-        PeerDb,
+    net::AsBannableAddress,
+    peer_manager::peerdb::storage::{PeerDbStorageRead, PeerDbTransactional},
+    testing_utils::{
+        peerdb_inmemory_store, P2pBasicTestTimeGetter, RandomAddressMaker, TestTcpAddressMaker,
     },
-    testing_utils::{peerdb_inmemory_store, P2pTestTimeGetter, RandomAddressMaker},
 };
 
-tests![unban_peer,];
+use super::PeerDb;
 
-async fn unban_peer<T, N, A>()
-where
-    N: NetworkingService,
-    A: RandomAddressMaker<Address = N::Address>,
-{
+#[test]
+fn unban_peer() {
     let db_store = peerdb_inmemory_store();
-    let time_getter = P2pTestTimeGetter::new();
-    let mut peerdb = PeerDb::<N, _>::new(
+    let time_getter = P2pBasicTestTimeGetter::new();
+    let mut peerdb = PeerDb::new(
         Arc::new(P2pConfig {
             bind_addresses: Default::default(),
             added_nodes: Default::default(),
+            max_inbound_connections: Default::default(),
             ban_threshold: Default::default(),
             ban_duration: Duration::from_secs(60).into(),
             outbound_connection_timeout: Default::default(),
@@ -57,26 +51,16 @@ where
     )
     .unwrap();
 
-    let address = A::new();
-    peerdb.ban_peer(&address).unwrap();
+    let address = TestTcpAddressMaker::new();
+    peerdb.ban_peer(&address);
 
-    assert!(peerdb.is_address_banned(&address.as_bannable()).unwrap());
-    let banned_addresses = peerdb
-        .get_storage_mut()
-        .transaction_ro()
-        .unwrap()
-        .get_banned_addresses()
-        .unwrap();
+    assert!(peerdb.is_address_banned(&address.as_bannable()));
+    let banned_addresses = peerdb.storage.transaction_ro().unwrap().get_banned_addresses().unwrap();
     assert_eq!(banned_addresses.len(), 1);
 
-    time_getter.advance_time(Duration::from_secs(120)).await;
+    time_getter.advance_time(Duration::from_secs(120));
 
-    assert!(!peerdb.is_address_banned(&address.as_bannable()).unwrap());
-    let banned_addresses = peerdb
-        .get_storage_mut()
-        .transaction_ro()
-        .unwrap()
-        .get_banned_addresses()
-        .unwrap();
+    assert!(!peerdb.is_address_banned(&address.as_bannable()));
+    let banned_addresses = peerdb.storage.transaction_ro().unwrap().get_banned_addresses().unwrap();
     assert_eq!(banned_addresses.len(), 0);
 }
diff --git a/p2p/src/peer_manager/tests/ban.rs b/p2p/src/peer_manager/tests/ban.rs
index e31727510..56eec84a8 100644
--- a/p2p/src/peer_manager/tests/ban.rs
+++ b/p2p/src/peer_manager/tests/ban.rs
@@ -61,11 +61,11 @@ where
     )
     .await;
     let peer_id = peer_info.peer_id;
-    pm2.accept_inbound_connection(address, peer_info, None).unwrap();
+    pm2.accept_connection(address, Role::Inbound, peer_info, None);
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 1000), Ok(()));
+    pm2.adjust_peer_score(peer_id, 1000);
     let addr1 = pm1.peer_connectivity_handle.local_addresses()[0].clone().as_bannable();
-    assert!(pm2.peerdb.is_address_banned(&addr1).unwrap());
+    assert!(pm2.peerdb.is_address_banned(&addr1));
     let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
     assert!(std::matches!(
         event,
@@ -108,11 +108,11 @@ where
     )
     .await;
     let peer_id = peer_info.peer_id;
-    pm2.accept_inbound_connection(address, peer_info, None).unwrap();
+    pm2.accept_connection(address, Role::Inbound, peer_info, None);
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 1000), Ok(()));
+    pm2.adjust_peer_score(peer_id, 1000);
     let addr1 = pm1.peer_connectivity_handle.local_addresses()[0].clone().as_bannable();
-    assert!(pm2.peerdb.is_address_banned(&addr1).unwrap());
+    assert!(pm2.peerdb.is_address_banned(&addr1));
     let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
     assert!(std::matches!(
         event,
@@ -163,25 +163,25 @@ where
     )
     .await;
     let peer_id = peer_info1.peer_id;
-    pm2.accept_inbound_connection(address, peer_info1, None).unwrap();
+    pm2.accept_connection(address, Role::Inbound, peer_info1, None);
 
     let remote_addr = pm1.peer_connectivity_handle.local_addresses()[0].clone();
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 10), Ok(()));
-    assert!(!pm2.peerdb.is_address_banned(&remote_addr.as_bannable()).unwrap());
+    pm2.adjust_peer_score(peer_id, 10);
+    assert!(!pm2.peerdb.is_address_banned(&remote_addr.as_bannable()));
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 90), Ok(()));
-    assert!(pm2.peerdb.is_address_banned(&remote_addr.as_bannable()).unwrap());
+    pm2.adjust_peer_score(peer_id, 90);
+    assert!(pm2.peerdb.is_address_banned(&remote_addr.as_bannable()));
 
     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).unwrap();
+    pm2.handle_connectivity_event(event.unwrap());
 
     let (tx, rx) = oneshot_nofail::channel();
-    pm2.connect(remote_addr, Some(tx)).unwrap();
+    pm2.connect(remote_addr, Some(tx));
     let res = rx.await.unwrap();
     match res {
         Err(P2pError::PeerError(PeerError::BannedAddress(_))) => {}
@@ -220,7 +220,7 @@ where
         make_peer_manager::<S>(A::make_transport(), A::make_address(), Arc::clone(&config)).await;
 
     // invalid magic bytes
-    let res = peer_manager.accept_connection(
+    let res = peer_manager.try_accept_connection(
         B::new(),
         Role::Outbound,
         net::types::PeerInfo {
@@ -232,12 +232,13 @@ where
         },
         None,
     );
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
+    assert!(res.is_err());
     assert!(!peer_manager.is_peer_connected(peer_id));
 
     // invalid version
-    let res = peer_manager.accept_connection(
-        B::new(),
+    let address = B::new();
+    let res = peer_manager.try_accept_connection(
+        address,
         Role::Outbound,
         net::types::PeerInfo {
             peer_id,
@@ -248,12 +249,12 @@ where
         },
         None,
     );
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
+    assert!(res.is_err());
     assert!(!peer_manager.is_peer_connected(peer_id));
 
     // valid connection
     let address = B::new();
-    let res = peer_manager.accept_connection(
+    let res = peer_manager.try_accept_connection(
         address.clone(),
         Role::Outbound,
         net::types::PeerInfo {
@@ -266,9 +267,8 @@ where
         None,
     );
     assert!(res.is_ok());
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
     assert!(peer_manager.is_peer_connected(peer_id));
-    assert!(!peer_manager.peerdb.is_address_banned(&address.as_bannable()).unwrap());
+    assert!(!peer_manager.peerdb.is_address_banned(&address.as_bannable()));
 }
 
 #[tokio::test]
@@ -313,8 +313,9 @@ where
         make_peer_manager::<S>(A::make_transport(), A::make_address(), Arc::clone(&config)).await;
 
     // invalid magic bytes
-    let res = peer_manager.accept_inbound_connection(
+    let res = peer_manager.try_accept_connection(
         B::new(),
+        Role::Inbound,
         net::types::PeerInfo {
             peer_id,
             network: [1, 2, 3, 4],
@@ -324,12 +325,13 @@ where
         },
         None,
     );
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
+    assert!(res.is_err());
     assert!(!peer_manager.is_peer_connected(peer_id));
 
     // invalid version
-    let res = peer_manager.accept_inbound_connection(
+    peer_manager.accept_connection(
         B::new(),
+        Role::Inbound,
         net::types::PeerInfo {
             peer_id,
             network: *config.magic_bytes(),
@@ -339,26 +341,28 @@ where
         },
         None,
     );
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
     assert!(!peer_manager.is_peer_connected(peer_id));
 
     // valid connection
     let address = B::new();
-    let res = peer_manager.accept_inbound_connection(
-        address.clone(),
-        net::types::PeerInfo {
-            peer_id,
-            network: *config.magic_bytes(),
-            version: SemVer::new(0, 1, 0),
-            agent: None,
-            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
-        },
-        None,
-    );
-    assert!(res.is_ok());
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
+    peer_manager
+        .try_accept_connection(
+            address.clone(),
+            Role::Inbound,
+            net::types::PeerInfo {
+                peer_id,
+                network: *config.magic_bytes(),
+                version: SemVer::new(0, 1, 0),
+                agent: None,
+                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
+                    .into_iter()
+                    .collect(),
+            },
+            None,
+        )
+        .unwrap();
     assert!(peer_manager.is_peer_connected(peer_id));
-    assert!(!peer_manager.peerdb.is_address_banned(&address.as_bannable()).unwrap());
+    assert!(!peer_manager.peerdb.is_address_banned(&address.as_bannable()));
 }
 
 #[tokio::test]
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index d7497b1c9..cca9800a6 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -22,11 +22,11 @@ use std::{
 use tokio::time::timeout;
 
 use crate::{
-    config::P2pConfig,
+    config::{MaxInboundConnections, P2pConfig},
     net::types::Role,
     peer_manager::tests::{get_connected_peers, run_peer_manager},
     testing_utils::{
-        connect_services, get_connectivity_event, peerdb_inmemory_store, P2pTestTimeGetter,
+        connect_services, get_connectivity_event, peerdb_inmemory_store, P2pTokioTestTimeGetter,
         TestTransportChannel, TestTransportMaker, TestTransportNoise, TestTransportTcp,
     },
     types::peer_id::PeerId,
@@ -124,10 +124,10 @@ where
     });
 
     // "discover" the other networking service
-    pm1.peerdb.peer_discovered(&addr).unwrap();
-    pm1.heartbeat().unwrap();
+    pm1.peerdb.peer_discovered(addr);
+    pm1.heartbeat().await;
 
-    assert_eq!(pm1.pending_connects.len(), 1);
+    assert_eq!(pm1.pending_outbound_connects.len(), 1);
     assert!(std::matches!(
         pm1.peer_connectivity_handle.poll_next().await,
         Ok(net::types::ConnectivityEvent::OutboundAccepted { .. })
@@ -260,10 +260,7 @@ where
         &mut pm2.peer_connectivity_handle,
     )
     .await;
-    assert_eq!(
-        pm2.accept_inbound_connection(address, peer_info, None),
-        Ok(())
-    );
+    pm2.try_accept_connection(address, Role::Inbound, peer_info, None).unwrap();
 }
 
 #[tokio::test]
@@ -316,7 +313,7 @@ where
     .await;
 
     assert_eq!(
-        pm2.accept_inbound_connection(address, peer_info, None),
+        pm2.try_accept_connection(address, Role::Inbound, peer_info, None),
         Err(P2pError::ProtocolError(ProtocolError::DifferentNetwork(
             [1, 2, 3, 4],
             *config::create_mainnet().magic_bytes(),
@@ -420,12 +417,9 @@ where
     let mut pm2 = make_peer_manager::<T>(A::make_transport(), addr2, Arc::clone(&config)).await;
 
     for peer in peers.into_iter() {
-        pm1.accept_connection(peer.0, Role::Inbound, peer.1, None).unwrap();
+        pm1.try_accept_connection(peer.0, Role::Inbound, peer.1, None).unwrap();
     }
-    assert_eq!(
-        pm1.active_peer_count(),
-        peer_manager::MAX_ACTIVE_CONNECTIONS
-    );
+    assert_eq!(pm1.inbound_peer_count(), *MaxInboundConnections::default());
 
     let (_address, peer_info, _) = connect_services::<T>(
         &mut pm1.peer_connectivity_handle,
@@ -448,7 +442,7 @@ where
 #[tokio::test]
 async fn inbound_connection_too_many_peers_tcp() {
     let config = Arc::new(config::create_mainnet());
-    let peers = (0..peer_manager::MAX_ACTIVE_CONNECTIONS)
+    let peers = (0..*MaxInboundConnections::default())
         .map(|index| {
             (
                 format!("127.0.0.1:{}", index + 10000).parse().expect("valid address"),
@@ -475,7 +469,7 @@ async fn inbound_connection_too_many_peers_tcp() {
 #[tokio::test]
 async fn inbound_connection_too_many_peers_channels() {
     let config = Arc::new(config::create_mainnet());
-    let peers = (0..peer_manager::MAX_ACTIVE_CONNECTIONS)
+    let peers = (0..*MaxInboundConnections::default())
         .map(|index| {
             (
                 format!("{}", index + 10000).parse().expect("valid address"),
@@ -502,7 +496,7 @@ async fn inbound_connection_too_many_peers_channels() {
 #[tokio::test]
 async fn inbound_connection_too_many_peers_noise() {
     let config = Arc::new(config::create_mainnet());
-    let peers = (0..peer_manager::MAX_ACTIVE_CONNECTIONS)
+    let peers = (0..*MaxInboundConnections::default())
         .map(|index| {
             (
                 format!("127.0.0.1:{}", index + 10000).parse().expect("valid address"),
@@ -673,13 +667,14 @@ where
     T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
 {
-    let time_getter = P2pTestTimeGetter::new();
+    let time_getter = P2pTokioTestTimeGetter::new();
     let chain_config = Arc::new(config::create_mainnet());
 
     // Start first peer manager
     let p2p_config_1 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
         added_nodes: Default::default(),
+        max_inbound_connections: Default::default(),
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
@@ -710,6 +705,7 @@ where
     let p2p_config_2 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
         added_nodes: bind_addresses,
+        max_inbound_connections: Default::default(),
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
@@ -774,12 +770,13 @@ where
 {
     let chain_config = Arc::new(config::create_mainnet());
 
-    let time_getter = P2pTestTimeGetter::new();
+    let time_getter = P2pTokioTestTimeGetter::new();
 
     // Start the first peer manager
     let p2p_config_1 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
         added_nodes: Default::default(),
+        max_inbound_connections: Default::default(),
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
@@ -811,6 +808,7 @@ where
     let p2p_config_2 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
         added_nodes: bind_addresses.clone(),
+        max_inbound_connections: Default::default(),
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
@@ -835,6 +833,7 @@ where
     let p2p_config_3 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
         added_nodes: bind_addresses,
+        max_inbound_connections: Default::default(),
         ban_threshold: Default::default(),
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
diff --git a/p2p/src/peer_manager/tests/ping.rs b/p2p/src/peer_manager/tests/ping.rs
index 4d63ccad2..4f2fcb972 100644
--- a/p2p/src/peer_manager/tests/ping.rs
+++ b/p2p/src/peer_manager/tests/ping.rs
@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use common::chain::config;
 
@@ -30,7 +30,7 @@ use crate::{
         types::{ConnectivityEvent, PeerInfo},
     },
     peer_manager::PeerManager,
-    testing_utils::{peerdb_inmemory_store, P2pTestTimeGetter},
+    testing_utils::{peerdb_inmemory_store, P2pTokioTestTimeGetter},
     types::peer_id::PeerId,
 };
 
@@ -39,7 +39,21 @@ 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 p2p_config: Arc<P2pConfig> = Arc::new(P2pConfig {
+        bind_addresses: Default::default(),
+        added_nodes: Default::default(),
+        max_inbound_connections: Default::default(),
+        ban_threshold: Default::default(),
+        ban_duration: Default::default(),
+        outbound_connection_timeout: Default::default(),
+        ping_check_period: Duration::from_secs(1).into(),
+        ping_timeout: Duration::from_secs(5).into(),
+        node_type: Default::default(),
+        allow_discover_private_ips: Default::default(),
+        msg_header_count_limit: Default::default(),
+        msg_max_locator_count: Default::default(),
+        max_request_blocks_count: Default::default(),
+    });
     let ping_check_period = *p2p_config.ping_check_period;
     let ping_timeout = *p2p_config.ping_timeout;
 
@@ -47,7 +61,7 @@ async fn ping_timeout() {
     let (conn_tx, conn_rx) = tokio::sync::mpsc::unbounded_channel();
     let (_peer_tx, peer_rx) =
         tokio::sync::mpsc::unbounded_channel::<PeerManagerEvent<TestNetworkingService>>();
-    let time_getter = P2pTestTimeGetter::new();
+    let time_getter = P2pTokioTestTimeGetter::new();
     let (sync_tx, _sync_rx) = tokio::sync::mpsc::unbounded_channel();
     let connectivity_handle = ConnectivityHandle::<TestNetworkingService, TcpTransportSocket>::new(
         vec![],
@@ -86,7 +100,7 @@ async fn ping_timeout() {
         .unwrap();
 
     // Receive ping requests and send responses normally
-    for _ in 0..30 {
+    for _ in 0..5 {
         time_getter.advance_time(ping_check_period).await;
 
         let event = cmd_rx.recv().await.unwrap();
diff --git a/p2p/src/testing_utils.rs b/p2p/src/testing_utils.rs
index 201871ea9..fc8b4b372 100644
--- a/p2p/src/testing_utils.rs
+++ b/p2p/src/testing_utils.rs
@@ -230,11 +230,11 @@ pub fn peerdb_inmemory_store() -> PeerDbStorageImpl<storage::inmemory::InMemory>
     PeerDbStorageImpl::new(storage).unwrap()
 }
 
-pub struct P2pTestTimeGetter {
+pub struct P2pBasicTestTimeGetter {
     current_time_millis: Arc<AtomicU64>,
 }
 
-impl P2pTestTimeGetter {
+impl P2pBasicTestTimeGetter {
     pub fn new() -> Self {
         let current_time = std::time::SystemTime::now()
             .duration_since(std::time::SystemTime::UNIX_EPOCH)
@@ -252,10 +252,30 @@ impl P2pTestTimeGetter {
         }))
     }
 
-    pub async fn advance_time(&self, duration: Duration) {
-        tokio::time::pause();
+    pub fn advance_time(&self, duration: Duration) {
         self.current_time_millis
             .fetch_add(duration.as_millis() as u64, Ordering::SeqCst);
+    }
+}
+
+pub struct P2pTokioTestTimeGetter {
+    time_getter: P2pBasicTestTimeGetter,
+}
+
+impl P2pTokioTestTimeGetter {
+    pub fn new() -> Self {
+        Self {
+            time_getter: P2pBasicTestTimeGetter::new(),
+        }
+    }
+
+    pub fn get_time_getter(&self) -> TimeGetter {
+        self.time_getter.get_time_getter()
+    }
+
+    pub async fn advance_time(&self, duration: Duration) {
+        tokio::time::pause();
+        self.time_getter.advance_time(duration);
         tokio::time::advance(duration).await;
         tokio::time::resume();
     }