Repository: mintlayer/mintlayer-core
Pull request #635 (2023-01-17 14:57:58)

Basic peer discovery
This PR adds new `AddrList` request and response types. With this nodes will be able query peers for newly discovered addresses. This new messages are sent to `PeerManager` because it's where `PeerDB` is used and where peers list maintenance happens. For now actual requests handling is very basic, just to show how it might look. Actual implementation will be done in next PR (`p2p/README.md` also needs some updates).



Patch:
diff --git a/p2p/src/error.rs b/p2p/src/error.rs
index 09a786318..f1a15d409 100644
--- a/p2p/src/error.rs
+++ b/p2p/src/error.rs
@@ -57,8 +57,6 @@ pub enum PeerError {
 pub enum PublishError {
     #[error("Failed to sign message")]
     SigningFailed,
-    #[error("No peers in topic")]
-    InsufficientPeers,
     #[error("Message is too large. Tried to send {0:?} bytes when limit is {1:?}")]
     MessageTooLarge(usize, usize),
     #[error("Failed to compress the message")]
@@ -209,7 +207,6 @@ impl BanScore for PublishError {
     fn ban_score(&self) -> u32 {
         match self {
             PublishError::SigningFailed => 0,
-            PublishError::InsufficientPeers => 0,
             PublishError::MessageTooLarge(_, _) => 100,
             PublishError::TransformFailed => 0,
         }
diff --git a/p2p/src/message.rs b/p2p/src/message.rs
index 82230991d..ea167e945 100644
--- a/p2p/src/message.rs
+++ b/p2p/src/message.rs
@@ -20,6 +20,8 @@ use common::{
 };
 use serialization::{Decode, Encode};
 
+use crate::types::peer_address::PeerAddress;
+
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub struct HeaderListRequest {
     locator: Locator,
@@ -49,7 +51,7 @@ impl BlockListRequest {
         Self { block_ids }
     }
 
-    pub fn block_ids(&self) -> &Vec<Id<Block>> {
+    pub fn block_ids(&self) -> &[Id<Block>] {
         &self.block_ids
     }
 
@@ -58,12 +60,28 @@ impl BlockListRequest {
     }
 }
 
+#[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
+pub struct AddrListRequest {}
+
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub enum Request {
     #[codec(index = 0)]
     HeaderListRequest(HeaderListRequest),
     #[codec(index = 1)]
     BlockListRequest(BlockListRequest),
+    #[codec(index = 2)]
+    AddrListRequest(AddrListRequest),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SyncRequest {
+    HeaderListRequest(HeaderListRequest),
+    BlockListRequest(BlockListRequest),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum PeerManagerRequest {
+    AddrListRequest(AddrListRequest),
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
@@ -76,7 +94,7 @@ impl HeaderListResponse {
         Self { headers }
     }
 
-    pub fn headers(&self) -> &Vec<BlockHeader> {
+    pub fn headers(&self) -> &[BlockHeader] {
         &self.headers
     }
 
@@ -95,7 +113,7 @@ impl BlockListResponse {
         Self { blocks }
     }
 
-    pub fn blocks(&self) -> &Vec<Block> {
+    pub fn blocks(&self) -> &[Block] {
         &self.blocks
     }
 
@@ -104,12 +122,40 @@ impl BlockListResponse {
     }
 }
 
+#[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
+pub struct AddrListResponse {
+    addresses: Vec<PeerAddress>,
+}
+
+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 enum Response {
     #[codec(index = 0)]
     HeaderListResponse(HeaderListResponse),
     #[codec(index = 1)]
     BlockListResponse(BlockListResponse),
+    #[codec(index = 2)]
+    AddrListResponse(AddrListResponse),
+}
+
+#[derive(Debug, Clone)]
+pub enum SyncResponse {
+    HeaderListResponse(HeaderListResponse),
+    BlockListResponse(BlockListResponse),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum PeerManagerResponse {
+    AddrListResponse(AddrListResponse),
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
@@ -117,3 +163,37 @@ pub enum Announcement {
     #[codec(index = 0)]
     Block(Block),
 }
+
+impl From<PeerManagerRequest> for Request {
+    fn from(request: PeerManagerRequest) -> Self {
+        match request {
+            PeerManagerRequest::AddrListRequest(request) => Request::AddrListRequest(request),
+        }
+    }
+}
+
+impl From<PeerManagerResponse> for Response {
+    fn from(response: PeerManagerResponse) -> Self {
+        match response {
+            PeerManagerResponse::AddrListResponse(response) => Response::AddrListResponse(response),
+        }
+    }
+}
+
+impl From<SyncRequest> for Request {
+    fn from(request: SyncRequest) -> Self {
+        match request {
+            SyncRequest::HeaderListRequest(request) => Request::HeaderListRequest(request),
+            SyncRequest::BlockListRequest(request) => Request::BlockListRequest(request),
+        }
+    }
+}
+
+impl From<SyncResponse> for Response {
+    fn from(response: SyncResponse) -> Self {
+        match response {
+            SyncResponse::HeaderListResponse(response) => Response::HeaderListResponse(response),
+            SyncResponse::BlockListResponse(response) => Response::BlockListResponse(response),
+        }
+    }
+}
diff --git a/p2p/src/net/mock/backend.rs b/p2p/src/net/mock/backend.rs
index a60b7ca15..718184ce0 100644
--- a/p2p/src/net/mock/backend.rs
+++ b/p2p/src/net/mock/backend.rs
@@ -15,11 +15,7 @@
 
 //! Mock networking backend
 //!
-//! The backend is modeled after libp2p.
-//!
-//! The peers are required to have unique IDs which they self-assign to themselves
-//! and advertise via the `Hello` message. Until the peer ID has been received, the
-//! peers are distinguished by their socket addresses.
+//! Every connected peer gets unique ID (generated locally from a counter).
 
 use std::{
     collections::{BTreeSet, HashMap},
@@ -42,7 +38,7 @@ use serialization::{Decode, Encode};
 use crate::{
     config::P2pConfig,
     error::{DialError, P2pError, PeerError, PublishError},
-    message,
+    message::{self, PeerManagerRequest, PeerManagerResponse, SyncRequest, SyncResponse},
     net::{
         mock::{
             constants::ANNOUNCEMENT_MAX_SIZE,
@@ -53,10 +49,9 @@ use crate::{
                 MockRequestId, PeerEvent, SyncingEvent,
             },
         },
-        types::{PubSubTopic, Role},
+        types::PubSubTopic,
         Announcement,
     },
-    types::peer_address::PeerAddress,
 };
 
 use super::{peer::PeerRole, transport::TransportAddress, types::HandshakeNonce};
@@ -198,18 +193,19 @@ where
     /// Sends a request to the remote peer.
     async fn send_request(
         &mut self,
+        request_id: MockRequestId,
         peer_id: MockPeerId,
         request: message::Request,
-    ) -> crate::Result<MockRequestId> {
+    ) -> crate::Result<()> {
         let peer = self
             .peers
             .get_mut(&peer_id)
             .ok_or(P2pError::PeerError(PeerError::PeerDoesntExist))?;
 
-        let (request_id, request) = self.request_mgr.make_request(request)?;
+        let request = self.request_mgr.make_request(request_id, request)?;
         peer.tx.send(MockEvent::SendMessage(request)).await.map_err(P2pError::from)?;
 
-        Ok(request_id)
+        Ok(())
     }
 
     /// Send response to a request
@@ -237,8 +233,7 @@ where
 
     /// Sends the announcement to all peers.
     ///
-    /// Returns the `InsufficientPeers` error if there are no peers that subscribed to the related
-    /// topic.
+    /// It is not an error if there are no peers that subscribed to the related topic.
     async fn announce_data(&mut self, topic: PubSubTopic, message: Vec<u8>) -> crate::Result<()> {
         let announcement = message::Announcement::decode(&mut &message[..])?;
 
@@ -259,14 +254,9 @@ where
             .collect();
         futures.shuffle(&mut make_pseudo_rng());
 
-        // TODO: We don't really need to return an error here. It is only needed temporarily in
-        // order to mimic the libp2p behavior.
-        if futures.is_empty() {
-            Err(P2pError::PublishError(PublishError::InsufficientPeers))
-        } else {
-            join_all(futures).await;
-            Ok(())
-        }
+        join_all(futures).await;
+
+        Ok(())
     }
 
     /// Handle incoming request
@@ -278,16 +268,37 @@ where
     ) -> crate::Result<()> {
         log::trace!("request received from peer {peer_id}, request id {request_id}");
 
-        let request_id = self.request_mgr.register_request(&peer_id, &request_id, &request)?;
+        let request_id = self.request_mgr.register_request(&peer_id, &request_id)?;
 
-        self.sync_tx
-            .send(SyncingEvent::Request {
-                peer_id,
-                request_id,
-                request,
-            })
-            .await
-            .map_err(P2pError::from)
+        match request {
+            message::Request::HeaderListRequest(request) => self
+                .sync_tx
+                .send(SyncingEvent::Request {
+                    peer_id,
+                    request_id,
+                    request: SyncRequest::HeaderListRequest(request),
+                })
+                .await
+                .map_err(P2pError::from),
+            message::Request::BlockListRequest(request) => self
+                .sync_tx
+                .send(SyncingEvent::Request {
+                    peer_id,
+                    request_id,
+                    request: SyncRequest::BlockListRequest(request),
+                })
+                .await
+                .map_err(P2pError::from),
+            message::Request::AddrListRequest(request) => self
+                .conn_tx
+                .send(ConnectivityEvent::Request {
+                    peer_id,
+                    request_id,
+                    request: PeerManagerRequest::AddrListRequest(request),
+                })
+                .await
+                .map_err(P2pError::from),
+        }
     }
 
     /// Handle incoming response
@@ -298,14 +309,36 @@ where
         response: message::Response,
     ) -> crate::Result<()> {
         log::trace!("response received from peer {peer_id}, request id {request_id}");
-        self.sync_tx
-            .send(SyncingEvent::Response {
-                peer_id,
-                request_id,
-                response,
-            })
-            .await
-            .map_err(P2pError::from)
+
+        match response {
+            message::Response::HeaderListResponse(response) => self
+                .sync_tx
+                .send(SyncingEvent::Response {
+                    peer_id,
+                    request_id,
+                    response: SyncResponse::HeaderListResponse(response),
+                })
+                .await
+                .map_err(P2pError::from),
+            message::Response::BlockListResponse(response) => self
+                .sync_tx
+                .send(SyncingEvent::Response {
+                    peer_id,
+                    request_id,
+                    response: SyncResponse::BlockListResponse(response),
+                })
+                .await
+                .map_err(P2pError::from),
+            message::Response::AddrListResponse(response) => self
+                .conn_tx
+                .send(ConnectivityEvent::Response {
+                    peer_id,
+                    request_id,
+                    response: PeerManagerResponse::AddrListResponse(response),
+                })
+                .await
+                .map_err(P2pError::from),
+        }
     }
 
     async fn handle_announcement(
@@ -367,19 +400,6 @@ where
         }
     }
 
-    /// Handle received listening port from inbound remote peer
-    fn handle_own_receiver_address(
-        &mut self,
-        receiver_address: PeerAddress,
-        role: Role,
-    ) -> crate::Result<()> {
-        log::debug!("new own receiver address {receiver_address} found from {role:?} connection");
-
-        // TODO: Handle receiver address
-
-        Ok(())
-    }
-
     /// Create new peer
     ///
     /// Move the connection to `pending` where it stays until either the connection is closed
@@ -453,7 +473,7 @@ where
                     self.pending.remove(&outbound_peer_id).expect("peer must exist");
 
                 log::info!(
-                    "self-connection detect on address {:?}",
+                    "self-connection detected on address {:?}",
                     outbound_pending.address
                 );
 
@@ -513,6 +533,7 @@ where
                                     agent: None,
                                     subscriptions: subscriptions.clone(),
                                 },
+                                receiver_address,
                             })
                             .await
                             .map_err(P2pError::from)?;
@@ -528,16 +549,13 @@ where
                                     agent: None,
                                     subscriptions: subscriptions.clone(),
                                 },
+                                receiver_address,
                             })
                             .await
                             .map_err(P2pError::from)?;
                     }
                 }
 
-                if let Some(receiver_address) = receiver_address {
-                    self.handle_own_receiver_address(receiver_address, peer_role.into())?;
-                }
-
                 self.peers.insert(peer_id, PeerContext { subscriptions, tx });
                 let _ = self.request_mgr.register_peer(peer_id);
             }
