Repository: mintlayer/mintlayer-core
Pull request #646 (2023-01-23 16:00:46)

Basic peer discovery
This adds a basic peer discovery mechanism similar to Bitcoin's. When a node connects somewhere, it tries to load addresses from remote peers and also advertises its own discovered address. Such advertised addresses are relayed to random peers (two peers) and spread across the network this way.


Patch:
diff --git a/node/src/config_files/p2p.rs b/node/src/config_files/p2p.rs
index 0830b23f4..7d7050e73 100644
--- a/node/src/config_files/p2p.rs
+++ b/node/src/config_files/p2p.rs
@@ -77,6 +77,9 @@ impl From<P2pConfigFile> for P2pConfig {
                 .map(|t| Duration::from_secs(t.into()))
                 .into(),
             node_type: c.node_type.map(Into::into).into(),
+            allow_discover_private_ips: Default::default(),
+            heartbeat_interval_min: Default::default(),
+            heartbeat_interval_max: Default::default(),
         }
     }
 }
diff --git a/p2p/src/config.rs b/p2p/src/config.rs
index 52246c987..44e1afd43 100644
--- a/p2p/src/config.rs
+++ b/p2p/src/config.rs
@@ -30,6 +30,9 @@ make_config_setting!(
     [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect()
 );
 make_config_setting!(NodeTypeSetting, NodeType, NodeType::Full);
+make_config_setting!(AllowDiscoverPrivateIps, bool, false);
+make_config_setting!(HeartbeatIntervalMin, Duration, Duration::from_secs(5));
+make_config_setting!(HeartbeatIntervalMax, Duration, Duration::from_secs(30));
 
 /// A node type.
 #[derive(Debug, Copy, Clone)]
@@ -71,4 +74,10 @@ pub struct P2pConfig {
     pub outbound_connection_timeout: OutboundConnectionTimeout,
     /// A node type.
     pub node_type: NodeTypeSetting,
+    /// Allow announcing and discovering local and private IPs. Should be used for testing only.
+    pub allow_discover_private_ips: AllowDiscoverPrivateIps,
+    /// Lower bound for how often `PeerManager::heartbeat` is called. Should be used for testing only.
+    pub heartbeat_interval_min: HeartbeatIntervalMin,
+    /// Upper bound for how often `PeerManager::heartbeat` is called. Should be used for testing only.
+    pub heartbeat_interval_max: HeartbeatIntervalMin,
 }
diff --git a/p2p/src/message.rs b/p2p/src/message.rs
index ea167e945..3e8b428c9 100644
--- a/p2p/src/message.rs
+++ b/p2p/src/message.rs
@@ -63,6 +63,11 @@ impl BlockListRequest {
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub struct AddrListRequest {}
 
+#[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
+pub struct AnnounceAddrRequest {
+    pub address: PeerAddress,
+}
+
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub enum Request {
     #[codec(index = 0)]
@@ -71,6 +76,8 @@ pub enum Request {
     BlockListRequest(BlockListRequest),
     #[codec(index = 2)]
     AddrListRequest(AddrListRequest),
+    #[codec(index = 3)]
+    AnnounceAddrRequest(AnnounceAddrRequest),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -82,6 +89,7 @@ pub enum SyncRequest {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum PeerManagerRequest {
     AddrListRequest(AddrListRequest),
+    AnnounceAddrRequest(AnnounceAddrRequest),
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
@@ -123,18 +131,11 @@ impl BlockListResponse {
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
-pub struct AddrListResponse {
-    addresses: Vec<PeerAddress>,
-}
+pub struct AnnounceAddrResponse {}
 
-impl AddrListResponse {
-    pub fn new(addresses: Vec<PeerAddress>) -> Self {
-        Self { addresses }
-    }
-
-    pub fn addresses(&self) -> &[PeerAddress] {
-        &self.addresses
-    }
+#[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
+pub struct AddrListResponse {
+    pub addresses: Vec<PeerAddress>,
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
@@ -145,6 +146,8 @@ pub enum Response {
     BlockListResponse(BlockListResponse),
     #[codec(index = 2)]
     AddrListResponse(AddrListResponse),
+    #[codec(index = 3)]
+    AnnounceAddrResponse(AnnounceAddrResponse),
 }
 
 #[derive(Debug, Clone)]
@@ -156,6 +159,7 @@ pub enum SyncResponse {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum PeerManagerResponse {
     AddrListResponse(AddrListResponse),
+    AnnounceAddrResponse(AnnounceAddrResponse),
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
@@ -168,6 +172,9 @@ impl From<PeerManagerRequest> for Request {
     fn from(request: PeerManagerRequest) -> Self {
         match request {
             PeerManagerRequest::AddrListRequest(request) => Request::AddrListRequest(request),
+            PeerManagerRequest::AnnounceAddrRequest(request) => {
+                Request::AnnounceAddrRequest(request)
+            }
         }
     }
 }
@@ -176,6 +183,9 @@ impl From<PeerManagerResponse> for Response {
     fn from(response: PeerManagerResponse) -> Self {
         match response {
             PeerManagerResponse::AddrListResponse(response) => Response::AddrListResponse(response),
+            PeerManagerResponse::AnnounceAddrResponse(response) => {
+                Response::AnnounceAddrResponse(response)
+            }
         }
     }
 }
diff --git a/p2p/src/net/default_backend/backend.rs b/p2p/src/net/default_backend/backend.rs
index 24b7da57e..0de4090ff 100644
--- a/p2p/src/net/default_backend/backend.rs
+++ b/p2p/src/net/default_backend/backend.rs
@@ -292,6 +292,15 @@ where
                 })
                 .await
                 .map_err(P2pError::from),
+            message::Request::AnnounceAddrRequest(request) => self
+                .conn_tx
+                .send(ConnectivityEvent::Request {
+                    peer_id,
+                    request_id,
+                    request: PeerManagerRequest::AnnounceAddrRequest(request),
+                })
+                .await
+                .map_err(P2pError::from),
         }
     }
 
@@ -332,6 +341,15 @@ where
                 })
                 .await
                 .map_err(P2pError::from),
+            message::Response::AnnounceAddrResponse(response) => self
+                .conn_tx
+                .send(ConnectivityEvent::Response {
+                    peer_id,
+                    request_id,
+                    response: PeerManagerResponse::AnnounceAddrResponse(response),
+                })
+                .await
+                .map_err(P2pError::from),
         }
     }
 
diff --git a/p2p/src/net/default_backend/mod.rs b/p2p/src/net/default_backend/mod.rs
index 73d14fe5c..88a7e8b87 100644
--- a/p2p/src/net/default_backend/mod.rs
+++ b/p2p/src/net/default_backend/mod.rs
@@ -211,8 +211,8 @@ where
         Ok(())
     }
 
-    async fn local_addresses(&self) -> crate::Result<Vec<S::Address>> {
-        Ok(self.local_addresses.clone())
+    fn local_addresses(&self) -> &[S::Address] {
+        &self.local_addresses
     }
 
     async fn poll_next(&mut self) -> crate::Result<ConnectivityEvent<S>> {
@@ -262,9 +262,6 @@ where
             types::ConnectivityEvent::Misbehaved { peer_id, error } => {
                 Ok(ConnectivityEvent::Misbehaved { peer_id, error })
             }
-            types::ConnectivityEvent::AddressDiscovered { address } => {
-                Ok(ConnectivityEvent::AddressDiscovered { address })
-            }
         }
     }
 }
@@ -397,7 +394,7 @@ mod tests {
         .await
         .unwrap();
 
-        let addr = conn2.local_addresses().await.unwrap();
+        let addr = conn2.local_addresses();
         assert_eq!(conn1.connect(addr[0].clone()).await, Ok(()));
 
         if let Ok(ConnectivityEvent::OutboundAccepted {
@@ -406,7 +403,7 @@ mod tests {
             receiver_address: _,
         }) = conn1.poll_next().await
         {
-            assert_eq!(address, conn2.local_addresses().await.unwrap()[0]);
+            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);
@@ -460,7 +457,7 @@ mod tests {
         .await
         .unwrap();
 
-        let bind_address = conn2.local_addresses().await.unwrap();
+        let bind_address = conn2.local_addresses();
         let (_res1, res2) = tokio::join!(conn1.connect(bind_address[0].clone()), conn2.poll_next());
         match res2.unwrap() {
             ConnectivityEvent::InboundAccepted {
@@ -520,7 +517,7 @@ mod tests {
         .unwrap();
 
         let (_res1, res2) = tokio::join!(
-            conn1.connect(conn2.local_addresses().await.unwrap()[0].clone()),
+            conn1.connect(conn2.local_addresses()[0].clone()),
             conn2.poll_next()
         );
 
@@ -578,12 +575,12 @@ mod tests {
         .unwrap();
 
         // Try connect to self
-        let addr = conn1.local_addresses().await.unwrap();
+        let addr = conn1.local_addresses();
         assert_eq!(conn1.connect(addr[0].clone()).await, Ok(()));
 
         // ConnectionError should be reported
         if let Ok(ConnectivityEvent::ConnectionError { address, error }) = conn1.poll_next().await {
-            assert_eq!(address, conn1.local_addresses().await.unwrap()[0]);
+            assert_eq!(address, conn1.local_addresses()[0]);
             assert_eq!(error, P2pError::DialError(DialError::AttemptToDialSelf));
         } else {
             panic!("invalid event received");
@@ -600,7 +597,7 @@ mod tests {
         }
 
         // Check that we can still connect normally after
-        let addr = conn2.local_addresses().await.unwrap();
+        let addr = conn2.local_addresses();
         assert_eq!(conn1.connect(addr[0].clone()).await, Ok(()));
         if let Ok(ConnectivityEvent::OutboundAccepted {
             address,
@@ -608,7 +605,7 @@ mod tests {
             receiver_address: _,
         }) = conn1.poll_next().await
         {
-            assert_eq!(address, conn2.local_addresses().await.unwrap()[0]);
+            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);
diff --git a/p2p/src/net/default_backend/transport/impls/channel.rs b/p2p/src/net/default_backend/transport/impls/channel.rs
index 8f8acb00d..bcb646d1e 100644
--- a/p2p/src/net/default_backend/transport/impls/channel.rs
+++ b/p2p/src/net/default_backend/transport/impls/channel.rs
@@ -56,8 +56,11 @@ impl TransportAddress for Address {
         })
     }
 
-    fn from_peer_address(_address: &PeerAddress) -> Option<Self> {
-        None
+    fn from_peer_address(address: &PeerAddress) -> Option<Self> {
+        match address {
+            PeerAddress::Ip4(socket) => Some(std::net::Ipv4Addr::from(socket.ip).into()),
+            PeerAddress::Ip6(_) => None,
+        }
     }
 }
 
diff --git a/p2p/src/net/default_backend/transport/traits/socket.rs b/p2p/src/net/default_backend/transport/traits/socket.rs
index 4523b9d00..ee55d57fc 100644
--- a/p2p/src/net/default_backend/transport/traits/socket.rs
+++ b/p2p/src/net/default_backend/transport/traits/socket.rs
@@ -33,6 +33,7 @@ pub trait TransportSocket: Send + Sync + 'static {
         + Clone
         + Debug
         + Eq
+        + Ord
         + Hash
         + Send
         + Sync
diff --git a/p2p/src/net/default_backend/types.rs b/p2p/src/net/default_backend/types.rs
index 425da8b19..29797dbec 100644
--- a/p2p/src/net/default_backend/types.rs
+++ b/p2p/src/net/default_backend/types.rs
@@ -111,9 +111,6 @@ pub enum ConnectivityEvent<T: TransportSocket> {
         peer_id: PeerId,
         error: error::P2pError,
     },
-    AddressDiscovered {
-        address: T::Address,
-    },
 }
 
 // TODO: use two events, one for txs and one for blocks?
@@ -144,7 +141,7 @@ impl std::fmt::Display for RequestId {
     }
 }
 
-#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug, Encode, Decode)]
+#[derive(Copy, Clone, PartialEq, Eq, Ord, PartialOrd, Hash, Debug, Encode, Decode)]
 pub struct PeerId(u64);
 
 impl FromStr for PeerId {
diff --git a/p2p/src/net/mod.rs b/p2p/src/net/mod.rs
index 3a27f44ec..eb2257cad 100644
--- a/p2p/src/net/mod.rs
+++ b/p2p/src/net/mod.rs
@@ -53,6 +53,7 @@ pub trait NetworkingService {
     type Address: Clone
         + Debug
         + Eq
+        + Ord
         + Hash
         + Send
         + Sync
@@ -68,7 +69,7 @@ pub trait NetworkingService {
     type BannableAddress: Debug + Eq + Ord + Send;
 
     /// Unique ID assigned to a peer on the network
-    type PeerId: Copy + Debug + Display + Eq + Hash + Send + Sync + ToString + FromStr;
+    type PeerId: Copy + Debug + Display + Eq + Ord + Hash + Send + Sync + ToString + FromStr;
 
     /// Unique ID assigned to each received request from a peer
     type PeerRequestId: Copy + Debug + Eq + Hash + Send + Sync;
@@ -137,10 +138,8 @@ where
         response: PeerManagerResponse,
     ) -> crate::Result<()>;
 
-    /// Return the socket address of the network service provider
-    ///
-    /// If the address isn't available yet, `None` is returned
-    async fn local_addresses(&self) -> crate::Result<Vec<T::Address>>;
+    /// Return the socket addresses of the network service provider
+    fn local_addresses(&self) -> &[T::Address];
 
     /// Poll events from the network service provider
     ///
diff --git a/p2p/src/net/types/mod.rs b/p2p/src/net/types/mod.rs
index d7780ef6c..34c5a7e4a 100644
--- a/p2p/src/net/types/mod.rs
+++ b/p2p/src/net/types/mod.rs
@@ -136,12 +136,6 @@ pub enum ConnectivityEvent<T: NetworkingService> {
         peer_id: T::PeerId,
     },
 
-    /// New peer discovered
-    AddressDiscovered {
-        /// Address information
-        address: T::Address,
-    },
-
     /// Protocol violation
     Misbehaved {
         /// Unique ID of the peer
diff --git a/p2p/src/peer_manager/global_ip.rs b/p2p/src/peer_manager/global_ip.rs
new file mode 100644
index 000000000..09f138a9a
--- /dev/null
+++ b/p2p/src/peer_manager/global_ip.rs
@@ -0,0 +1,146 @@
+// Copyright 2021 Protocol Labs.
+// 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.
+
+// Based on libp2p sources:
+// https://github.com/libp2p/rust-libp2p/blob/73cbbe29679f5d56d81d5477d3796e82712f9ac2/protocols/autonat/src/behaviour.rs
+
+pub trait IsGlobalIp {
+    fn is_global_unicast_ip(&self) -> bool;
+}
+
+impl IsGlobalIp for std::net::Ipv4Addr {
+    // NOTE: The below logic is based on `std::net::Ipv4Addr::is_global`,
+    // which is at the time of writing behind the unstable `ip` feature.
+    // See https://github.com/rust-lang/rust/issues/27709 for more info.
+    fn is_global_unicast_ip(&self) -> bool {
+        // Copied from the unstable method `std::net::Ipv4Addr::is_shared`.
+        fn is_shared(addr: &std::net::Ipv4Addr) -> bool {
+            addr.octets()[0] == 100 && (addr.octets()[1] & 0b1100_0000 == 0b0100_0000)
+        }
+
+        // Copied from the unstable method `std::net::Ipv4Addr::is_reserved`.
+        fn is_reserved(addr: &std::net::Ipv4Addr) -> bool {
+            addr.octets()[0] & 240 == 240 && !addr.is_broadcast()
+        }
+
+        // Copied from the unstable method `std::net::Ipv4Addr::is_benchmarking`.
+        fn is_benchmarking(addr: &std::net::Ipv4Addr) -> bool {
+            addr.octets()[0] == 198 && (addr.octets()[1] & 0xfe) == 18
+        }
+
+        // Addresses reserved for future protocols (`192.0.0.0/24`)
+        fn is_future_protocol(addr: &std::net::Ipv4Addr) -> bool {
+            addr.octets()[0] == 192 && addr.octets()[1] == 0 && addr.octets()[2] == 0
+        }
+
+        // The 0.0.0.0/8 block is reserved
+        fn is_reserved2(addr: &std::net::Ipv4Addr) -> bool {
+            addr.octets()[0] == 0
+        }
+
+        !self.is_multicast()
+            && !self.is_private()
+            && !self.is_loopback()
+            && !self.is_link_local()
+            && !self.is_broadcast()
+            && !self.is_documentation()
+            && !is_shared(self)
+            && !is_future_protocol(self)
+            && !is_reserved(self)
+            && !is_benchmarking(self)
+            && !is_reserved2(self)
+    }
+}
+
+impl IsGlobalIp for std::net::Ipv6Addr {
+    // NOTE: The below logic is based on `std::net::Ipv6Addr::is_global`,
+    // which is at the time of writing behind the unstable `ip` feature.
+    // See https://github.com/rust-lang/rust/issues/27709 for more info.
+    fn is_global_unicast_ip(&self) -> bool {
+        // Copied from the unstable method `std::net::Ipv6Addr::is_unicast_link_local`.
+        fn is_unicast_link_local(addr: &std::net::Ipv6Addr) -> bool {
+            (addr.segments()[0] & 0xffc0) == 0xfe80
+        }
+        // Copied from the unstable method `std::net::Ipv6Addr::is_unique_local`.
+        fn is_unique_local(addr: &std::net::Ipv6Addr) -> bool {
+            (addr.segments()[0] & 0xfe00) == 0xfc00
+        }
+        // Copied from the unstable method `std::net::Ipv6Addr::is_documentation`.
+        fn is_documentation(addr: &std::net::Ipv6Addr) -> bool {
+            (addr.segments()[0] == 0x2001) && (addr.segments()[1] == 0xdb8)
+        }
+
+        // Copied from the unstable method `std::net::Ipv6Addr::is_unicast_global`.
+        !self.is_multicast()
+            && !self.is_loopback()
+            && !self.is_unspecified()
+            && !is_unicast_link_local(self)
+            && !is_unique_local(self)
+            && !is_documentation(self)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::IsGlobalIp;
+
+    fn is_global_unicast(ip: &str) -> bool {
+        match ip.parse::<std::net::IpAddr>().unwrap() {
+            std::net::IpAddr::V4(ip) => ip.is_global_unicast_ip(),
+            std::net::IpAddr::V6(ip) => ip.is_global_unicast_ip(),
+        }
+    }
+
+    #[test]
+    fn test_addresses() {
+        let global_unicast_ips = ["142.250.184.142", "2a00:1450:4017:815::200e"];
+
+        let non_global_unicast_ips = [
+            "0.0.0.0",             // Unspecified
+            "255.255.255.255",     // Broadcast
+            "127.0.0.0",           // Local
+            "127.255.255.255",     // Local
+            "10.0.0.0",            // Private
+            "10.255.255.255",      // Private
+            "100.64.0.0",          // Private
+            "100.127.255.255",     // Private
+            "172.16.0.0",          // Private
+            "172.31.255.255",      // Private
+            "192.168.0.0",         // Private
+            "192.168.255.255",     // Private
+            "169.254.1.0",         // Link-local
+            "224.0.0.0",           // Multicast
+            "239.255.255.255",     // Multicast
+            "::",                  // Unspecified
+            "::1",                 // Local
+            "fd12:3456:789a:1::1", // Private
+            "ff02::1",             // Multicast
+        ];
+
+        for ip in global_unicast_ips {
+            assert!(
+                is_global_unicast(ip),
+                "{ip} is expected to be global unicast IP address"
+            );
+        }
+        for ip in non_global_unicast_ips {
+            assert!(
+                !is_global_unicast(ip),
+                "{ip} is expected to be not global unicast IP address"
+            );
+        }
+    }
+}
diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs
index a1865bff2..67d1abe89 100644
--- a/p2p/src/peer_manager/mod.rs
+++ b/p2p/src/peer_manager/mod.rs
@@ -21,12 +21,13 @@
 
 #![allow(rustdoc::private_intra_doc_links)]
 
+mod global_ip;
 pub mod peerdb;
 
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     sync::Arc,
-    time::{Duration, Instant},
+    time::Instant,
 };
 
 use tokio::sync::{mpsc, oneshot};
@@ -40,23 +41,30 @@ use crate::{
     config::P2pConfig,
     error::{P2pError, PeerError, ProtocolError},
     event::{PeerManagerEvent, SyncControlEvent},
-    message::{AddrListRequest, AddrListResponse, PeerManagerRequest, PeerManagerResponse},
+    message::{
+        AddrListRequest, AddrListResponse, AnnounceAddrRequest, AnnounceAddrResponse,
+        PeerManagerRequest, PeerManagerResponse,
+    },
     net::{
         self,
         default_backend::transport::TransportAddress,
-        types::{PeerInfo, Role},
+        types::PeerInfo,
+        types::{ConnectivityEvent, Role},
         AsBannableAddress, ConnectivityService, NetworkingService,
     },
-    types::peer_address::PeerAddress,
+    types::peer_address::{PeerAddress, PeerAddressIp4, PeerAddressIp6},
 };
 
+use self::global_ip::IsGlobalIp;
+
 /// Maximum number of connections the [`PeerManager`] is allowed to have open
 const MAX_ACTIVE_CONNECTIONS: usize = 128;
 
-/// Lower bound for how often [`PeerManager::heartbeat()`] is called
-const PEER_MGR_HEARTBEAT_INTERVAL_MIN: Duration = Duration::from_secs(5);
-/// Upper bound for how often [`PeerManager::heartbeat()`] is called
-const PEER_MGR_HEARTBEAT_INTERVAL_MAX: Duration = Duration::from_secs(30);
+/// How many addresses are allowed to be sent
+const MAX_ADDRESS_COUNT: usize = 1000;
+
+/// To how many peers re-send received announced address
+const ANNOUNCED_RESEND_COUNT: usize = 2;
 
 pub struct PeerManager<T>
 where
@@ -66,7 +74,7 @@ where
     chain_config: Arc<ChainConfig>,
 
     /// P2P configuration.
-    _p2p_config: Arc<P2pConfig>,
+    p2p_config: Arc<P2pConfig>,
 
     /// Handle for sending/receiving connectivity events
     peer_connectivity_handle: T::ConnectivityHandle,
@@ -83,7 +91,13 @@ where
     /// Peer database
     peerdb: peerdb::PeerDb<T>,
 
+    /// Last time when heartbeat was called
     last_heartbeat: Instant,
+
+    /// 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).
+    announced_addresses: HashMap<T::PeerId, HashSet<T::Address>>,
 }
 
 impl<T> PeerManager<T>
@@ -105,16 +119,12 @@ where
             peerdb: peerdb::PeerDb::new(Arc::clone(&p2p_config))?,
             pending: HashMap::new(),
             chain_config,
-            _p2p_config: p2p_config,
+            p2p_config,
             last_heartbeat: Instant::now(),
+            announced_addresses: HashMap::new(),
         })
     }
 
-    /// Update the list of known peers or known peer's list of addresses
-    fn peer_discovered(&mut self, address: &T::Address) {
-        self.peerdb.peer_discovered(address);
-    }
-
     /// Verify software version compatibility
     ///
     /// Make sure that local and remote peer have the same software version
@@ -123,20 +133,99 @@ where
         version == self.chain_config.version()
     }
 
+    fn is_peer_address_valid(&self, address: &PeerAddress) -> bool {
+        // The IP must be globally routable
+        match &address {
+            PeerAddress::Ip4(socket) => {
+                std::net::Ipv4Addr::from(socket.ip).is_global_unicast_ip()
+                    || *self.p2p_config.allow_discover_private_ips
+            }
+            PeerAddress::Ip6(socket) => {
+                std::net::Ipv6Addr::from(socket.ip).is_global_unicast_ip()
+                    || *self.p2p_config.allow_discover_private_ips
+            }
+        }
+    }
+
+    /// Discover public addresses for this node after a new outbound connection is made
+    ///
+    /// *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.
+    async fn handle_outbound_receiver_address(
+        &mut self,
+        peer_id: T::PeerId,
+        receiver_address: PeerAddress,
+    ) -> crate::Result<()> {
+        if !self.is_peer_address_valid(&receiver_address) {
+            return Ok(());
+        }
+
+        // Take IP and use port numbers from all listening sockets (with same IP version)
+        let discovered_own_addresses = self
+            .peer_connectivity_handle
+            .local_addresses()
+            .iter()
+            .map(TransportAddress::as_peer_address)
+            .filter_map(
+                |listening_address| match (&receiver_address, listening_address) {
+                    (PeerAddress::Ip4(receiver), PeerAddress::Ip4(listener)) => {
+                        Some(PeerAddress::Ip4(PeerAddressIp4 {
+                            ip: receiver.ip,
+                            port: listener.port,
+                        }))
+                    }
+                    (PeerAddress::Ip6(receiver), PeerAddress::Ip6(listener)) => {
+                        Some(PeerAddress::Ip6(PeerAddressIp6 {
+                            ip: receiver.ip,
+                            port: listener.port,
+                        }))
+                    }
+                    _ => None,
+                },
+            )
+            .filter_map(|address| TransportAddress::from_peer_address(&address))
+            .collect::<Vec<_>>();
+
+        for address in discovered_own_addresses {
+            self.send_announced_address(peer_id, address).await?;
+        }
+
+        Ok(())
+    }
+
+    async fn send_announced_address(
+        &mut self,
+        peer_id: T::PeerId,
+        address: T::Address,
+    ) -> crate::Result<()> {
+        let peer_addresses = self.announced_addresses.entry(peer_id).or_default();
+        if !peer_addresses.contains(&address) {
+            self.peer_connectivity_handle
+                .send_request(
+                    peer_id,
+                    PeerManagerRequest::AnnounceAddrRequest(AnnounceAddrRequest {
+                        address: address.as_peer_address(),
+                    }),
+                )
+                .await?;
+            peer_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(
+    async fn accept_connection(
         &mut self,
         address: T::Address,
         role: Role,
         info: PeerInfo<T::PeerId>,
-        _receiver_address: Option<PeerAddress>,
+        receiver_address: Option<PeerAddress>,
     ) -> crate::Result<()> {
-        // TODO: Handle receiver_address
-
         let peer_id = info.peer_id;
 
         ensure!(
@@ -162,6 +251,19 @@ where
             P2pError::PeerError(PeerError::PeerAlreadyExists),
         );
 
+        if let (Some(receiver_address), Role::Outbound) = (receiver_address, role) {
+            self.handle_outbound_receiver_address(peer_id, receiver_address).await?;
+        }
+
+        if role == Role::Outbound {
+            self.peer_connectivity_handle
+                .send_request(
+                    peer_id,
+                    PeerManagerRequest::AddrListRequest(AddrListRequest {}),
+                )
+                .await?;
+        }
+
         log::debug!("peer {peer_id} connected, address {address:?}, {info}");
         self.peerdb.peer_connected(address, role, info);
         self.tx_sync.send(SyncControlEvent::Connected(peer_id)).map_err(P2pError::from)
@@ -176,7 +278,7 @@ where
     /// 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(
+    async fn accept_inbound_connection(
         &mut self,
         address: T::Address,
         info: net::types::PeerInfo<T::PeerId>,
@@ -203,7 +305,7 @@ where
             return Err(P2pError::PeerError(PeerError::TooManyPeers));
         }
 
-        self.accept_connection(address, Role::Inbound, info, receiver_address)
+        self.accept_connection(address, Role::Inbound, info, receiver_address).await
     }
 
     /// Close connection to a remote peer
@@ -220,6 +322,9 @@ where
 
             self.peerdb.peer_disconnected(&peer_id);
         }
+
+        self.announced_addresses.remove(&peer_id);
+
         Ok(())
     }
 
@@ -261,12 +366,16 @@ where
     /// networking backend which then reports at some point in the future
     /// whether the connection failed or succeeded.
     async fn connect(&mut self, address: T::Address) -> crate::Result<()> {
-        // TODO: verify that the peer is not already part of our swarm (needs peerdb)
         ensure!(
             !self.pending.contains_key(&address),
             P2pError::PeerError(PeerError::Pending(address.to_string())),
         );
 
+        ensure!(
+            !self.peerdb.is_address_connected(&address),
+            P2pError::PeerError(PeerError::PeerAlreadyExists),
+        );
+
         let bannable_address = address.as_bannable();
         ensure!(
             !self.peerdb.is_address_banned(&bannable_address),
@@ -283,7 +392,7 @@ where
     /// is checked and updated at least once every 30 seconds. In high-traffic scenarios the
     /// update interval is clamped to a sensible lower bound. `PeerManager` will keep track of
     /// when it last update its own state and if the time since last update is less than the
-    /// configured lower bound, it exits early from the function.
+    /// configured lower bound, *heartbeat* won't be called.
     ///
     /// This function maintains the overall connectivity state of peers by culling
     /// low-reputation peers and establishing new connections with peers that have higher
@@ -298,23 +407,23 @@ where
     /// establish new connections. After that it updates the peer scores and discards any records
     /// that no longer need to be stored.
     async fn heartbeat(&mut self) -> crate::Result<()> {
-        let npeers = std::cmp::min(
+        let count = std::cmp::min(
             self.peerdb.available_addresses_count(),
             MAX_ACTIVE_CONNECTIONS
                 .saturating_sub(self.peerdb.available_addresses_count())
                 .saturating_sub(self.pending.len()),
         );
 
-        for _ in 0..npeers {
-            if let Some(addr) = self.peerdb.take_best_peer_addr() {
-                match self.connect(addr.clone()).await {
-                    Ok(_) => {
-                        self.pending.insert(addr, None);
-                    }
-                    Err(err) => {
-                        self.peerdb.report_outbound_failure(addr.clone());
-                        self.handle_result(None, Err(err)).await?;
-                    }
+        let addresses = self.peerdb.random_known_addresses(count);
+
+        for address in addresses {
+            match self.connect(address.clone()).await {
+                Ok(_) => {
+                    self.pending.insert(address, None);
+                }
+                Err(err) => {
+                    self.peerdb.report_outbound_failure(address.clone());
+                    self.handle_result(None, Err(err)).await?;
                 }
             }
         }
@@ -326,22 +435,46 @@ where
 
     async fn handle_incoming_request(
         &mut self,
-        _peer_id: T::PeerId,
+        peer_id: T::PeerId,
         request_id: T::PeerRequestId,
         request: PeerManagerRequest,
     ) -> crate::Result<()> {
         match request {
-            // TODO: Rework this
             PeerManagerRequest::AddrListRequest(AddrListRequest {}) => {
-                let addresses = self.peerdb.known_addresses();
+                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_response(
                         request_id,
-                        PeerManagerResponse::AddrListResponse(AddrListResponse::new(addresses)),
+                        PeerManagerResponse::AddrListResponse(AddrListResponse { addresses }),
                     )
                     .await
             }
+            PeerManagerRequest::AnnounceAddrRequest(AnnounceAddrRequest { address }) => {
+                // TODO: Rate limit announce address requests to prevent DoS attacks.
+                // For example it's 0.1 req/sec in Bitcoin Core.
+                let is_address_valid = self.is_peer_address_valid(&address);
+                if let (true, Some(address)) = (
+                    is_address_valid,
+                    TransportAddress::from_peer_address(&address),
+                ) {
+                    self.peerdb.peer_discovered(&address);
+
+                    self.announced_addresses.entry(peer_id).or_default().insert(address.clone());
+
+                    let peer_ids = self.peerdb.random_peer_ids(ANNOUNCED_RESEND_COUNT);
+                    for new_peer_id in peer_ids {
+                        self.send_announced_address(new_peer_id, address.clone()).await?;
+                    }
+                }
+                Ok(())
+            }
         }
     }
 
@@ -351,16 +484,20 @@ where
         _request_id: T::PeerRequestId,
         response: PeerManagerResponse,
     ) -> crate::Result<()> {
+        // TODO: Check that unsolicited responses are not allowed
         match response {
-            // TODO: Rework this
-            PeerManagerResponse::AddrListResponse(response) => {
-                for address in response.addresses() {
-                    if let Some(address) = TransportAddress::from_peer_address(address) {
+            PeerManagerResponse::AddrListResponse(AddrListResponse { addresses }) => {
+                for address in addresses {
+                    if let (true, Some(address)) = (
+                        self.is_peer_address_valid(&address),
+                        TransportAddress::from_peer_address(&address),
+                    ) {
                         self.peerdb.peer_discovered(&address);
                     }
                 }
                 Ok(())
             }
+            PeerManagerResponse::AnnounceAddrResponse(AnnounceAddrResponse {}) => Ok(()),
         }
     }
 
@@ -400,6 +537,147 @@ where
         }
     }
 
+    /// Handle control event.
+    ///
+    /// Handle events from an outside controller (rpc, for example) that sets/gets values for PeerManager
+    async fn handle_control_event(&mut self, event: PeerManagerEvent<T>) -> crate::Result<()> {
+        match event {
+            PeerManagerEvent::Connect(addr, response) => {
+                log::debug!("try to establish outbound connection to peer at address {addr:?}");
+
+                let res = self.connect(addr.clone()).await.map(|_| {
+                    self.pending.insert(addr.clone(), Some(response));
+                });
+                self.handle_result(None, res).await?;
+            }
+            PeerManagerEvent::Disconnect(peer_id, response) => {
+                log::debug!("disconnect peer {peer_id}");
+
+                response
+                    .send(self.peer_connectivity_handle.disconnect(peer_id).await)
+                    .map_err(|_| P2pError::ChannelClosed)?;
+            }
+            PeerManagerEvent::AdjustPeerScore(peer_id, score, response) => {
+                log::debug!("adjust peer {peer_id} score: {score}");
+
+                response
+                    .send(self.adjust_peer_score(peer_id, score).await)
+                    .map_err(|_| P2pError::ChannelClosed)?;
+            }
+            PeerManagerEvent::GetPeerCount(response) => {
+                response
+                    .send(self.peerdb.active_peer_count())
+                    .map_err(|_| P2pError::ChannelClosed)?;
+            }
+            PeerManagerEvent::GetBindAddresses(response) => {
+                let addr = self
+                    .peer_connectivity_handle
+                    .local_addresses()
+                    .iter()
+                    .map(|addr| addr.to_string())
+                    .collect();
+                response.send(addr).map_err(|_| P2pError::ChannelClosed)?;
+            }
+            PeerManagerEvent::GetConnectedPeers(response) => {
+                let peers = self.peerdb.get_connected_peers();
+                response.send(peers).map_err(|_| P2pError::ChannelClosed)?
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Handle connectivity event.
+    async fn handle_connectivity_event_result(
+        &mut self,
+        event_res: crate::Result<ConnectivityEvent<T>>,
+    ) -> crate::Result<()> {
+        match event_res {
+            Ok(event) => match event {
+                net::types::ConnectivityEvent::Request {
+                    peer_id,
+                    request_id,
+                    request,
+                } => {
+                    self.handle_incoming_request(peer_id, request_id, request).await?;
+                }
+                net::types::ConnectivityEvent::Response {
+                    peer_id,
+                    request_id,
+                    response,
+                } => {
+                    self.handle_incoming_response(peer_id, request_id, response)?;
+                }
+                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).await
+                    {
+                        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).await?;
+                        }
+                        Err(P2pError::ProtocolError(err)) => {
+                            log::warn!("peer error for peer {peer_id}: {err}");
+                            self.adjust_peer_score(peer_id, err.ban_score()).await?;
+                        }
+                        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,
+                        )
+                        .await;
+                    self.handle_result(Some(peer_id), res).await?;
+
+                    match self.pending.remove(&address) {
+                        Some(Some(channel)) => {
+                            channel.send(Ok(())).map_err(|_| P2pError::ChannelClosed)?
+                        }
+                        Some(None) => {}
+                        None => log::error!("connection accepted but it's not pending?"),
+                    }
+                }
+                net::types::ConnectivityEvent::ConnectionClosed { peer_id } => {
+                    let res = self.close_connection(peer_id);
+                    self.handle_result(Some(peer_id), res).await?;
+                }
+                net::types::ConnectivityEvent::ConnectionError { address, error } => {
+                    let res = self.handle_outbound_error(address, error);
+                    self.handle_result(None, res).await?;
+                }
+                net::types::ConnectivityEvent::Misbehaved { peer_id, error } => {
+                    let res = self.adjust_peer_score(peer_id, error.ban_score()).await;
+                    self.handle_result(Some(peer_id), res).await?;
+                }
+            },
+            Err(err) => {
+                log::error!("failed to read network event: {err:?}");
+                self.handle_result(None, Err(err)).await?;
+            }
+        }
+
+        Ok(())
+    }
+
     /// Runs the `PeerManager` event loop.
     ///
     /// The event loop has three main responsibilities:
@@ -417,110 +695,18 @@ where
     pub async fn run(&mut self) -> crate::Result<void::Void> {
         loop {
             tokio::select! {
-                event = self.rx_peer_manager.recv() => match event.ok_or(P2pError::ChannelClosed)? {
-                    //
-                    // Handle events from an outside controller (rpc, for example) that sets/gets values for PeerManager
-                    //
-                    PeerManagerEvent::Connect(addr, response) => {
-                        log::debug!("try to establish outbound connection to peer at address {addr:?}");
-
-                        let res = self.connect(addr.clone()).await.map(|_| {
-                            self.pending.insert(addr.clone(), Some(response));
-                        });
-                        self.handle_result(None, res).await?;
-                    }
-                    PeerManagerEvent::Disconnect(peer_id, response) => {
-                        log::debug!("disconnect peer {peer_id}");
-
-                        response
-                            .send(self.peer_connectivity_handle.disconnect(peer_id).await)
-                            .map_err(|_| P2pError::ChannelClosed)?;
-                    }
-                    PeerManagerEvent::AdjustPeerScore(peer_id, score, response) => {
-                        log::debug!("adjust peer {peer_id} score: {score}");
-
-                        response
-                            .send(self.adjust_peer_score(peer_id, score).await)
-                            .map_err(|_| P2pError::ChannelClosed)?;
-                    }
-                    PeerManagerEvent::GetPeerCount(response) => {
-                        response.send(self.peerdb.active_peer_count()).map_err(|_| P2pError::ChannelClosed)?;
-                    }
-                    PeerManagerEvent::GetBindAddresses(response) => {
-                        let addr = self.peer_connectivity_handle.local_addresses();
-                        let addr = addr.await.unwrap_or_default().into_iter().map(|addr| addr.to_string()).collect();
-                        response.send(addr).map_err(|_| P2pError::ChannelClosed)?;
-                    }
-                    PeerManagerEvent::GetConnectedPeers(response) => {
-                        let peers = self.peerdb.get_connected_peers();
-                        response.send(peers).map_err(|_| P2pError::ChannelClosed)?
-                    }
+                event = self.rx_peer_manager.recv() => {
+                    self.handle_control_event(event.ok_or(P2pError::ChannelClosed)?).await?;
                 },
-                event = self.peer_connectivity_handle.poll_next() => match event {
-                    Ok(event) => match event {
-                        net::types::ConnectivityEvent::Request { peer_id, request_id, request } => {
-                            self.handle_incoming_request(peer_id, request_id, request).await?;
-                        },
-                        net::types::ConnectivityEvent::Response { peer_id, request_id, response } => {
-                            self.handle_incoming_response(peer_id, request_id, response)?;
-                        },
-                        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).await?;
-                                }
-                                Err(P2pError::ProtocolError(err)) => {
-                                    log::warn!("peer error for peer {peer_id}: {err}");
-                                    self.adjust_peer_score(peer_id, err.ban_score()).await?;
-                                }
-                                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).await?;
-
-                            match self.pending.remove(&address) {
-                                Some(Some(channel)) => channel.send(Ok(())).map_err(|_| P2pError::ChannelClosed)?,
-                                Some(None) => {},
-                                None => log::error!("connection accepted but it's not pending?"),
-                            }
-                        }
-                        net::types::ConnectivityEvent::ConnectionClosed { peer_id } => {
-                            let res = self.close_connection(peer_id);
-                            self.handle_result(Some(peer_id), res).await?;
-                        }
-                        net::types::ConnectivityEvent::ConnectionError { address, error } => {
-                            let res = self.handle_outbound_error(address, error);
-                            self.handle_result(None, res).await?;
-                        }
-                        net::types::ConnectivityEvent::AddressDiscovered { address } => {
-                            self.peer_discovered(&address);
-                        }
-                        net::types::ConnectivityEvent::Misbehaved { peer_id, error } => {
-                            let res = self.adjust_peer_score(peer_id, error.ban_score()).await;
-                            self.handle_result(Some(peer_id), res).await?;
-                        }
-                    }
-                    Err(err) => {
-                        log::error!("failed to read network event: {err:?}");
-                        self.handle_result(None, Err(err)).await?;
-                    }
+                event_res = self.peer_connectivity_handle.poll_next() => {
+                    self.handle_connectivity_event_result(event_res).await?;
                 },
-                _event = tokio::time::sleep(PEER_MGR_HEARTBEAT_INTERVAL_MAX) => {}
+                _event = tokio::time::sleep(*self.p2p_config.heartbeat_interval_max) => {}
             }
 
             // finally update peer manager state
             let now = Instant::now();
-            if now.duration_since(self.last_heartbeat) > PEER_MGR_HEARTBEAT_INTERVAL_MIN {
+            if now.duration_since(self.last_heartbeat) > *self.p2p_config.heartbeat_interval_min {
                 self.heartbeat().await?;
                 self.last_heartbeat = now;
             }
diff --git a/p2p/src/peer_manager/peerdb/mod.rs b/p2p/src/peer_manager/peerdb/mod.rs
index 7390c85b6..e8cbb6928 100644
--- a/p2p/src/peer_manager/peerdb/mod.rs
+++ b/p2p/src/peer_manager/peerdb/mod.rs
@@ -26,11 +26,12 @@
 //! if the actual number of active connection is less than the desired number of connections.
 
 use std::{
-    collections::{BTreeMap, HashMap, HashSet},
+    collections::{BTreeMap, BTreeSet},
     sync::Arc,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
+use crypto::random::{make_pseudo_rng, SliceRandom};
 use logging::log;
 
 use crate::{
@@ -38,11 +39,9 @@ use crate::{
     error::{ConversionError, P2pError},
     interface::types::ConnectedPeer,
     net::{
-        default_backend::transport::TransportAddress,
         types::{self, Role},
         AsBannableAddress, NetworkingService,
     },
-    types::peer_address::PeerAddress,
 };
 
 #[derive(Debug)]
@@ -79,12 +78,13 @@ pub struct PeerDb<T: NetworkingService> {
     p2p_config: Arc<config::P2pConfig>,
 
     /// Set of active peers
-    peers: HashMap<T::PeerId, PeerContext<T>>,
+    peers: BTreeMap<T::PeerId, PeerContext<T>>,
 
-    addresses: HashSet<T::Address>,
+    /// Set of currently connected addresses
+    connected_addresses: BTreeSet<T::Address>,
 
-    /// Set of available addresses
-    available: HashSet<T::Address>,
+    /// Set of all known addresses
+    known_addresses: BTreeSet<T::Address>,
 
     /// Banned addresses along with the duration of the ban.
     ///
@@ -103,13 +103,13 @@ impl<T: NetworkingService> PeerDb<T> {
                     P2pError::ConversionError(ConversionError::InvalidAddress(addr.clone()))
                 })
             })
-            .collect::<Result<HashSet<_>, _>>()?;
+            .collect::<Result<BTreeSet<_>, _>>()?;
         Ok(Self {
             peers: Default::default(),
-            addresses: Default::default(),
+            connected_addresses: Default::default(),
             // 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.
-            available: added_nodes,
+            known_addresses: added_nodes,
             banned: Default::default(),
             p2p_config,
         })
@@ -117,7 +117,7 @@ impl<T: NetworkingService> PeerDb<T> {
 
     /// Get the number of idle (available) addresses
     pub fn available_addresses_count(&self) -> usize {
-        self.available.len()
+        self.known_addresses.len()
     }
 
     /// Get the number of active peers
@@ -125,17 +125,39 @@ impl<T: NetworkingService> PeerDb<T> {
         self.peers.len()
     }
 
+    /// Returns short info about all connected peers
     pub fn get_connected_peers(&self) -> Vec<ConnectedPeer> {
         self.peers.values().map(Into::into).collect()
     }
 
     /// Checks if the given address is already connected.
     pub fn is_address_connected(&self, address: &T::Address) -> bool {
-        self.addresses.contains(address)
+        self.connected_addresses.contains(address)
     }
 
-    pub fn known_addresses(&self) -> Vec<PeerAddress> {
-        self.addresses.iter().map(TransportAddress::as_peer_address).collect()
+    /// Selects requested count of peer addresses from the DB randomly.
+    ///
+    /// 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<_>>()
+    }
+
+    /// Selects requested count of connected peer ids randomly.
+    ///
+    /// It can be used to distribute data in the gossip protocol
+    /// (for example, to relay announced addresses to a small group of peers).
+    pub fn random_peer_ids(&self, count: usize) -> Vec<T::PeerId> {
+        // There are normally not many connected peers, so iterating over the whole list should be OK
+        let all_peer_ids = self.peers.keys().cloned().collect::<Vec<_>>();
+        all_peer_ids
+            .choose_multiple(&mut make_pseudo_rng(), count)
+            .cloned()
+            .collect::<Vec<_>>()
     }
 
     /// Checks if the given address is banned.
@@ -163,17 +185,13 @@ impl<T: NetworkingService> PeerDb<T> {
 
     /// Get socket address of the next best peer (TODO: in terms of peer score).
     // TODO: Rewrite this.
-    pub fn take_best_peer_addr(&mut self) -> Option<T::Address> {
-        let address = self.available.iter().next().cloned();
-        if let Some(address) = &address {
-            self.available.remove(address);
-        }
-        address
+    pub fn get_best_peer_addr(&mut self) -> Option<T::Address> {
+        self.random_known_addresses(1).into_iter().next()
     }
 
     /// Add new peer addresses
     pub fn peer_discovered(&mut self, address: &T::Address) {
-        self.available.insert(address.clone());
+        self.known_addresses.insert(address.clone());
     }
 
     /// Report outbound connection failure
@@ -211,7 +229,7 @@ impl<T: NetworkingService> PeerDb<T> {
         );
         assert!(old_value.is_none());
 
-        let old_value = self.addresses.insert(address);
+        let old_value = self.connected_addresses.insert(address);
         assert!(old_value);
     }
 
@@ -222,7 +240,7 @@ impl<T: NetworkingService> PeerDb<T> {
         let removed = self.peers.remove(peer_id);
         let peer = removed.expect("peer must be known");
 
-        let removed = self.addresses.remove(&peer.address);
+        let removed = self.connected_addresses.remove(&peer.address);
         assert!(removed);
 
         log::info!(


Test patch:
diff --git a/p2p/backend-test-suite/src/block_announcement.rs b/p2p/backend-test-suite/src/block_announcement.rs
index 665e5d0f5..42d94aca8 100644
--- a/p2p/backend-test-suite/src/block_announcement.rs
+++ b/p2p/backend-test-suite/src/block_announcement.rs
@@ -136,6 +136,9 @@ where
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
         node_type: NodeType::Inactive.into(),
+        allow_discover_private_ips: Default::default(),
+        heartbeat_interval_min: Default::default(),
+        heartbeat_interval_max: Default::default(),
     });
     let (mut conn1, mut sync1) = S::start(
         T::make_transport(),
diff --git a/p2p/backend-test-suite/src/connect.rs b/p2p/backend-test-suite/src/connect.rs
index a398de46b..a763a7b12 100644
--- a/p2p/backend-test-suite/src/connect.rs
+++ b/p2p/backend-test-suite/src/connect.rs
@@ -61,7 +61,7 @@ where
     .await
     .unwrap();
 
-    let addresses = connectivity.local_addresses().await.unwrap();
+    let addresses = connectivity.local_addresses().to_vec();
     let res = S::start(T::make_transport(), addresses, config, Default::default())
         .await
         .expect_err("address is not in use");
@@ -100,7 +100,7 @@ where
     .await
     .unwrap();
 
-    let conn_addr = service1.local_addresses().await.unwrap();
+    let conn_addr = service1.local_addresses().to_vec();
     let (res1, res2) = tokio::join!(service1.poll_next(), service2.connect(conn_addr[0].clone()));
     res1.unwrap();
     res2.unwrap();
diff --git a/p2p/backend-test-suite/src/peer_manager_peerdb.rs b/p2p/backend-test-suite/src/peer_manager_peerdb.rs
index 3c889f23b..7bcfcd636 100644
--- a/p2p/backend-test-suite/src/peer_manager_peerdb.rs
+++ b/p2p/backend-test-suite/src/peer_manager_peerdb.rs
@@ -61,6 +61,9 @@ where
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
         node_type: Default::default(),
+        allow_discover_private_ips: Default::default(),
+        heartbeat_interval_min: Default::default(),
+        heartbeat_interval_max: Default::default(),
     };
     let mut peerdb = PeerDb::<S>::new(Arc::new(config)).unwrap();
 
@@ -83,6 +86,9 @@ where
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
         node_type: Default::default(),
+        allow_discover_private_ips: Default::default(),
+        heartbeat_interval_min: Default::default(),
+        heartbeat_interval_max: Default::default(),
     };
     let mut peerdb = PeerDb::<S>::new(Arc::new(config)).unwrap();
 
@@ -104,6 +110,9 @@ where
         ban_duration: Duration::from_secs(2).into(),
         outbound_connection_timeout: Default::default(),
         node_type: Default::default(),
+        allow_discover_private_ips: Default::default(),
+        heartbeat_interval_min: Default::default(),
+        heartbeat_interval_max: Default::default(),
     }))
     .unwrap();
 
diff --git a/p2p/backend-test-suite/src/sync.rs b/p2p/backend-test-suite/src/sync.rs
index ae0fea7f2..e640967d2 100644
--- a/p2p/backend-test-suite/src/sync.rs
+++ b/p2p/backend-test-suite/src/sync.rs
@@ -27,7 +27,7 @@ use common::{
     primitives::{Id, Idable},
 };
 
-use p2p::testing_utils::{connect_services, filter_connectivity_event, TestTransportMaker};
+use p2p::testing_utils::{connect_services, get_connectivity_event, TestTransportMaker};
 use p2p::{
     config::P2pConfig,
     error::P2pError,
@@ -1063,10 +1063,7 @@ where
     mgr1.unregister_peer(peer_info2.peer_id);
     assert_eq!(conn1.disconnect(peer_info2.peer_id).await, Ok(()));
 
-    let event = filter_connectivity_event::<S, _>(&mut conn2, |event| {
-        !std::matches!(event, Ok(ConnectivityEvent::AddressDiscovered { .. }))
-    })
-    .await;
+    let event = get_connectivity_event::<S>(&mut conn2).await;
     assert!(std::matches!(
         event,
         Ok(ConnectivityEvent::ConnectionClosed { .. })
diff --git a/p2p/src/peer_manager/tests/ban.rs b/p2p/src/peer_manager/tests/ban.rs
index 6b2d52a3b..0dc901cc5 100644
--- a/p2p/src/peer_manager/tests/ban.rs
+++ b/p2p/src/peer_manager/tests/ban.rs
@@ -18,7 +18,7 @@ use std::sync::Arc;
 use crate::{
     net::types::Role,
     testing_utils::{
-        connect_services, filter_connectivity_event, RandomAddressMaker, TestChannelAddressMaker,
+        connect_services, get_connectivity_event, RandomAddressMaker, TestChannelAddressMaker,
         TestTcpAddressMaker, TestTransportChannel, TestTransportMaker, TestTransportNoise,
         TestTransportTcp,
     },
@@ -60,20 +60,12 @@ where
     )
     .await;
     let peer_id = peer_info.peer_id;
-    pm2.accept_inbound_connection(address, peer_info, None).unwrap();
+    pm2.accept_inbound_connection(address, peer_info, None).await.unwrap();
 
     assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
-    let addr1 = pm1.peer_connectivity_handle.local_addresses().await.unwrap()[0]
-        .clone()
-        .as_bannable();
+    let addr1 = pm1.peer_connectivity_handle.local_addresses()[0].clone().as_bannable();
     assert!(pm2.peerdb.is_address_banned(&addr1));
-    let event = filter_connectivity_event::<T, _>(&mut pm2.peer_connectivity_handle, |event| {
-        !std::matches!(
-            event,
-            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
-        )
-    })
-    .await;
+    let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
     assert!(std::matches!(
         event,
         Ok(net::types::ConnectivityEvent::ConnectionClosed { .. })
@@ -115,20 +107,12 @@ where
     )
     .await;
     let peer_id = peer_info.peer_id;
-    pm2.accept_inbound_connection(address, peer_info, None).unwrap();
+    pm2.accept_inbound_connection(address, peer_info, None).await.unwrap();
 
     assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
-    let addr1 = pm1.peer_connectivity_handle.local_addresses().await.unwrap()[0]
-        .clone()
-        .as_bannable();
+    let addr1 = pm1.peer_connectivity_handle.local_addresses()[0].clone().as_bannable();
     assert!(pm2.peerdb.is_address_banned(&addr1));
-    let event = filter_connectivity_event::<T, _>(&mut pm2.peer_connectivity_handle, |event| {
-        !std::matches!(
-            event,
-            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
-        )
-    })
-    .await;
+    let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
     assert!(std::matches!(
         event,
         Ok(net::types::ConnectivityEvent::ConnectionClosed { .. })
@@ -178,26 +162,18 @@ where
     )
     .await;
     let peer_id = peer_info1.peer_id;
-    pm2.accept_inbound_connection(address, peer_info1, None).unwrap();
+    pm2.accept_inbound_connection(address, peer_info1, None).await.unwrap();
 
     assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
-    let addr1 = pm1.peer_connectivity_handle.local_addresses().await.unwrap()[0]
-        .clone()
-        .as_bannable();
+    let addr1 = pm1.peer_connectivity_handle.local_addresses()[0].clone().as_bannable();
     assert!(pm2.peerdb.is_address_banned(&addr1));
-    let event = filter_connectivity_event::<T, _>(&mut pm2.peer_connectivity_handle, |event| {
-        !std::matches!(
-            event,
-            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
-        )
-    })
-    .await;
+    let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
     assert!(matches!(
         event,
         Ok(net::types::ConnectivityEvent::ConnectionClosed { .. })
     ));
 
-    let remote_addr = pm1.peer_connectivity_handle.local_addresses().await.unwrap()[0].clone();
+    let remote_addr = pm1.peer_connectivity_handle.local_addresses()[0].clone();
 
     tokio::spawn(async move {
         loop {
@@ -248,51 +224,63 @@ where
         make_peer_manager::<S>(A::make_transport(), A::make_address(), Arc::clone(&config)).await;
 
     // invalid magic bytes
-    let res = peer_manager.accept_connection(
-        B::new(),
-        Role::Outbound,
-        net::types::PeerInfo::<S::PeerId> {
-            peer_id,
-            network: [1, 2, 3, 4],
-            version: SemVer::new(0, 1, 0),
-            agent: None,
-            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
-        },
-        None,
-    );
+    let res = peer_manager
+        .accept_connection(
+            B::new(),
+            Role::Outbound,
+            net::types::PeerInfo::<S::PeerId> {
+                peer_id,
+                network: [1, 2, 3, 4],
+                version: SemVer::new(0, 1, 0),
+                agent: None,
+                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
+                    .into_iter()
+                    .collect(),
+            },
+            None,
+        )
+        .await;
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
 
     // invalid version
-    let res = peer_manager.accept_connection(
-        B::new(),
-        Role::Outbound,
-        net::types::PeerInfo::<S::PeerId> {
-            peer_id,
-            network: *config.magic_bytes(),
-            version: SemVer::new(1, 1, 1),
-            agent: None,
-            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
-        },
-        None,
-    );
+    let res = peer_manager
+        .accept_connection(
+            B::new(),
+            Role::Outbound,
+            net::types::PeerInfo::<S::PeerId> {
+                peer_id,
+                network: *config.magic_bytes(),
+                version: SemVer::new(1, 1, 1),
+                agent: None,
+                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
+                    .into_iter()
+                    .collect(),
+            },
+            None,
+        )
+        .await;
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
 
     // valid connection
     let address = B::new();
-    let res = peer_manager.accept_connection(
-        address.clone(),
-        Role::Outbound,
-        net::types::PeerInfo::<S::PeerId> {
-            peer_id,
-            network: *config.magic_bytes(),
-            version: SemVer::new(0, 1, 0),
-            agent: None,
-            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
-        },
-        None,
-    );
+    let res = peer_manager
+        .accept_connection(
+            address.clone(),
+            Role::Outbound,
+            net::types::PeerInfo::<S::PeerId> {
+                peer_id,
+                network: *config.magic_bytes(),
+                version: SemVer::new(0, 1, 0),
+                agent: None,
+                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
+                    .into_iter()
+                    .collect(),
+            },
+            None,
+        )
+        .await;
     assert!(res.is_ok());
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(peer_manager.peerdb.is_active_peer(&peer_id));
@@ -341,48 +329,60 @@ 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(
-        B::new(),
-        net::types::PeerInfo::<S::PeerId> {
-            peer_id,
-            network: [1, 2, 3, 4],
-            version: SemVer::new(0, 1, 0),
-            agent: None,
-            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
-        },
-        None,
-    );
+    let res = peer_manager
+        .accept_inbound_connection(
+            B::new(),
+            net::types::PeerInfo::<S::PeerId> {
+                peer_id,
+                network: [1, 2, 3, 4],
+                version: SemVer::new(0, 1, 0),
+                agent: None,
+                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
+                    .into_iter()
+                    .collect(),
+            },
+            None,
+        )
+        .await;
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
 
     // invalid version
-    let res = peer_manager.accept_inbound_connection(
-        B::new(),
-        net::types::PeerInfo::<S::PeerId> {
-            peer_id,
-            network: *config.magic_bytes(),
-            version: SemVer::new(1, 1, 1),
-            agent: None,
-            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
-        },
-        None,
-    );
+    let res = peer_manager
+        .accept_inbound_connection(
+            B::new(),
+            net::types::PeerInfo::<S::PeerId> {
+                peer_id,
+                network: *config.magic_bytes(),
+                version: SemVer::new(1, 1, 1),
+                agent: None,
+                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
+                    .into_iter()
+                    .collect(),
+            },
+            None,
+        )
+        .await;
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
 
     // valid connection
     let address = B::new();
-    let res = peer_manager.accept_inbound_connection(
-        address.clone(),
-        net::types::PeerInfo::<S::PeerId> {
-            peer_id,
-            network: *config.magic_bytes(),
-            version: SemVer::new(0, 1, 0),
-            agent: None,
-            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
-        },
-        None,
-    );
+    let res = peer_manager
+        .accept_inbound_connection(
+            address.clone(),
+            net::types::PeerInfo::<S::PeerId> {
+                peer_id,
+                network: *config.magic_bytes(),
+                version: SemVer::new(0, 1, 0),
+                agent: None,
+                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
+                    .into_iter()
+                    .collect(),
+            },
+            None,
+        )
+        .await;
     assert!(res.is_ok());
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(peer_manager.peerdb.is_active_peer(&peer_id));
@@ -451,13 +451,7 @@ where
     // that tries to connect to the first manager
     tokio::spawn(async move { pm1.run().await });
 
-    let event = filter_connectivity_event::<T, _>(&mut pm2.peer_connectivity_handle, |event| {
-        !std::matches!(
-            event,
-            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
-        )
-    })
-    .await;
+    let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
     if let Ok(net::types::ConnectivityEvent::ConnectionClosed { peer_id }) = event {
         assert_eq!(peer_id, peer_info.peer_id);
     } else {
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index fcb87111d..f24197d22 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -26,7 +26,7 @@ use crate::{
     net::types::Role,
     peer_manager::tests::run_peer_manager,
     testing_utils::{
-        connect_services, filter_connectivity_event, TestTransportChannel, TestTransportMaker,
+        connect_services, get_connectivity_event, TestTransportChannel, TestTransportMaker,
         TestTransportNoise, TestTransportTcp,
     },
 };
@@ -114,7 +114,7 @@ where
     let mut pm1 = make_peer_manager::<T>(A::make_transport(), addr1, Arc::clone(&config)).await;
     let mut pm2 = make_peer_manager::<T>(A::make_transport(), addr2, config).await;
 
-    let addr = pm2.peer_connectivity_handle.local_addresses().await.unwrap()[0].clone();
+    let addr = pm2.peer_connectivity_handle.local_addresses()[0].clone();
 
     tokio::spawn(async move {
         loop {
@@ -123,7 +123,7 @@ where
     });
 
     // "discover" the other networking service
-    pm1.peer_discovered(&addr);
+    pm1.peerdb.peer_discovered(&addr);
     pm1.heartbeat().await.unwrap();
 
     assert_eq!(pm1.pending.len(), 1);
@@ -260,7 +260,7 @@ where
     )
     .await;
     assert_eq!(
-        pm2.accept_inbound_connection(address, peer_info, None),
+        pm2.accept_inbound_connection(address, peer_info, None).await,
         Ok(())
     );
 }
@@ -315,7 +315,7 @@ where
     .await;
 
     assert_eq!(
-        pm2.accept_inbound_connection(address, peer_info, None),
+        pm2.accept_inbound_connection(address, peer_info, None).await,
         Err(P2pError::ProtocolError(ProtocolError::DifferentNetwork(
             [1, 2, 3, 4],
             *config::create_mainnet().magic_bytes(),
@@ -436,13 +436,7 @@ where
     // that tries to connect to the first manager
     tokio::spawn(async move { pm1.run().await });
 
-    let event = filter_connectivity_event::<T, _>(&mut pm2.peer_connectivity_handle, |event| {
-        !std::matches!(
-            event,
-            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
-        )
-    })
-    .await;
+    let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
     if let Ok(net::types::ConnectivityEvent::ConnectionClosed { peer_id }) = event {
         assert_eq!(peer_id, peer_info.peer_id);
     } else {
@@ -542,7 +536,7 @@ where
     pm1.peer_connectivity_handle.connect(addr2).await.expect("dial to succeed");
 
     match timeout(
-        *pm1._p2p_config.outbound_connection_timeout,
+        *pm1.p2p_config.outbound_connection_timeout,
         pm1.peer_connectivity_handle.poll_next(),
     )
     .await
@@ -676,7 +670,17 @@ where
     let chain_config = Arc::new(config::create_mainnet());
 
     // Start first peer manager
-    let p2p_config_1 = Arc::new(P2pConfig::default());
+    let p2p_config_1 = Arc::new(P2pConfig {
+        bind_addresses: Default::default(),
+        added_nodes: Default::default(),
+        ban_threshold: Default::default(),
+        ban_duration: Default::default(),
+        outbound_connection_timeout: Default::default(),
+        node_type: Default::default(),
+        allow_discover_private_ips: Default::default(),
+        heartbeat_interval_min: Default::default(),
+        heartbeat_interval_max: Default::default(),
+    });
     let tx1 = run_peer_manager::<T>(
         A::make_transport(),
         A::make_address(),
@@ -699,6 +703,9 @@ where
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
         node_type: Default::default(),
+        allow_discover_private_ips: Default::default(),
+        heartbeat_interval_min: Default::default(),
+        heartbeat_interval_max: Default::default(),
     });
     let tx1 = run_peer_manager::<T>(
         A::make_transport(),
@@ -715,13 +722,13 @@ where
         let (rtx, rrx) = oneshot::channel();
         tx1.send(PeerManagerEvent::GetConnectedPeers(rtx)).unwrap();
         let connected_peers = timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap();
-        // The backend will also make reverse connection.
-        if connected_peers.len() == 1 || connected_peers.len() == 2 {
+        if connected_peers.len() == 1 {
             break;
         }
         assert!(
             Instant::now().duration_since(started_at) < Duration::from_secs(10),
-            "no peer connected on time"
+            "Unexpected peer count: {}",
+            connected_peers.len()
         );
     }
 }
@@ -741,3 +748,154 @@ async fn connection_add_node_channel() {
     connection_add_node::<TestTransportChannel, DefaultNetworkingService<MpscChannelTransport>>()
         .await;
 }
+
+// Verify that peers announce own addresses and are discovered by other peers
+async fn discovered_node<A, T>(expected_count: usize)
+where
+    A: TestTransportMaker<Transport = T::Transport, Address = T::Address>,
+    T: NetworkingService + 'static + std::fmt::Debug,
+    T::ConnectivityHandle: ConnectivityService<T>,
+{
+    let chain_config = Arc::new(config::create_mainnet());
+
+    // Start the first peer manager
+    let p2p_config_1 = Arc::new(P2pConfig {
+        bind_addresses: Default::default(),
+        added_nodes: Default::default(),
+        ban_threshold: Default::default(),
+        ban_duration: Default::default(),
+        outbound_connection_timeout: Default::default(),
+        node_type: Default::default(),
+        allow_discover_private_ips: true.into(),
+        heartbeat_interval_min: Duration::from_secs(1).into(),
+        heartbeat_interval_max: Duration::from_secs(2).into(),
+    });
+    let tx1 = run_peer_manager::<T>(
+        A::make_transport(),
+        A::make_address(),
+        Arc::clone(&chain_config),
+        p2p_config_1,
+    )
+    .await;
+
+    // Get the first peer manager's bind address
+    let (rtx, rrx) = oneshot::channel();
+    tx1.send(PeerManagerEvent::GetBindAddresses(rtx)).unwrap();
+    let bind_addresses = timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap();
+    assert_eq!(bind_addresses.len(), 1);
+
+    // Start the second peer manager and let it know about the first peer using added_nodes
+    let p2p_config_2 = Arc::new(P2pConfig {
+        bind_addresses: Default::default(),
+        added_nodes: bind_addresses.clone(),
+        ban_threshold: Default::default(),
+        ban_duration: Default::default(),
+        outbound_connection_timeout: Default::default(),
+        node_type: Default::default(),
+        allow_discover_private_ips: true.into(),
+        heartbeat_interval_min: Duration::from_secs(1).into(),
+        heartbeat_interval_max: Duration::from_secs(2).into(),
+    });
+    let tx2 = run_peer_manager::<T>(
+        A::make_transport(),
+        A::make_address(),
+        Arc::clone(&chain_config),
+        p2p_config_2,
+    )
+    .await;
+
+    // Start the third peer manager and let it know about the first peer using added_nodes
+    let p2p_config_3 = Arc::new(P2pConfig {
+        bind_addresses: Default::default(),
+        added_nodes: bind_addresses,
+        ban_threshold: Default::default(),
+        ban_duration: Default::default(),
+        outbound_connection_timeout: Default::default(),
+        node_type: Default::default(),
+        allow_discover_private_ips: true.into(),
+        heartbeat_interval_min: Duration::from_secs(1).into(),
+        heartbeat_interval_max: Duration::from_secs(2).into(),
+    });
+    let tx3 = run_peer_manager::<T>(
+        A::make_transport(),
+        A::make_address(),
+        Arc::clone(&chain_config),
+        p2p_config_3,
+    )
+    .await;
+
+    let started_at = Instant::now();
+
+    // The second peer should discover the third peer
+    loop {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        let (rtx, rrx) = oneshot::channel();
+        tx2.send(PeerManagerEvent::GetConnectedPeers(rtx)).unwrap();
+        let connected_peers = timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap();
+        if connected_peers.len() == expected_count {
+            break;
+        }
+        assert!(
+            Instant::now().duration_since(started_at) < Duration::from_secs(10),
+            "Unexpected peer count: {}, expected: {}",
+            connected_peers.len(),
+            expected_count,
+        );
+    }
+
+    // Check the third peer manager
+    loop {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        let (rtx, rrx) = oneshot::channel();
+        tx3.send(PeerManagerEvent::GetConnectedPeers(rtx)).unwrap();
+        let connected_peers = timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap();
+        if connected_peers.len() == expected_count {
+            break;
+        }
+        assert!(
+            Instant::now().duration_since(started_at) < Duration::from_secs(10),
+            "Unexpected peer count: {}, expected: {}",
+            connected_peers.len(),
+            expected_count,
+        );
+    }
+
+    // And the first peer manager
+    loop {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        let (rtx, rrx) = oneshot::channel();
+        tx1.send(PeerManagerEvent::GetConnectedPeers(rtx)).unwrap();
+        let connected_peers = timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap();
+        if connected_peers.len() == expected_count {
+            break;
+        }
+        assert!(
+            Instant::now().duration_since(started_at) < Duration::from_secs(10),
+            "Unexpected peer count: {}, expected: {}",
+            connected_peers.len(),
+            expected_count,
+        );
+    }
+}
+
+#[tokio::test]
+async fn discovered_node_tcp() {
+    // With TCP all listening addresses are discovered and multiple connections are made:
+    // For example A mades outbound connections to B and C and accepts connections from B and C.
+    // There will be 4 connections reported for A.
+    discovered_node::<TestTransportTcp, DefaultNetworkingService<TcpTransportSocket>>(4).await;
+}
+
+#[tokio::test]
+async fn discovered_node_noise() {
+    // Same as with TCP
+    discovered_node::<TestTransportNoise, DefaultNetworkingService<NoiseTcpTransport>>(4).await;
+}
+
+#[tokio::test]
+async fn discovered_node_channel() {
+    // With TestTransport peers made lower total connection count because there is only one "port" available.
+    // So when peer A connects to peer B, peer B won't open connection to A.
+    discovered_node::<TestTransportChannel, DefaultNetworkingService<MpscChannelTransport>>(2)
+        .await;
+}
diff --git a/p2p/src/sync/tests/mod.rs b/p2p/src/sync/tests/mod.rs
index 8384b728b..7af607a42 100644
--- a/p2p/src/sync/tests/mod.rs
+++ b/p2p/src/sync/tests/mod.rs
@@ -76,6 +76,9 @@ where
         ban_duration: Default::default(),
         outbound_connection_timeout: Default::default(),
         node_type: NodeType::Full.into(),
+        allow_discover_private_ips: Default::default(),
+        heartbeat_interval_min: Default::default(),
+        heartbeat_interval_max: Default::default(),
     });
     let (conn, sync) = T::start(
         transport,
diff --git a/p2p/src/testing_utils.rs b/p2p/src/testing_utils.rs
index f3e5865af..06e862c9a 100644
--- a/p2p/src/testing_utils.rs
+++ b/p2p/src/testing_utils.rs
@@ -152,10 +152,7 @@ where
     T: NetworkingService + Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
 {
-    let addr = timeout(Duration::from_secs(5), conn2.local_addresses())
-        .await
-        .expect("local address fetch not to timeout")
-        .unwrap();
+    let addr = conn2.local_addresses();
     conn1.connect(addr[0].clone()).await.expect("dial to succeed");
 
     let (address, peer_info1) = match timeout(Duration::from_secs(5), conn2.poll_next()).await {
@@ -185,11 +182,7 @@ where
     (address, peer_info1, peer_info2)
 }
 
-/// Return first event that is accepted by predicate.
-///
-/// Used to skip events that are not of interest or that are different between backends
-/// (for example ConnectivityEvent::Discovered).
-/// Can be used in tests only, will panic in case of errors.
+/// Returns first event that is accepted by predicate or panics on timeout.
 pub async fn filter_connectivity_event<T, F>(
     conn: &mut T::ConnectivityHandle,
     predicate: F,
@@ -199,12 +192,27 @@ where
     T::ConnectivityHandle: ConnectivityService<T>,
     F: Fn(&crate::Result<ConnectivityEvent<T>>) -> bool,
 {
-    loop {
-        let result = timeout(Duration::from_secs(10), conn.poll_next())
-            .await
-            .expect("unexpected timeout receiving connectivity event");
-        if predicate(&result) {
-            return result;
+    let recv_fut = async {
+        loop {
+            let result = conn.poll_next().await;
+            if predicate(&result) {
+                break result;
+            }
         }
-    }
+    };
+
+    timeout(Duration::from_secs(10), recv_fut)
+        .await
+        .expect("unexpected timeout receiving connectivity event")
+}
+
+/// Returns first event or panics on timeout.
+pub async fn get_connectivity_event<T>(
+    conn: &mut T::ConnectivityHandle,
+) -> crate::Result<ConnectivityEvent<T>>
+where
+    T: NetworkingService,
+    T::ConnectivityHandle: ConnectivityService<T>,
+{
+    filter_connectivity_event(conn, |_event| true).await
 }