@@ -623,13 +641,16 @@ where
             .boxed(),
             Command::SendRequest {
                 peer_id,
+                request_id,
                 message,
-                response,
             } => async move {
                 boxed_cb(move |this| {
                     async move {
-                        let res = this.send_request(peer_id, message).await;
-                        response.send(res).map_err(|_| P2pError::ChannelClosed)
+                        let res = this.send_request(request_id, peer_id, message).await;
+                        if let Err(e) = res {
+                            log::debug!("Failed to send request to peer {peer_id}: {e}")
+                        }
+                        Ok(())
                     }
                     .boxed()
                 })
@@ -638,26 +659,27 @@ where
             Command::SendResponse {
                 request_id,
                 message,
-                response,
             } => async move {
                 boxed_cb(move |this| {
                     async move {
                         let res = this.send_response(request_id, message).await;
-                        response.send(res).map_err(|_| P2pError::ChannelClosed)
+                        if let Err(e) = res {
+                            log::debug!("Failed to send response to peer: {e}")
+                        }
+                        Ok(())
                     }
                     .boxed()
                 })
             }
             .boxed(),
-            Command::AnnounceData {
-                topic,
-                message,
-                response,
-            } => async move {
+            Command::AnnounceData { topic, message } => async move {
                 boxed_cb(move |this| {
                     async move {
                         let res = this.announce_data(topic, message).await;
-                        response.send(res).map_err(|_| P2pError::ChannelClosed)
+                        if let Err(e) = res {
+                            log::error!("Failed to send announce data: {e}")
+                        }
+                        Ok(())
                     }
                     .boxed()
                 })
diff --git a/p2p/src/net/mock/mod.rs b/p2p/src/net/mock/mod.rs
index 47a343344..aea91a75a 100644
--- a/p2p/src/net/mock/mod.rs
+++ b/p2p/src/net/mock/mod.rs
@@ -31,7 +31,7 @@ use serialization::Encode;
 use crate::{
     config,
     error::{P2pError, PublishError},
-    message,
+    message::{self, PeerManagerRequest, PeerManagerResponse, SyncRequest, SyncResponse},
     net::{
         mock::{
             constants::ANNOUNCEMENT_MAX_SIZE,
@@ -111,7 +111,7 @@ impl<T: TransportSocket> NetworkingService for MockService<T> {
     type Address = T::Address;
     type BannableAddress = T::BannableAddress;
     type PeerId = MockPeerId;
-    type SyncingPeerRequestId = MockRequestId;
+    type PeerRequestId = MockRequestId;
     type ConnectivityHandle = MockConnectivityHandle<Self, T>;
     type SyncingMessagingHandle = MockSyncingMessagingHandle<Self, T>;
 
@@ -164,7 +164,8 @@ impl<T: TransportSocket> NetworkingService for MockService<T> {
 #[async_trait]
 impl<S, T> ConnectivityService<S> for MockConnectivityHandle<S, T>
 where
-    S: NetworkingService<Address = T::Address, PeerId = MockPeerId> + Send,
+    S: NetworkingService<Address = T::Address, PeerId = MockPeerId, PeerRequestId = MockRequestId>
+        + Send,
     MockPeerInfo: TryInto<PeerInfo<S>, Error = P2pError>,
     T: TransportSocket,
 {
@@ -199,24 +200,80 @@ where
         rx.await?
     }
 
+    async fn send_request(
+        &mut self,
+        peer_id: S::PeerId,
+        request: PeerManagerRequest,
+    ) -> crate::Result<S::PeerRequestId> {
+        let request_id = MockRequestId::new();
+
+        self.cmd_tx
+            .send(types::Command::SendRequest {
+                peer_id,
+                request_id,
+                message: request.into(),
+            })
+            .await?;
+
+        Ok(request_id)
+    }
+
+    async fn send_response(
+        &mut self,
+        request_id: S::PeerRequestId,
+        response: PeerManagerResponse,
+    ) -> crate::Result<()> {
+        self.cmd_tx
+            .send(types::Command::SendResponse {
+                request_id,
+                message: response.into(),
+            })
+            .await?;
+        Ok(())
+    }
+
     async fn local_addresses(&self) -> crate::Result<Vec<S::Address>> {
         Ok(self.local_addresses.clone())
     }
 
     async fn poll_next(&mut self) -> crate::Result<ConnectivityEvent<S>> {
         match self.conn_rx.recv().await.ok_or(P2pError::ChannelClosed)? {
-            types::ConnectivityEvent::OutboundAccepted { address, peer_info } => {
-                Ok(ConnectivityEvent::OutboundAccepted {
-                    address,
-                    peer_info: peer_info.try_into()?,
-                })
-            }
-            types::ConnectivityEvent::InboundAccepted { address, peer_info } => {
-                Ok(ConnectivityEvent::InboundAccepted {
-                    address,
-                    peer_info: peer_info.try_into()?,
-                })
-            }
+            types::ConnectivityEvent::Request {
+                peer_id,
+                request_id,
+                request,
+            } => Ok(ConnectivityEvent::Request {
+                peer_id,
+                request_id,
+                request,
+            }),
+            types::ConnectivityEvent::Response {
+                peer_id,
+                request_id,
+                response,
+            } => Ok(ConnectivityEvent::Response {
+                peer_id,
+                request_id,
+                response,
+            }),
+            types::ConnectivityEvent::InboundAccepted {
+                address,
+                peer_info,
+                receiver_address,
+            } => Ok(ConnectivityEvent::InboundAccepted {
+                address,
+                peer_info: peer_info.try_into()?,
+                receiver_address,
+            }),
+            types::ConnectivityEvent::OutboundAccepted {
+                address,
+                peer_info,
+                receiver_address,
+            } => Ok(ConnectivityEvent::OutboundAccepted {
+                address,
+                peer_info: peer_info.try_into()?,
+                receiver_address,
+            }),
             types::ConnectivityEvent::ConnectionError { address, error } => {
                 Ok(ConnectivityEvent::ConnectionError { address, error })
             }
@@ -227,9 +284,7 @@ where
                 Ok(ConnectivityEvent::Misbehaved { peer_id, error })
             }
             types::ConnectivityEvent::AddressDiscovered { address } => {
-                Ok(ConnectivityEvent::Discovered {
-                    addresses: vec![address],
-                })
+                Ok(ConnectivityEvent::AddressDiscovered { address })
             }
         }
     }
@@ -238,40 +293,39 @@ where
 #[async_trait]
 impl<S, T> SyncingMessagingService<S> for MockSyncingMessagingHandle<S, T>
 where
-    S: NetworkingService<PeerId = MockPeerId, SyncingPeerRequestId = MockRequestId> + Send,
+    S: NetworkingService<PeerId = MockPeerId, PeerRequestId = MockRequestId> + Send,
     T: TransportSocket,
 {
     async fn send_request(
         &mut self,
         peer_id: S::PeerId,
-        request: message::Request,
-    ) -> crate::Result<S::SyncingPeerRequestId> {
-        let (tx, rx) = oneshot::channel();
+        request: SyncRequest,
+    ) -> crate::Result<S::PeerRequestId> {
+        let request_id = MockRequestId::new();
 
         self.cmd_tx
             .send(types::Command::SendRequest {
                 peer_id,
-                message: request,
-                response: tx,
+                request_id,
+                message: request.into(),
             })
             .await?;
-        rx.await?
+
+        Ok(request_id)
     }
 
     async fn send_response(
         &mut self,
-        request_id: S::SyncingPeerRequestId,
-        response: message::Response,
+        request_id: S::PeerRequestId,
+        response: SyncResponse,
     ) -> crate::Result<()> {
-        let (tx, rx) = oneshot::channel();
         self.cmd_tx
             .send(types::Command::SendResponse {
                 request_id,
-                message: response,
-                response: tx,
+                message: response.into(),
             })
             .await?;
-        rx.await?
+        Ok(())
     }
 
     async fn make_announcement(
@@ -290,15 +344,9 @@ where
             message::Announcement::Block(_) => PubSubTopic::Blocks,
         };
 
-        let (response, receiver) = oneshot::channel();
-        self.cmd_tx
-            .send(types::Command::AnnounceData {
-                topic,
-                message,
-                response,
-            })
-            .await?;
-        receiver.await?
+        self.cmd_tx.send(types::Command::AnnounceData { topic, message }).await?;
+
+        Ok(())
     }
 
     async fn poll_next(&mut self) -> crate::Result<SyncingEvent<S>> {
@@ -373,8 +421,11 @@ mod tests {
         let addr = conn2.local_addresses().await.unwrap();
         assert_eq!(conn1.connect(addr[0].clone()).await, Ok(()));
 
-        if let Ok(ConnectivityEvent::OutboundAccepted { address, peer_info }) =
-            conn1.poll_next().await
+        if let Ok(ConnectivityEvent::OutboundAccepted {
+            address,
+            peer_info,
+            receiver_address: _,
+        }) = conn1.poll_next().await
         {
             assert_eq!(address, conn2.local_addresses().await.unwrap()[0]);
             assert_eq!(&peer_info.magic_bytes, config.magic_bytes());
@@ -436,6 +487,7 @@ mod tests {
             ConnectivityEvent::InboundAccepted {
                 address: _,
                 peer_info,
+                receiver_address: _,
             } => {
                 assert_eq!(peer_info.magic_bytes, *config.magic_bytes());
                 assert_eq!(
@@ -497,6 +549,7 @@ mod tests {
             ConnectivityEvent::InboundAccepted {
                 address: _,
                 peer_info,
+                receiver_address: _,
             } => {
                 assert_eq!(conn2.disconnect(peer_info.peer_id).await, Ok(()));
             }
@@ -570,8 +623,11 @@ mod tests {
         // Check that we can still connect normally after
         let addr = conn2.local_addresses().await.unwrap();
         assert_eq!(conn1.connect(addr[0].clone()).await, Ok(()));
-        if let Ok(ConnectivityEvent::OutboundAccepted { address, peer_info }) =
-            conn1.poll_next().await
+        if let Ok(ConnectivityEvent::OutboundAccepted {
+            address,
+            peer_info,
+            receiver_address: _,
+        }) = conn1.poll_next().await
         {
             assert_eq!(address, conn2.local_addresses().await.unwrap()[0]);
             assert_eq!(&peer_info.magic_bytes, config.magic_bytes());
diff --git a/p2p/src/net/mock/peer.rs b/p2p/src/net/mock/peer.rs
index 395f0da53..07ccd2127 100644
--- a/p2p/src/net/mock/peer.rs
+++ b/p2p/src/net/mock/peer.rs
@@ -493,7 +493,7 @@ mod tests {
         assert!(socket2.recv().now_or_never().is_none());
         socket2
             .send(types::Message::Request {
-                request_id: types::MockRequestId::new(1337u64),
+                request_id: types::MockRequestId::new(),
                 request: message::Request::HeaderListRequest(message::HeaderListRequest::new(
                     Locator::new(vec![]),
                 )),
diff --git a/p2p/src/net/mock/request_manager.rs b/p2p/src/net/mock/request_manager.rs
index a4ba6c3d9..2100cbb26 100644
--- a/p2p/src/net/mock/request_manager.rs
+++ b/p2p/src/net/mock/request_manager.rs
@@ -36,12 +36,6 @@ pub struct RequestManager {
     /// Active ephemeral IDs
     ephemerals: HashMap<types::MockPeerId, HashSet<types::MockRequestId>>,
 
-    /// The next ID for an outbound request.
-    next_request_id: types::MockRequestId,
-
-    /// Next ephemeral request ID
-    next_ephemeral_id: types::MockRequestId,
-
     /// Ephemeral requests IDs which are mapped to remote peer ID/request ID pair
     ephemeral: HashMap<types::MockRequestId, (types::MockPeerId, types::MockRequestId)>,
 }
@@ -76,17 +70,13 @@ impl RequestManager {
     /// Create new outgoing request
     pub fn make_request(
         &mut self,
+        request_id: types::MockRequestId,
         request: message::Request,
-    ) -> crate::Result<(types::MockRequestId, Box<types::Message>)> {
-        let request_id = self.next_request_id.fetch_and_inc();
-
-        Ok((
+    ) -> crate::Result<Box<types::Message>> {
+        Ok(Box::new(types::Message::Request {
             request_id,
-            Box::new(types::Message::Request {
-                request_id,
-                request,
-            }),
-        ))
+            request,
+        }))
     }
 
     /// Create new outgoing response
@@ -115,18 +105,18 @@ impl RequestManager {
     ///
     /// The request ID is stored into a temporary storage holding all pending
     /// inbound requests.
+    // TODO: Use different type in result so it's not possible to mixup ephemeral and real request ids.
     pub fn register_request(
         &mut self,
         peer_id: &types::MockPeerId,
         request_id: &types::MockRequestId,
-        _request: &message::Request,
     ) -> crate::Result<types::MockRequestId> {
         let peer_ephemerals = self
             .ephemerals
             .get_mut(peer_id)
             .ok_or(P2pError::PeerError(PeerError::PeerDoesntExist))?;
 
-        let ephemeral_id = self.next_ephemeral_id.fetch_and_inc();
+        let ephemeral_id = types::MockRequestId::new();
 
         peer_ephemerals.insert(ephemeral_id);
         self.ephemeral.insert(ephemeral_id, (*peer_id, *request_id));
diff --git a/p2p/src/net/mock/transport/impls/channel.rs b/p2p/src/net/mock/transport/impls/channel.rs
index ee6c78f9f..244b19940 100644
--- a/p2p/src/net/mock/transport/impls/channel.rs
+++ b/p2p/src/net/mock/transport/impls/channel.rs
@@ -198,9 +198,9 @@ impl AsBannableAddress for Address {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::net::{
-        message::{BlockListRequest, Request},
-        mock::{
+    use crate::{
+        message::{BlockListRequest, SyncRequest},
+        net::mock::{
             transport::BufferedTranscoder,
             types::{Message, MockRequestId},
         },
@@ -217,13 +217,13 @@ mod tests {
         let server_stream = server_res.unwrap().0;
         let peer_stream = peer_res.unwrap();
 
-        let request_id = MockRequestId::new(1337u64);
-        let request = Request::BlockListRequest(BlockListRequest::new(vec![]));
+        let request_id = MockRequestId::new();
+        let request = SyncRequest::BlockListRequest(BlockListRequest::new(vec![]));
         let mut peer_stream = BufferedTranscoder::new(peer_stream);
         peer_stream
             .send(Message::Request {
                 request_id,
-                request: request.clone(),
+                request: request.clone().into(),
             })
             .await
             .unwrap();
@@ -233,7 +233,7 @@ mod tests {
             server_stream.recv().await.unwrap(),
             Message::Request {
                 request_id,
-                request,
+                request: request.into(),
             }
         );
     }
diff --git a/p2p/src/net/mock/transport/impls/tcp.rs b/p2p/src/net/mock/transport/impls/tcp.rs
index 3c37ca9f1..c21ab2ee3 100644
--- a/p2p/src/net/mock/transport/impls/tcp.rs
+++ b/p2p/src/net/mock/transport/impls/tcp.rs
@@ -158,15 +158,15 @@ impl PeerStream for TcpTransportStream {}
 
 #[cfg(test)]
 mod tests {
-    use crate::testing_utils::{TestTransportMaker, TestTransportTcp};
+    use crate::{
+        message::{BlockListRequest, SyncRequest},
+        testing_utils::{TestTransportMaker, TestTransportTcp},
+    };
 
     use super::*;
-    use crate::net::{
-        message::{BlockListRequest, Request},
-        mock::{
-            transport::BufferedTranscoder,
-            types::{Message, MockRequestId},
-        },
+    use crate::net::mock::{
+        transport::BufferedTranscoder,
+        types::{Message, MockRequestId},
     };
 
     #[tokio::test]
@@ -179,13 +179,13 @@ mod tests {
         let server_stream = server_res.unwrap().0;
         let peer_stream = peer_res.unwrap();
 
-        let request_id = MockRequestId::new(1337u64);
-        let request = Request::BlockListRequest(BlockListRequest::new(vec![]));
+        let request_id = MockRequestId::new();
+        let request = SyncRequest::BlockListRequest(BlockListRequest::new(vec![]));
         let mut peer_stream = BufferedTranscoder::new(peer_stream);
         peer_stream
             .send(Message::Request {
                 request_id,
-                request: request.clone(),
+                request: request.clone().into(),
             })
             .await
             .unwrap();
@@ -195,7 +195,7 @@ mod tests {
             server_stream.recv().await.unwrap(),
             Message::Request {
                 request_id,
-                request,
+                request: request.into(),
             }
         );
     }
@@ -210,22 +210,22 @@ mod tests {
         let server_stream = server_res.unwrap().0;
         let peer_stream = peer_res.unwrap();
 
-        let id_1 = MockRequestId::new(1337u64);
-        let request = Request::BlockListRequest(BlockListRequest::new(vec![]));
+        let id_1 = MockRequestId::new();
+        let request = SyncRequest::BlockListRequest(BlockListRequest::new(vec![]));
         let mut peer_stream = BufferedTranscoder::new(peer_stream);
         peer_stream
             .send(Message::Request {
                 request_id: id_1,
-                request: request.clone(),
+                request: request.clone().into(),
             })
             .await
             .unwrap();
 
-        let id_2 = MockRequestId::new(1338u64);
+        let id_2 = MockRequestId::new();
         peer_stream
             .send(Message::Request {
                 request_id: id_2,
-                request: request.clone(),
+                request: request.clone().into(),
             })
             .await
             .unwrap();
@@ -235,14 +235,14 @@ mod tests {
             server_stream.recv().await.unwrap(),
             Message::Request {
                 request_id: id_1,
-                request: request.clone(),
+                request: request.clone().into(),
             }
         );
         assert_eq!(
             server_stream.recv().await.unwrap(),
             Message::Request {
                 request_id: id_2,
-                request,
+                request: request.into(),
             }
         );
     }
diff --git a/p2p/src/net/mock/transport/traits/socket.rs b/p2p/src/net/mock/transport/traits/socket.rs
index 0c2e7b4e4..4523b9d00 100644
--- a/p2p/src/net/mock/transport/traits/socket.rs
+++ b/p2p/src/net/mock/transport/traits/socket.rs
@@ -52,6 +52,6 @@ pub trait TransportSocket: Send + Sync + 'static {
     /// Creates a new listener bound to the specified address.
     async fn bind(&self, address: Vec<Self::Address>) -> Result<Self::Listener>;
 
-    /// Open a connection to the given address.
+    /// Returns a future that opens a connection to the given address.
     fn connect(&self, address: Self::Address) -> BoxFuture<'static, crate::Result<Self::Stream>>;
 }
diff --git a/p2p/src/net/mock/types.rs b/p2p/src/net/mock/types.rs
index e62479690..c756a8ad2 100644
--- a/p2p/src/net/mock/types.rs
+++ b/p2p/src/net/mock/types.rs
@@ -42,19 +42,17 @@ pub enum Command<T: TransportSocket> {
     },
     SendRequest {
         peer_id: MockPeerId,
+        request_id: MockRequestId,
         message: message::Request,
-        response: oneshot::Sender<crate::Result<MockRequestId>>,
     },
     /// Send response to remote peer
     SendResponse {
         request_id: MockRequestId,
         message: message::Response,
-        response: oneshot::Sender<crate::Result<()>>,
     },
     AnnounceData {
         topic: PubSubTopic,
         message: Vec<u8>,
-        response: oneshot::Sender<crate::Result<()>>,
     },
 }
 
@@ -62,12 +60,12 @@ pub enum SyncingEvent {
     Request {
         peer_id: MockPeerId,
         request_id: MockRequestId,
-        request: message::Request,
+        request: message::SyncRequest,
     },
     Response {
         peer_id: MockPeerId,
         request_id: MockRequestId,
-        response: message::Response,
+        response: message::SyncResponse,
     },
     Announcement {
         peer_id: MockPeerId,
@@ -77,13 +75,25 @@ pub enum SyncingEvent {
 
 #[derive(Debug, PartialEq, Eq)]
 pub enum ConnectivityEvent<T: TransportSocket> {
+    Request {
+        peer_id: MockPeerId,
+        request_id: MockRequestId,
+        request: message::PeerManagerRequest,
+    },
+    Response {
+        peer_id: MockPeerId,
+        request_id: MockRequestId,
+        response: message::PeerManagerResponse,
+    },
     InboundAccepted {
         address: T::Address,
         peer_info: MockPeerInfo,
+        receiver_address: Option<PeerAddress>,
     },
     OutboundAccepted {
         address: T::Address,
         peer_info: MockPeerInfo,
+        receiver_address: Option<PeerAddress>,
     },
     ConnectionError {
         address: T::Address,
@@ -112,18 +122,14 @@ pub enum PubSubEvent<T: TransportSocket> {
     },
 }
 
+static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1);
+
 #[derive(Copy, Clone, PartialEq, Eq, Hash, Debug, Encode, Decode, Default)]
 pub struct MockRequestId(u64);
 
 impl MockRequestId {
-    pub fn new(request_id: u64) -> Self {
-        Self(request_id)
-    }
-
-    pub fn fetch_and_inc(&mut self) -> Self {
-        let id = self.0;
-        self.0 += 1;
-
+    pub fn new() -> Self {
+        let id = NEXT_REQUEST_ID.fetch_add(1, Ordering::Relaxed);
         Self(id)
     }
 }
diff --git a/p2p/src/net/mod.rs b/p2p/src/net/mod.rs
index 58b776d41..349eec33a 100644
--- a/p2p/src/net/mod.rs
+++ b/p2p/src/net/mod.rs
@@ -25,7 +25,13 @@ use std::{
 
 use async_trait::async_trait;
 
-use crate::{config, message, message::Announcement};
+use crate::{
+    config,
+    message::{Announcement, PeerManagerRequest},
+    message::{PeerManagerResponse, SyncRequest, SyncResponse},
+};
+
+use self::mock::transport::TransportAddress;
 
 /// [NetworkingService] provides the low-level network interface
 /// that each network service provider must implement
@@ -52,6 +58,7 @@ pub trait NetworkingService {
         + Sync
         + ToString
         + FromStr
+        + TransportAddress
         + AsBannableAddress<BannableAddress = Self::BannableAddress>;
 
     /// An address type that can be banned.
@@ -64,7 +71,7 @@ pub trait NetworkingService {
     type PeerId: Copy + Debug + Display + Eq + Hash + Send + Sync + ToString + FromStr;
 
     /// Unique ID assigned to each received request from a peer
-    type SyncingPeerRequestId: Copy + Debug + Eq + Hash + Send + Sync;
+    type PeerRequestId: Copy + Debug + Eq + Hash + Send + Sync;
 
     /// Handle for sending/receiving connectivity-related events
     type ConnectivityHandle: Send;
@@ -108,6 +115,28 @@ where
     /// `peer_id` - Peer ID of the remote node
     async fn disconnect(&mut self, peer_id: T::PeerId) -> crate::Result<()>;
 
+    /// Send PeerManager's request to remote
+    ///
+    /// # Arguments
+    /// * `peer_id` - Unique ID of the peer the request is sent to
+    /// * `request` - Request to be sent
+    async fn send_request(
+        &mut self,
+        peer_id: T::PeerId,
+        request: PeerManagerRequest,
+    ) -> crate::Result<T::PeerRequestId>;
+
+    /// Send response to remote
+    ///
+    /// # Arguments
+    /// * `request_id` - ID of the request this is a response to
+    /// * `response` - Response to be sent
+    async fn send_response(
+        &mut self,
+        request_id: T::PeerRequestId,
+        response: PeerManagerResponse,
+    ) -> crate::Result<()>;
+
     /// Return the socket address of the network service provider
     ///
     /// If the address isn't available yet, `None` is returned
@@ -137,18 +166,18 @@ where
     async fn send_request(
         &mut self,
         peer_id: T::PeerId,
-        request: message::Request,
-    ) -> crate::Result<T::SyncingPeerRequestId>;
+        request: SyncRequest,
+    ) -> crate::Result<T::PeerRequestId>;
 
     /// Send block/header response to remote
     ///
     /// # Arguments
     /// * `request_id` - ID of the request this is a response to
-    /// * `message` - Response to be sent
+    /// * `response` - Response to be sent
     async fn send_response(
         &mut self,
-        request_id: T::SyncingPeerRequestId,
-        response: message::Response,
+        request_id: T::PeerRequestId,
+        response: SyncResponse,
     ) -> crate::Result<()>;
 
     /// Publishes an announcement on the network.
diff --git a/p2p/src/net/types/mod.rs b/p2p/src/net/types/mod.rs
index 3bad7f91d..cf7e00f4b 100644
--- a/p2p/src/net/types/mod.rs
+++ b/p2p/src/net/types/mod.rs
@@ -18,7 +18,7 @@ use std::{collections::BTreeSet, fmt::Display};
 use common::primitives::semver::SemVer;
 use serialization::{Decode, Encode};
 
-use crate::{message, NetworkingService, P2pError};
+use crate::{message, types::peer_address::PeerAddress, NetworkingService, P2pError};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum Role {
@@ -72,6 +72,28 @@ impl<T: NetworkingService> Display for PeerInfo<T> {
 /// Connectivity-related events received from the network
 #[derive(Debug)]
 pub enum ConnectivityEvent<T: NetworkingService> {
+    /// An incoming request.
+    Request {
+        /// Unique ID of the sender
+        peer_id: T::PeerId,
+
+        /// Unique ID of the request
+        request_id: T::PeerRequestId,
+
+        /// Received request
+        request: message::PeerManagerRequest,
+    },
+    /// An incoming response.
+    Response {
+        /// Unique ID of the sender
+        peer_id: T::PeerId,
+
+        /// Unique ID of the request this message is a response to
+        request_id: T::PeerRequestId,
+
+        /// Received response
+        response: message::PeerManagerResponse,
+    },
     /// Outbound connection accepted
     OutboundAccepted {
         /// Peer address
@@ -79,6 +101,9 @@ pub enum ConnectivityEvent<T: NetworkingService> {
 
         /// Peer information
         peer_info: PeerInfo<T>,
+
+        /// Socket address of this node as seen by remote peer
+        receiver_address: Option<PeerAddress>,
     },
 
     /// Inbound connection received
@@ -88,6 +113,9 @@ pub enum ConnectivityEvent<T: NetworkingService> {
 
         /// Peer information
         peer_info: PeerInfo<T>,
+
+        /// Socket address of this node as seen by remote peer
+        receiver_address: Option<PeerAddress>,
     },
 
     /// Outbound connection failed
@@ -105,16 +133,10 @@ pub enum ConnectivityEvent<T: NetworkingService> {
         peer_id: T::PeerId,
     },
 
-    /// One or more peers discovered
-    Discovered {
-        /// Address information
-        addresses: Vec<T::Address>,
-    },
-
-    /// One one more peers have expired (libp2p defines expired addresses as addresses that haven't appeared in later refreshes of available addresses)
-    Expired {
+    /// New peer discovered
+    AddressDiscovered {
         /// Address information
-        addresses: Vec<T::Address>,
+        address: T::Address,
     },
 
     /// Protocol violation
@@ -136,10 +158,10 @@ pub enum SyncingEvent<T: NetworkingService> {
         peer_id: T::PeerId,
 
         /// Unique ID of the request
-        request_id: T::SyncingPeerRequestId,
+        request_id: T::PeerRequestId,
 
         /// Received request
-        request: message::Request,
+        request: message::SyncRequest,
     },
     /// An incoming response.
     Response {
@@ -147,10 +169,10 @@ pub enum SyncingEvent<T: NetworkingService> {
         peer_id: T::PeerId,
 
         /// Unique ID of the request this message is a response to
-        request_id: T::SyncingPeerRequestId,
+        request_id: T::PeerRequestId,
 
         /// Received response
-        response: message::Response,
+        response: message::SyncResponse,
     },
     /// An announcement that is broadcast to all peers.
     Announcement {
diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs
index 80b4bc847..f99b0be8f 100644
--- a/p2p/src/peer_manager/mod.rs
+++ b/p2p/src/peer_manager/mod.rs
@@ -40,7 +40,12 @@ use crate::{
     config::P2pConfig,
     error::{P2pError, PeerError, ProtocolError},
     event::{PeerManagerEvent, SyncControlEvent},
-    net::{self, types::Role, AsBannableAddress, ConnectivityService, NetworkingService},
+    message::{AddrListRequest, AddrListResponse, PeerManagerRequest, PeerManagerResponse},
+    net::{
+        self, mock::transport::TransportAddress, types::Role, AsBannableAddress,
+        ConnectivityService, NetworkingService,
+    },
+    types::peer_address::PeerAddress,
 };
 
 /// Maximum number of connections the [`PeerManager`] is allowed to have open
@@ -104,17 +109,8 @@ where
     }
 
     /// Update the list of known peers or known peer's list of addresses
-    fn peer_discovered(&mut self, addresses: &[T::Address]) {
-        addresses.iter().for_each(|peer| {
-            self.peerdb.peer_discovered(peer);
-        })
-    }
-
-    /// Update the list of known peers or known peer's list of addresses
-    fn peer_expired(&mut self, addresses: &[T::Address]) {
-        addresses.iter().for_each(|peer| {
-            self.peerdb.peer_expired(peer);
-        })
+    fn peer_discovered(&mut self, address: &T::Address) {
+        self.peerdb.peer_discovered(address);
     }
 
     /// Verify software version compatibility
@@ -135,7 +131,10 @@ where
         address: T::Address,
         role: Role,
         info: net::types::PeerInfo<T>,
+        _receiver_address: Option<PeerAddress>,
     ) -> crate::Result<()> {
+        // TODO: Handle receiver_address
+
         let peer_id = info.peer_id;
 
         ensure!(
@@ -179,6 +178,7 @@ where
         &mut self,
         address: T::Address,
         info: net::types::PeerInfo<T>,
+        receiver_address: Option<PeerAddress>,
     ) -> crate::Result<()> {
         log::debug!("validate inbound connection, inbound address {address:?}");
 
@@ -201,7 +201,7 @@ where
             return Err(P2pError::PeerError(PeerError::TooManyPeers));
         }
 
-        self.accept_connection(address, Role::Inbound, info)
+        self.accept_connection(address, Role::Inbound, info, receiver_address)
     }
 
     /// Close connection to a remote peer
@@ -322,6 +322,46 @@ where
         Ok(())
     }
 
+    async fn handle_incoming_request(
+        &mut self,
+        _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();
+
+                self.peer_connectivity_handle
+                    .send_response(
+                        request_id,
+                        PeerManagerResponse::AddrListResponse(AddrListResponse::new(addresses)),
+                    )
+                    .await
+            }
+        }
+    }
+
+    fn handle_incoming_response(
+        &mut self,
+        _peer_id: T::PeerId,
+        _request_id: T::PeerRequestId,
+        response: PeerManagerResponse,
+    ) -> crate::Result<()> {
+        match response {
+            // TODO: Rework this
+            PeerManagerResponse::AddrListResponse(response) => {
+                for address in response.addresses() {
+                    if let Some(address) = TransportAddress::from_peer_address(address) {
+                        self.peerdb.peer_discovered(&address);
+                    }
+                }
+                Ok(())
+            }
+        }
+    }
+
     /// Handle the result of a control/network event
     ///
     /// Currently only subsystem/channel-related errors are considered fatal.
@@ -329,7 +369,7 @@ where
     /// 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 scored crossed the ban threshold.
+    /// to that peer is possibly closed if their score crossed the ban threshold.
     ///
     /// # Arguments
     /// `peer_id` - peer ID of the remote peer, if available
@@ -416,10 +456,16 @@ where
                 },
                 event = self.peer_connectivity_handle.poll_next() => match event {
                     Ok(event) => match event {
-                        net::types::ConnectivityEvent::InboundAccepted { address, peer_info } => {
+                        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) {
+                            match self.accept_inbound_connection(address, peer_info, receiver_address) {
                                 Ok(_) => {},
                                 Err(P2pError::ChannelClosed) => return Err(P2pError::ChannelClosed),
                                 Err(P2pError::PeerError(err)) => {
@@ -435,9 +481,9 @@ where
                                 }
                             }
                         }
-                        net::types::ConnectivityEvent::OutboundAccepted { address, peer_info } => {
+                        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);
+                            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) {
@@ -454,11 +500,8 @@ where
                             let res = self.handle_outbound_error(address, error);
                             self.handle_result(None, res).await?;
                         }
-                        net::types::ConnectivityEvent::Discovered { addresses } => {
-                            self.peer_discovered(&addresses);
-                        }
-                        net::types::ConnectivityEvent::Expired { addresses } => {
-                            self.peer_expired(&addresses);
+                        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;
diff --git a/p2p/src/peer_manager/peerdb/mod.rs b/p2p/src/peer_manager/peerdb/mod.rs
index aef11b2a4..fb9c6e7ad 100644
--- a/p2p/src/peer_manager/peerdb/mod.rs
+++ b/p2p/src/peer_manager/peerdb/mod.rs
@@ -38,9 +38,11 @@ use crate::{
     error::{ConversionError, P2pError},
     interface::types::ConnectedPeer,
     net::{
+        mock::transport::TransportAddress,
         types::{self, Role},
         AsBannableAddress, NetworkingService,
     },
+    types::peer_address::PeerAddress,
 };
 
 #[derive(Debug)]
@@ -132,6 +134,10 @@ impl<T: NetworkingService> PeerDb<T> {
         self.addresses.contains(address)
     }
 
+    pub fn known_addresses(&self) -> Vec<PeerAddress> {
+        self.addresses.iter().map(TransportAddress::as_peer_address).collect()
+    }
+
     /// Checks if the given address is banned.
     pub fn is_address_banned(&mut self, address: &T::BannableAddress) -> bool {
         if let Some(banned_till) = self.banned.get(address) {
@@ -170,11 +176,6 @@ impl<T: NetworkingService> PeerDb<T> {
         self.available.insert(address.clone());
     }
 
-    /// Expire discovered peer addresses
-    pub fn peer_expired(&mut self, address: &T::Address) {
-        self.available.remove(address);
-    }
-
     /// Report outbound connection failure
     ///
     /// When [`crate::peer_manager::PeerManager::heartbeat()`] has initiated an outbound connection
diff --git a/p2p/src/sync/mod.rs b/p2p/src/sync/mod.rs
index a7a704448..7bb55d4e3 100644
--- a/p2p/src/sync/mod.rs
+++ b/p2p/src/sync/mod.rs
@@ -40,7 +40,7 @@ use crate::{
     config::P2pConfig,
     error::{P2pError, PeerError, ProtocolError},
     event::{PeerManagerEvent, SyncControlEvent},
-    message::{self, Announcement},
+    message::{self, Announcement, SyncRequest},
     net::{types::SyncingEvent, NetworkingService, SyncingMessagingService},
 };
 
@@ -86,7 +86,7 @@ impl<T> BlockSyncManager<T>
 where
     T: NetworkingService,
     T::SyncingMessagingHandle: SyncingMessagingService<T>,
-    T::SyncingPeerRequestId: 'static,
+    T::PeerRequestId: 'static,
     T::PeerId: 'static,
 {
     pub fn new(
@@ -127,7 +127,7 @@ where
 
         self.send_request(
             peer_id,
-            message::Request::HeaderListRequest(message::HeaderListRequest::new(locator.clone())),
+            SyncRequest::HeaderListRequest(message::HeaderListRequest::new(locator.clone())),
         )
         .await
         .map(|_| {
@@ -147,7 +147,7 @@ where
     pub async fn process_header_request(
         &mut self,
         peer_id: T::PeerId,
-        request_id: T::SyncingPeerRequestId,
+        request_id: T::PeerRequestId,
         locator: Locator,
     ) -> crate::Result<()> {
         log::debug!("send header response to peer {peer_id}, request_id: {request_id:?}");
@@ -161,7 +161,7 @@ where
     pub async fn process_block_request(
         &mut self,
         peer_id: T::PeerId,
-        request_id: T::SyncingPeerRequestId,
+        request_id: T::PeerRequestId,
         headers: Vec<Id<Block>>,
     ) -> crate::Result<()> {
         ensure!(
@@ -328,18 +328,18 @@ where
     pub async fn process_response(
         &mut self,
         peer_id: T::PeerId,
-        request_id: T::SyncingPeerRequestId,
-        response: message::Response,
+        request_id: T::PeerRequestId,
+        response: message::SyncResponse,
     ) -> crate::Result<()> {
         match response {
-            message::Response::HeaderListResponse(response) => {
+            message::SyncResponse::HeaderListResponse(response) => {
                 log::debug!("process header response (id {request_id:?}) from peer {peer_id}");
                 log::trace!("received headers: {:#?}", response.headers());
 
                 let result = self.process_header_response(peer_id, response.into_headers()).await;
                 self.handle_error(peer_id, result).await?;
             }
-            message::Response::BlockListResponse(response) => {
+            message::SyncResponse::BlockListResponse(response) => {
                 log::debug!("process block response (id {request_id:?}) from peer {peer_id}");
                 log::trace!(
                     "# of received blocks: {}, block ids: {:#?}",
@@ -450,7 +450,7 @@ where
                         request_id,
                         request,
                     } => match request {
-                        message::Request::HeaderListRequest(request) => {
+                        message::SyncRequest::HeaderListRequest(request) => {
                             log::debug!("process header request (id {request_id:?}) from peer {peer_id}");
                             log::trace!("locator: {:#?}", request.locator());
 
@@ -461,7 +461,7 @@ where
                             ).await;
                             self.handle_error(peer_id, result).await?;
                         }
-                        message::Request::BlockListRequest(request) => {
+                        message::SyncRequest::BlockListRequest(request) => {
                             log::debug!("process block request (id {request_id:?}) from peer {peer_id}");
                             log::trace!("requested block ids: {:#?}", request.block_ids());
 
diff --git a/p2p/src/sync/request.rs b/p2p/src/sync/request.rs
index cd050dba5..537088e2a 100644
--- a/p2p/src/sync/request.rs
+++ b/p2p/src/sync/request.rs
@@ -25,7 +25,7 @@ use utils::ensure;
 
 use crate::{
     error::{P2pError, PeerError},
-    message,
+    message::{self, SyncRequest, SyncResponse},
     sync::{peer::PeerSyncState, BlockSyncManager},
     NetworkingService, SyncingMessagingService,
 };
@@ -34,40 +34,40 @@ impl<T> BlockSyncManager<T>
 where
     T: NetworkingService,
     T::SyncingMessagingHandle: SyncingMessagingService<T>,
-    T::SyncingPeerRequestId: 'static,
+    T::PeerRequestId: 'static,
     T::PeerId: 'static,
 {
     /// Creates a blocks request message.
-    pub fn make_block_request(&self, block_ids: Vec<Id<Block>>) -> message::Request {
-        message::Request::BlockListRequest(message::BlockListRequest::new(block_ids))
+    pub fn make_block_request(&self, block_ids: Vec<Id<Block>>) -> SyncRequest {
+        SyncRequest::BlockListRequest(message::BlockListRequest::new(block_ids))
     }
 
     /// Creates a headers request message with the given locator.
-    pub fn make_header_request(&self, locator: Locator) -> message::Request {
-        message::Request::HeaderListRequest(message::HeaderListRequest::new(locator))
+    pub fn make_header_request(&self, locator: Locator) -> SyncRequest {
+        SyncRequest::HeaderListRequest(message::HeaderListRequest::new(locator))
     }
 
     /// Make header response
     ///
     /// # Arguments
     /// * `headers` - the headers that were requested
-    pub fn make_header_response(&self, headers: Vec<BlockHeader>) -> message::Response {
-        message::Response::HeaderListResponse(message::HeaderListResponse::new(headers))
+    pub fn make_header_response(&self, headers: Vec<BlockHeader>) -> SyncResponse {
+        SyncResponse::HeaderListResponse(message::HeaderListResponse::new(headers))
     }
 
     /// Make block response
     ///
     /// # Arguments
     /// * `blocks` - the blocks that were requested
-    pub fn make_block_response(&self, blocks: Vec<Block>) -> message::Response {
-        message::Response::BlockListResponse(message::BlockListResponse::new(blocks))
+    pub fn make_block_response(&self, blocks: Vec<Block>) -> SyncResponse {
+        SyncResponse::BlockListResponse(message::BlockListResponse::new(blocks))
     }
 
     /// Sends a request to the given peer.
     pub async fn send_request(
         &mut self,
         peer_id: T::PeerId,
-        request: message::Request,
+        request: SyncRequest,
     ) -> crate::Result<()> {
         self.peer_sync_handle.send_request(peer_id, request).await.map(|_| ())
     }
@@ -146,7 +146,7 @@ where
     /// * `headers` - headers that the remote requested
     pub async fn send_header_response(
         &mut self,
-        request_id: T::SyncingPeerRequestId,
+        request_id: T::PeerRequestId,
         headers: Vec<BlockHeader>,
     ) -> crate::Result<()> {
         log::trace!("send header response, request id {request_id:?}");
@@ -166,7 +166,7 @@ where
     /// * `headers` - headers that the remote requested
     pub async fn send_block_response(
         &mut self,
-        request_id: T::SyncingPeerRequestId,
+        request_id: T::PeerRequestId,
         blocks: Vec<Block>,
     ) -> crate::Result<()> {
         log::trace!("send block response, request id {request_id:?}");


Test patch:
diff --git a/p2p/backend-test-suite/src/ban.rs b/p2p/backend-test-suite/src/ban.rs
index 90ef4e61f..593697665 100644
--- a/p2p/backend-test-suite/src/ban.rs
+++ b/p2p/backend-test-suite/src/ban.rs
@@ -19,9 +19,8 @@ use tokio::sync::mpsc;
 
 use p2p::{
     config::P2pConfig,
-    error::{P2pError, PublishError},
     event::PeerManagerEvent,
-    message::{Announcement, HeaderListResponse, Request, Response},
+    message::{Announcement, HeaderListResponse, SyncRequest, SyncResponse},
     net::{types::SyncingEvent, ConnectivityService, NetworkingService, SyncingMessagingService},
     sync::BlockSyncManager,
     testing_utils::{connect_services, TestTransportMaker},
@@ -90,30 +89,19 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(_),
+                request: SyncRequest::HeaderListRequest(_),
             } => request_id,
             e => panic!("Unexpected event type: {e:?}"),
         };
         sync2
             .send_response(
                 request_id,
-                Response::HeaderListResponse(HeaderListResponse::new(Vec::new())),
+                SyncResponse::HeaderListResponse(HeaderListResponse::new(Vec::new())),
             )
             .await
             .unwrap();
 
-        loop {
-            let res = sync2.make_announcement(Announcement::Block(blocks[2].clone())).await;
-
-            if res.is_ok() {
-                break;
-            } else {
-                assert_eq!(
-                    res,
-                    Err(P2pError::PublishError(PublishError::InsufficientPeers))
-                );
-            }
-        }
+        sync2.make_announcement(Announcement::Block(blocks[2].clone())).await.unwrap();
     });
 
     match rx_peer_manager.recv().await {
diff --git a/p2p/backend-test-suite/src/block_announcement.rs b/p2p/backend-test-suite/src/block_announcement.rs
index 31fae6129..868e7977e 100644
--- a/p2p/backend-test-suite/src/block_announcement.rs
+++ b/p2p/backend-test-suite/src/block_announcement.rs
@@ -15,8 +15,6 @@
 
 use std::{fmt::Debug, sync::Arc};
 
-use tokio::{pin, select, time::Duration};
-
 use common::{
     chain::{
         block::{consensus_data::ConsensusData, timestamp::BlockTimestamp, Block, BlockReward},
@@ -76,26 +74,19 @@ where
 
     connect_services::<S>(&mut conn1, &mut conn2).await;
 
-    // Spam the message until until we have a peer.
-    loop {
-        let res = sync1
-            .make_announcement(Announcement::Block(
-                Block::new(
-                    vec![],
-                    Id::new(H256([0x01; 32])),
-                    BlockTimestamp::from_int_seconds(1337u64),
-                    ConsensusData::None,
-                    BlockReward::new(Vec::new()),
-                )
-                .unwrap(),
-            ))
-            .await;
-
-        match res {
-            Ok(()) => break,
-            Err(e) => assert_eq!(e, P2pError::PublishError(PublishError::InsufficientPeers)),
-        }
-    }
+    sync1
+        .make_announcement(Announcement::Block(
+            Block::new(
+                vec![],
+                Id::new(H256([0x01; 32])),
+                BlockTimestamp::from_int_seconds(1337u64),
+                ConsensusData::None,
+                BlockReward::new(Vec::new()),
+            )
+            .unwrap(),
+        ))
+        .await
+        .unwrap();
 
     // Poll an event from the network for server2.
     let block = match sync2.poll_next().await.unwrap() {
@@ -165,25 +156,19 @@ where
 
     connect_services::<S>(&mut conn1, &mut conn2).await;
 
-    let timeout = tokio::time::sleep(Duration::from_secs(1));
-    pin!(timeout);
-    loop {
-        select! {
-            res = sync1.make_announcement(Announcement::Block(
-                Block::new(
-                    vec![],
-                    Id::new(H256([0x01; 32])),
-                    BlockTimestamp::from_int_seconds(1337u64),
-                    ConsensusData::None,
-                    BlockReward::new(Vec::new()),
-                )
-                .unwrap(),
-            )) => {
-                assert_eq!(Err(P2pError::PublishError(PublishError::InsufficientPeers)), res);
-            }
-            _ = &mut timeout => break,
-        }
-    }
+    sync1
+        .make_announcement(Announcement::Block(
+            Block::new(
+                vec![],
+                Id::new(H256([0x01; 32])),
+                BlockTimestamp::from_int_seconds(1337u64),
+                ConsensusData::None,
+                BlockReward::new(Vec::new()),
+            )
+            .unwrap(),
+        ))
+        .await
+        .unwrap();
 }
 
 async fn block_announcement_too_big_message<T, S, A>()
diff --git a/p2p/backend-test-suite/src/sync.rs b/p2p/backend-test-suite/src/sync.rs
index f2a6147ad..ae0fea7f2 100644
--- a/p2p/backend-test-suite/src/sync.rs
+++ b/p2p/backend-test-suite/src/sync.rs
@@ -32,7 +32,7 @@ use p2p::{
     config::P2pConfig,
     error::P2pError,
     event::{PeerManagerEvent, SyncControlEvent},
-    message::{BlockListRequest, BlockListResponse, HeaderListResponse, Request, Response},
+    message::{BlockListRequest, BlockListResponse, HeaderListResponse, SyncRequest, SyncResponse},
     net::{
         types::{ConnectivityEvent, SyncingEvent},
         ConnectivityService, NetworkingService, SyncingMessagingService,
@@ -151,7 +151,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(request),
+                request: SyncRequest::HeaderListRequest(request),
             } => {
                 let headers = mgr2_handle
                     .call(move |this| this.get_headers(request.into_locator()))
@@ -161,7 +161,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::HeaderListResponse(HeaderListResponse::new(headers)),
+                        SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
                     .await
                     .unwrap()
@@ -169,7 +169,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::BlockListRequest(request),
+                request: SyncRequest::BlockListRequest(request),
             } => {
                 assert_eq!(request.block_ids().len(), 1);
                 let id = request.block_ids()[0];
@@ -182,7 +182,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::BlockListResponse(BlockListResponse::new(blocks)),
+                        SyncResponse::BlockListResponse(BlockListResponse::new(blocks)),
                     )
                     .await
                     .unwrap();
@@ -190,7 +190,7 @@ where
             SyncingEvent::Response {
                 peer_id: _,
                 request_id: _,
-                response: Response::HeaderListResponse(_response),
+                response: SyncResponse::HeaderListResponse(_response),
             } => {}
             msg => panic!("invalid message received: {:?}", msg),
         }
@@ -248,7 +248,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(request),
+                request: SyncRequest::HeaderListRequest(request),
             } => {
                 let headers = mgr2_handle
                     .call(move |this| this.get_headers(request.into_locator()))
@@ -258,7 +258,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::HeaderListResponse(HeaderListResponse::new(headers)),
+                        SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
                     .await
                     .unwrap()
@@ -266,7 +266,7 @@ where
             SyncingEvent::Response {
                 peer_id,
                 request_id: _,
-                response: Response::BlockListResponse(response),
+                response: SyncResponse::BlockListResponse(response),
             } => {
                 assert_eq!(response.blocks().len(), 1);
                 let block = response.blocks()[0].clone();
@@ -280,7 +280,7 @@ where
                     mgr2.handle_mut()
                         .send_request(
                             peer_id,
-                            Request::BlockListRequest(BlockListRequest::new(vec![header])),
+                            SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                         )
                         .await
                         .unwrap();
@@ -292,7 +292,7 @@ where
             SyncingEvent::Response {
                 peer_id,
                 request_id: _,
-                response: Response::HeaderListResponse(response),
+                response: SyncResponse::HeaderListResponse(response),
             } => {
                 assert_eq!(response.headers().len(), 12);
                 let headers = mgr2_handle
@@ -305,7 +305,7 @@ where
                 mgr2.handle_mut()
                     .send_request(
                         peer_id,
-                        Request::BlockListRequest(BlockListRequest::new(vec![header])),
+                        SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                     )
                     .await
                     .unwrap();
@@ -373,7 +373,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(request),
+                request: SyncRequest::HeaderListRequest(request),
             } => {
                 let headers = mgr2_handle
                     .call(move |this| this.get_headers(request.into_locator()))
@@ -383,7 +383,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::HeaderListResponse(HeaderListResponse::new(headers)),
+                        SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
                     .await
                     .unwrap()
@@ -391,7 +391,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::BlockListRequest(request),
+                request: SyncRequest::BlockListRequest(request),
             } => {
                 assert_eq!(request.block_ids().len(), 1);
                 let id = request.block_ids()[0];
@@ -404,7 +404,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::BlockListResponse(BlockListResponse::new(blocks)),
+                        SyncResponse::BlockListResponse(BlockListResponse::new(blocks)),
                     )
                     .await
                     .unwrap();
@@ -412,7 +412,7 @@ where
             SyncingEvent::Response {
                 peer_id,
                 request_id: _,
-                response: Response::BlockListResponse(response),
+                response: SyncResponse::BlockListResponse(response),
             } => {
                 assert_eq!(response.blocks().len(), 1);
                 let block = response.blocks()[0].clone();
@@ -426,7 +426,7 @@ where
                     mgr2.handle_mut()
                         .send_request(
                             peer_id,
-                            Request::BlockListRequest(BlockListRequest::new(vec![header])),
+                            SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                         )
                         .await
                         .unwrap();
@@ -435,7 +435,7 @@ where
             SyncingEvent::Response {
                 peer_id,
                 request_id: _,
-                response: Response::HeaderListResponse(response),
+                response: SyncResponse::HeaderListResponse(response),
             } => {
                 let headers = mgr2_handle
                     .call(move |this| this.filter_already_existing_blocks(response.into_headers()))
@@ -447,7 +447,7 @@ where
                 mgr2.handle_mut()
                     .send_request(
                         peer_id,
-                        Request::BlockListRequest(BlockListRequest::new(vec![header])),
+                        SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                     )
                     .await
                     .unwrap();
@@ -517,7 +517,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(request),
+                request: SyncRequest::HeaderListRequest(request),
             } => {
                 let headers = mgr2_handle
                     .call(move |this| this.get_headers(request.into_locator()))
@@ -527,7 +527,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::HeaderListResponse(HeaderListResponse::new(headers)),
+                        SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
                     .await
                     .unwrap()
@@ -535,7 +535,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::BlockListRequest(request),
+                request: SyncRequest::BlockListRequest(request),
             } => {
                 assert_eq!(request.block_ids().len(), 1);
                 let id = request.block_ids()[0];
@@ -548,7 +548,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::BlockListResponse(BlockListResponse::new(blocks)),
+                        SyncResponse::BlockListResponse(BlockListResponse::new(blocks)),
                     )
                     .await
                     .unwrap();
@@ -556,7 +556,7 @@ where
             SyncingEvent::Response {
                 peer_id,
                 request_id: _,
-                response: Response::BlockListResponse(response),
+                response: SyncResponse::BlockListResponse(response),
             } => {
                 assert_eq!(response.blocks().len(), 1);
                 let block = response.blocks()[0].clone();
@@ -570,7 +570,7 @@ where
                     mgr2.handle_mut()
                         .send_request(
                             peer_id,
-                            Request::BlockListRequest(BlockListRequest::new(vec![header])),
+                            SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                         )
                         .await
                         .unwrap();
@@ -579,7 +579,7 @@ where
             SyncingEvent::Response {
                 peer_id,
                 request_id: _,
-                response: Response::HeaderListResponse(response),
+                response: SyncResponse::HeaderListResponse(response),
             } => {
                 let headers = mgr2_handle
                     .call(move |this| this.filter_already_existing_blocks(response.into_headers()))
@@ -591,7 +591,7 @@ where
                 mgr2.handle_mut()
                     .send_request(
                         peer_id,
-                        Request::BlockListRequest(BlockListRequest::new(vec![header])),
+                        SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                     )
                     .await
                     .unwrap();
@@ -678,14 +678,14 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(request),
+                request: SyncRequest::HeaderListRequest(request),
             } => {
                 let headers = mgr_handle
                     .call(move |this| this.get_headers(request.into_locator()))
                     .await
                     .unwrap()
                     .unwrap();
-                let msg = Response::HeaderListResponse(HeaderListResponse::new(headers));
+                let msg = SyncResponse::HeaderListResponse(HeaderListResponse::new(headers));
 
                 if dest_peer_id == peer_info21.peer_id {
                     mgr2.handle_mut().send_response(request_id, msg).await.unwrap()
@@ -696,11 +696,11 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::BlockListRequest(request),
+                request: SyncRequest::BlockListRequest(request),
             } => {
                 assert_eq!(request.block_ids().len(), 1);
                 let id = request.block_ids()[0];
-                let msg = Response::BlockListResponse(BlockListResponse::new(vec![mgr_handle
+                let msg = SyncResponse::BlockListResponse(BlockListResponse::new(vec![mgr_handle
                     .call(move |this| this.get_block(id))
                     .await
                     .unwrap()
@@ -716,7 +716,7 @@ where
             SyncingEvent::Response {
                 peer_id: _,
                 request_id: _,
-                response: Response::HeaderListResponse(_response),
+                response: SyncResponse::HeaderListResponse(_response),
             } => {}
             msg => panic!("invalid message received: {:?}", msg),
         }
@@ -817,14 +817,14 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(request),
+                request: SyncRequest::HeaderListRequest(request),
             } => {
                 let headers = mgr_handle
                     .call(move |this| this.get_headers(request.into_locator()))
                     .await
                     .unwrap()
                     .unwrap();
-                let msg = Response::HeaderListResponse(HeaderListResponse::new(headers));
+                let msg = SyncResponse::HeaderListResponse(HeaderListResponse::new(headers));
 
                 if dest_peer_id == peer_info21.peer_id {
                     mgr2.handle_mut().send_response(request_id, msg).await.unwrap()
@@ -835,11 +835,11 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::BlockListRequest(request),
+                request: SyncRequest::BlockListRequest(request),
             } => {
                 assert_eq!(request.block_ids().len(), 1);
                 let id = request.block_ids()[0];
-                let msg = Response::BlockListResponse(BlockListResponse::new(vec![mgr_handle
+                let msg = SyncResponse::BlockListResponse(BlockListResponse::new(vec![mgr_handle
                     .call(move |this| this.get_block(id))
                     .await
                     .unwrap()
@@ -855,7 +855,7 @@ where
             SyncingEvent::Response {
                 peer_id: _,
                 request_id: _,
-                response: Response::HeaderListResponse(_response),
+                response: SyncResponse::HeaderListResponse(_response),
             } => {}
             msg => panic!("invalid message received: {:?}", msg),
         }
@@ -954,14 +954,14 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(request),
+                request: SyncRequest::HeaderListRequest(request),
             } => {
                 let headers = mgr_handle
                     .call(move |this| this.get_headers(request.into_locator()))
                     .await
                     .unwrap()
                     .unwrap();
-                let msg = Response::HeaderListResponse(HeaderListResponse::new(headers));
+                let msg = SyncResponse::HeaderListResponse(HeaderListResponse::new(headers));
 
                 if dest_peer_id == peer_info21.peer_id {
                     mgr2.handle_mut().send_response(request_id, msg).await.unwrap()
@@ -982,11 +982,11 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::BlockListRequest(request),
+                request: SyncRequest::BlockListRequest(request),
             } => {
                 assert_eq!(request.block_ids().len(), 1);
                 let id = request.block_ids()[0];
-                let msg = Response::BlockListResponse(BlockListResponse::new(vec![mgr_handle
+                let msg = SyncResponse::BlockListResponse(BlockListResponse::new(vec![mgr_handle
                     .call(move |this| this.get_block(id))
                     .await
                     .unwrap()
@@ -1002,7 +1002,7 @@ where
             SyncingEvent::Response {
                 peer_id: _,
                 request_id: _,
-                response: Response::HeaderListResponse(_response),
+                response: SyncResponse::HeaderListResponse(_response),
             } => {}
             msg => panic!("invalid message received: {:?}", msg),
         }
@@ -1064,7 +1064,7 @@ where
     assert_eq!(conn1.disconnect(peer_info2.peer_id).await, Ok(()));
 
     let event = filter_connectivity_event::<S, _>(&mut conn2, |event| {
-        !std::matches!(event, Ok(ConnectivityEvent::Discovered { .. }))
+        !std::matches!(event, Ok(ConnectivityEvent::AddressDiscovered { .. }))
     })
     .await;
     assert!(std::matches!(
@@ -1094,7 +1094,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::HeaderListRequest(request),
+                request: SyncRequest::HeaderListRequest(request),
             } => {
                 let headers = mgr2_handle
                     .call(move |this| this.get_headers(request.into_locator()))
@@ -1104,7 +1104,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::HeaderListResponse(HeaderListResponse::new(headers)),
+                        SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
                     .await
                     .unwrap()
@@ -1112,7 +1112,7 @@ where
             SyncingEvent::Request {
                 peer_id: _,
                 request_id,
-                request: Request::BlockListRequest(request),
+                request: SyncRequest::BlockListRequest(request),
             } => {
                 assert_eq!(request.block_ids().len(), 1);
                 let id = request.block_ids()[0];
@@ -1125,7 +1125,7 @@ where
                 mgr2.handle_mut()
                     .send_response(
                         request_id,
-                        Response::BlockListResponse(BlockListResponse::new(blocks)),
+                        SyncResponse::BlockListResponse(BlockListResponse::new(blocks)),
                     )
                     .await
                     .unwrap();
@@ -1133,7 +1133,7 @@ where
             SyncingEvent::Response {
                 peer_id: _,
                 request_id: _,
-                response: Response::HeaderListResponse(_response),
+                response: SyncResponse::HeaderListResponse(_response),
             } => {}
             msg => panic!("invalid message received: {:?}", msg),
         }
@@ -1160,7 +1160,7 @@ where
     T: NetworkingService,
     T::ConnectivityHandle: ConnectivityService<T>,
     T::SyncingMessagingHandle: SyncingMessagingService<T>,
-    T::SyncingPeerRequestId: 'static,
+    T::PeerRequestId: 'static,
     T::PeerId: 'static,
 {
     let (tx_p2p_sync, rx_p2p_sync) = mpsc::unbounded_channel();
@@ -1256,14 +1256,14 @@ async fn process_header_request<T>(
 where
     T: NetworkingService,
     T::SyncingMessagingHandle: SyncingMessagingService<T>,
-    T::SyncingPeerRequestId: 'static,
+    T::PeerRequestId: 'static,
     T::PeerId: 'static,
 {
     match mgr.handle_mut().poll_next().await.unwrap() {
         SyncingEvent::Request {
             peer_id: _,
             request_id,
-            request: Request::HeaderListRequest(request),
+            request: SyncRequest::HeaderListRequest(request),
         } => {
             let headers = handle
                 .call(move |this| this.get_headers(request.into_locator()))
@@ -1273,7 +1273,7 @@ where
             mgr.handle_mut()
                 .send_response(
                     request_id,
-                    Response::HeaderListResponse(HeaderListResponse::new(headers)),
+                    SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                 )
                 .await
         }
@@ -1285,7 +1285,7 @@ async fn advance_mgr_state<T>(mgr: &mut BlockSyncManager<T>) -> Result<(), P2pEr
 where
     T: NetworkingService,
     T::SyncingMessagingHandle: SyncingMessagingService<T>,
-    T::SyncingPeerRequestId: 'static,
+    T::PeerRequestId: 'static,
     T::PeerId: 'static,
 {
     match mgr.handle_mut().poll_next().await.unwrap() {
@@ -1294,10 +1294,10 @@ where
             request_id,
             request,
         } => match request {
-            Request::HeaderListRequest(request) => {
+            SyncRequest::HeaderListRequest(request) => {
                 mgr.process_header_request(peer_id, request_id, request.into_locator()).await?;
             }
-            Request::BlockListRequest(request) => {
+            SyncRequest::BlockListRequest(request) => {
                 mgr.process_block_request(peer_id, request_id, request.into_block_ids()).await?;
             }
         },
@@ -1306,10 +1306,10 @@ where
             request_id: _,
             response,
         } => match response {
-            Response::HeaderListResponse(response) => {
+            SyncResponse::HeaderListResponse(response) => {
                 mgr.process_header_response(peer_id, response.into_headers()).await?;
             }
-            Response::BlockListResponse(response) => {
+            SyncResponse::BlockListResponse(response) => {
                 mgr.process_block_response(peer_id, response.into_blocks()).await?;
             }
         },
diff --git a/p2p/src/net/mock/transport/impls/stream_adapter/wrapped_transport/tests.rs b/p2p/src/net/mock/transport/impls/stream_adapter/wrapped_transport/tests.rs
index 56b728468..e0c17d3d4 100644
--- a/p2p/src/net/mock/transport/impls/stream_adapter/wrapped_transport/tests.rs
+++ b/p2p/src/net/mock/transport/impls/stream_adapter/wrapped_transport/tests.rs
@@ -208,7 +208,7 @@ async fn send_2_reqs() {
     let server_stream = server_res.unwrap().0;
     let peer_stream = peer_res.unwrap();
 
-    let id_1 = MockRequestId::new(1337u64);
+    let id_1 = MockRequestId::new();
     let request = Request::BlockListRequest(BlockListRequest::new(vec![]));
     let mut peer_stream = BufferedTranscoder::new(peer_stream);
     peer_stream
@@ -219,7 +219,7 @@ async fn send_2_reqs() {
         .await
         .unwrap();
 
-    let id_2 = MockRequestId::new(1338u64);
+    let id_2 = MockRequestId::new();
     peer_stream
         .send(Message::Request {
             request_id: id_2,
diff --git a/p2p/src/peer_manager/tests/ban.rs b/p2p/src/peer_manager/tests/ban.rs
index 272305c24..a1cc45f53 100644
--- a/p2p/src/peer_manager/tests/ban.rs
+++ b/p2p/src/peer_manager/tests/ban.rs
@@ -60,7 +60,7 @@ where
     )
     .await;
     let peer_id = peer_info.peer_id;
-    pm2.accept_inbound_connection(address, peer_info).unwrap();
+    pm2.accept_inbound_connection(address, peer_info, None).unwrap();
 
     assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
     let addr1 = pm1.peer_connectivity_handle.local_addresses().await.unwrap()[0]
@@ -68,7 +68,10 @@ where
         .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::Discovered { .. }))
+        !std::matches!(
+            event,
+            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
+        )
     })
     .await;
     assert!(std::matches!(
@@ -111,7 +114,7 @@ where
     )
     .await;
     let peer_id = peer_info.peer_id;
-    pm2.accept_inbound_connection(address, peer_info).unwrap();
+    pm2.accept_inbound_connection(address, peer_info, None).unwrap();
 
     assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
     let addr1 = pm1.peer_connectivity_handle.local_addresses().await.unwrap()[0]
@@ -119,7 +122,10 @@ where
         .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::Discovered { .. }))
+        !std::matches!(
+            event,
+            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
+        )
     })
     .await;
     assert!(std::matches!(
@@ -164,7 +170,7 @@ where
     )
     .await;
     let peer_id = peer_info1.peer_id;
-    pm2.accept_inbound_connection(address, peer_info1).unwrap();
+    pm2.accept_inbound_connection(address, peer_info1, None).unwrap();
 
     assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
     let addr1 = pm1.peer_connectivity_handle.local_addresses().await.unwrap()[0]
@@ -172,7 +178,10 @@ where
         .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::Discovered { .. }))
+        !std::matches!(
+            event,
+            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
+        )
     })
     .await;
     assert!(matches!(
@@ -237,6 +246,7 @@ where
             agent: None,
             subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
         },
+        None,
     );
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
@@ -252,6 +262,7 @@ where
             agent: None,
             subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
         },
+        None,
     );
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
@@ -268,6 +279,7 @@ where
             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).await, Ok(()));
@@ -326,6 +338,7 @@ where
             agent: None,
             subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
         },
+        None,
     );
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
@@ -340,6 +353,7 @@ where
             agent: None,
             subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
         },
+        None,
     );
     assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
     assert!(!peer_manager.peerdb.is_active_peer(&peer_id));
@@ -355,6 +369,7 @@ where
             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).await, Ok(()));
@@ -425,7 +440,10 @@ where
     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::Discovered { .. }))
+        !std::matches!(
+            event,
+            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
+        )
     })
     .await;
     if let Ok(net::types::ConnectivityEvent::ConnectionClosed { peer_id }) = event {
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index 6b3faa45f..3d8ab0c64 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -105,7 +105,7 @@ where
     });
 
     // "discover" the other networking service
-    pm1.peer_discovered(&[addr]);
+    pm1.peer_discovered(&addr);
     pm1.heartbeat().await.unwrap();
 
     assert_eq!(pm1.pending.len(), 1);
@@ -227,7 +227,10 @@ where
         &mut pm2.peer_connectivity_handle,
     )
     .await;
-    assert_eq!(pm2.accept_inbound_connection(address, peer_info), Ok(()));
+    assert_eq!(
+        pm2.accept_inbound_connection(address, peer_info, None),
+        Ok(())
+    );
 }
 
 #[tokio::test]
@@ -274,7 +277,7 @@ where
     .await;
 
     assert_eq!(
-        pm2.accept_inbound_connection(address, peer_info),
+        pm2.accept_inbound_connection(address, peer_info, None),
         Err(P2pError::ProtocolError(ProtocolError::DifferentNetwork(
             [1, 2, 3, 4],
             *config::create_mainnet().magic_bytes(),
@@ -383,7 +386,10 @@ where
     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::Discovered { .. }))
+        !std::matches!(
+            event,
+            Ok(net::types::ConnectivityEvent::AddressDiscovered { .. })
+        )
     })
     .await;
     if let Ok(net::types::ConnectivityEvent::ConnectionClosed { peer_id }) = event {
diff --git a/p2p/src/sync/tests/mod.rs b/p2p/src/sync/tests/mod.rs
index d0d4458ff..be4309dc2 100644
--- a/p2p/src/sync/tests/mod.rs
+++ b/p2p/src/sync/tests/mod.rs
@@ -45,7 +45,7 @@ where
     T: NetworkingService,
     T::ConnectivityHandle: ConnectivityService<T>,
     T::SyncingMessagingHandle: SyncingMessagingService<T>,
-    T::SyncingPeerRequestId: 'static,
+    T::PeerRequestId: 'static,
     T::PeerId: 'static,
 {
     let (tx_p2p_sync, rx_p2p_sync) = mpsc::unbounded_channel();
diff --git a/p2p/src/sync/tests/request_response.rs b/p2p/src/sync/tests/request_response.rs
index 57ccee6b6..c8855029b 100644
--- a/p2p/src/sync/tests/request_response.rs
+++ b/p2p/src/sync/tests/request_response.rs
@@ -20,7 +20,7 @@ use tokio::time::timeout;
 use chainstate::Locator;
 
 use crate::{
-    message::{HeaderListRequest, HeaderListResponse, Request, Response},
+    message::{HeaderListRequest, HeaderListResponse, SyncRequest, SyncResponse},
     net::{
         mock::{
             transport::{MockChannelTransport, NoiseTcpTransport, TcpTransportSocket},
@@ -54,7 +54,7 @@ where
     mgr1.peer_sync_handle
         .send_request(
             peer_info2.peer_id,
-            Request::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
+            SyncRequest::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
         )
         .await
         .unwrap();
@@ -67,13 +67,13 @@ where
     {
         assert_eq!(
             request,
-            Request::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![])))
+            SyncRequest::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![])))
         );
 
         mgr2.peer_sync_handle
             .send_response(
                 request_id,
-                Response::HeaderListResponse(HeaderListResponse::new(vec![])),
+                SyncResponse::HeaderListResponse(HeaderListResponse::new(vec![])),
             )
             .await
             .unwrap();
@@ -120,7 +120,7 @@ where
         .peer_sync_handle
         .send_request(
             peer_info2.peer_id,
-            Request::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
+            SyncRequest::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
         )
         .await
         .unwrap();
@@ -130,7 +130,7 @@ where
         .peer_sync_handle
         .send_request(
             peer_info2.peer_id,
-            Request::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
+            SyncRequest::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
         )
         .await
         .unwrap();
@@ -145,7 +145,7 @@ where
                     mgr2.peer_sync_handle
                         .send_response(
                             request_id,
-                            Response::HeaderListResponse(HeaderListResponse::new(vec![])),
+                            SyncResponse::HeaderListResponse(HeaderListResponse::new(vec![])),
                         )
                         .await
                         .unwrap();
diff --git a/p2p/src/testing_utils.rs b/p2p/src/testing_utils.rs
index 7689ec7d4..bea874170 100644
--- a/p2p/src/testing_utils.rs
+++ b/p2p/src/testing_utils.rs
@@ -159,7 +159,11 @@ where
 
     let (address, peer_info1) = match timeout(Duration::from_secs(5), conn2.poll_next()).await {
         Ok(event) => match event.unwrap() {
-            ConnectivityEvent::InboundAccepted { address, peer_info } => (address, peer_info),
+            ConnectivityEvent::InboundAccepted {
+                address,
+                peer_info,
+                receiver_address: _,
+            } => (address, peer_info),
             event => panic!("expected `InboundAccepted`, got {event:?}"),
         },
         Err(_err) => panic!("did not receive `InboundAccepted` in time"),
@@ -170,6 +174,7 @@ where
             ConnectivityEvent::OutboundAccepted {
                 address: _,
                 peer_info,
+                receiver_address: _,
             } => peer_info,
             event => panic!("expected `OutboundAccepted`, got {event:?}"),
         },
diff --git a/p2p/tests/block_announcement.rs b/p2p/tests/block_announcement.rs
index 165e1e190..e689460b4 100644
--- a/p2p/tests/block_announcement.rs
+++ b/p2p/tests/block_announcement.rs
@@ -25,7 +25,6 @@ use p2p::testing_utils::{
     TestTransportTcp,
 };
 use p2p::{
-    error::{P2pError, PublishError},
     message::Announcement,
     net::{
         mock::{
@@ -82,30 +81,19 @@ where
     connect_services::<S>(&mut peer1.0, &mut peer2.0).await;
     connect_services::<S>(&mut peer2.0, &mut peer3.0).await;
 
-    // Spam the message until we have a peer.
-    loop {
-        let res = sync1
-            .make_announcement(Announcement::Block(
-                Block::new(
-                    vec![],
-                    Id::new(H256([0x03; 32])),
-                    BlockTimestamp::from_int_seconds(1337u64),
-                    ConsensusData::None,
-                    BlockReward::new(Vec::new()),
-                )
-                .unwrap(),
-            ))
-            .await;
-
-        if res.is_ok() {
-            break;
-        } else {
-            assert_eq!(
-                res,
-                Err(P2pError::PublishError(PublishError::InsufficientPeers))
-            );
-        }
-    }
+    sync1
+        .make_announcement(Announcement::Block(
+            Block::new(
+                vec![],
+                Id::new(H256([0x03; 32])),
+                BlockTimestamp::from_int_seconds(1337u64),
+                ConsensusData::None,
+                BlockReward::new(Vec::new()),
+            )
+            .unwrap(),
+        ))
+        .await
+        .unwrap();
 
     // Verify that all peers received the message even though they weren't directly connected.
     let event = peer1.1.poll_next().await.unwrap();