Repository: mintlayer/mintlayer-core
Pull request #662 (2023-01-31 14:49:36)

Use unbounded channels in default p2p backend
Using unbounded channels should fix the problem with potential deadlocks in p2p (when two channels are full).
I have removed a lot of async annotations that are no longer needed.

I think to apply back-pressure on the connected peers we could do something like this: count in `p2p::net::default_backend::peer::Peer::run` how many unprocessed requests  there are and if more than some limit then stop reading from the network socket, until some responses have been processed and sent back. I think this should solve the node overload problem nicely. It may look like this:

```
response = self.cmd_rx.recv() {
    // Request was processed, send response to the peer...
    request_count -= 1;
},
// Read from the socket if only there is some free capacity
request = self.socket.recv(), if request_count < 10 => {
    // New request received, send it to backend...
    request_count += 1;
}
```

We will need to ensure that for each request received, the backend would send a response. But I suppose it should be easy enough to do.


Patch:
diff --git a/p2p/src/net/default_backend/backend.rs b/p2p/src/net/default_backend/backend.rs
index 77a98a948..60e80abac 100644
--- a/p2p/src/net/default_backend/backend.rs
+++ b/p2p/src/net/default_backend/backend.rs
@@ -22,11 +22,7 @@ use std::{
     sync::Arc,
 };
 
-use futures::{
-    future::{join_all, BoxFuture},
-    stream::FuturesUnordered,
-    FutureExt, StreamExt, TryFutureExt,
-};
+use futures::{future::BoxFuture, stream::FuturesUnordered, FutureExt, StreamExt};
 use tokio::{sync::mpsc, time::timeout};
 
 use common::chain::ChainConfig;
@@ -58,14 +54,21 @@ use super::{peer::PeerRole, transport::TransportAddress, types::HandshakeNonce};
 /// Active peer data
 struct PeerContext {
     subscriptions: BTreeSet<PubSubTopic>,
-    tx: mpsc::Sender<Event>,
+
+    /// Channel used to send messages to the peer's event loop.
+    ///
+    /// Note that sending may fail unexpectedly if the connection is closed!
+    /// Do not propagate ChannelClosed error to the higher level, handle it locally!
+    tx: mpsc::UnboundedSender<Event>,
 }
 
 /// Pending peer data (until handshake message is received)
 struct PendingPeerContext<A> {
     address: A,
+
     peer_role: PeerRole,
-    tx: mpsc::Sender<Event>,
+
+    tx: mpsc::UnboundedSender<Event>,
 }
 
 pub struct Backend<T: TransportSocket> {
@@ -82,7 +85,7 @@ pub struct Backend<T: TransportSocket> {
     p2p_config: Arc<P2pConfig>,
 
     /// RX channel for receiving commands from the frontend
-    cmd_rx: mpsc::Receiver<Command<T>>,
+    cmd_rx: mpsc::UnboundedReceiver<Command<T>>,
 
     /// Active peers
     peers: HashMap<PeerId, PeerContext>,
@@ -93,15 +96,15 @@ pub struct Backend<T: TransportSocket> {
     /// RX channel for receiving events from peers
     #[allow(clippy::type_complexity)]
     peer_chan: (
-        mpsc::Sender<(PeerId, PeerEvent)>,
-        mpsc::Receiver<(PeerId, PeerEvent)>,
+        mpsc::UnboundedSender<(PeerId, PeerEvent)>,
+        mpsc::UnboundedReceiver<(PeerId, PeerEvent)>,
     ),
 
     /// TX channel for sending events to the frontend
-    conn_tx: mpsc::Sender<ConnectivityEvent<T>>,
+    conn_tx: mpsc::UnboundedSender<ConnectivityEvent<T>>,
 
     /// TX channel for sending syncing events
-    sync_tx: mpsc::Sender<SyncingEvent>,
+    sync_tx: mpsc::UnboundedSender<SyncingEvent>,
 
     /// Request manager for managing inbound/outbound requests and responses
     request_mgr: request_manager::RequestManager,
@@ -121,9 +124,9 @@ where
         socket: T::Listener,
         chain_config: Arc<ChainConfig>,
         p2p_config: Arc<P2pConfig>,
-        cmd_rx: mpsc::Receiver<Command<T>>,
-        conn_tx: mpsc::Sender<ConnectivityEvent<T>>,
-        sync_tx: mpsc::Sender<SyncingEvent>,
+        cmd_rx: mpsc::UnboundedReceiver<Command<T>>,
+        conn_tx: mpsc::UnboundedSender<ConnectivityEvent<T>>,
+        sync_tx: mpsc::UnboundedSender<SyncingEvent>,
     ) -> Self {
         Self {
             transport,
@@ -135,14 +138,14 @@ where
             sync_tx,
             peers: HashMap::new(),
             pending: HashMap::new(),
-            peer_chan: mpsc::channel(64),
+            peer_chan: mpsc::unbounded_channel(),
             request_mgr: request_manager::RequestManager::new(),
             command_queue: FuturesUnordered::new(),
         }
     }
 
     /// Handle connection result to a remote peer
-    async fn handle_connect_res(
+    fn handle_connect_res(
         &mut self,
         address: T::Address,
         connection_res: crate::Result<T::Stream>,
@@ -166,26 +169,25 @@ where
                         address,
                         error: P2pError::DialError(DialError::ConnectionRefusedOrTimedOut),
                     })
-                    .await
                     .map_err(P2pError::from)
             }
         }
     }
 
-    /// Disconnect remote peer by id
-    async fn disconnect_peer(&mut self, peer_id: &PeerId) -> crate::Result<()> {
+    /// Disconnect remote peer by id. Might fail if the peer is already disconnected.
+    fn disconnect_peer(&mut self, peer_id: &PeerId) -> crate::Result<()> {
         self.request_mgr.unregister_peer(peer_id);
-        if let Some(context) = self.peers.remove(peer_id) {
-            context.tx.send(Event::Disconnect).await.map_err(P2pError::from)
-        } else {
-            // TODO: Think about error handling. Currently we simply follow the libp2p behaviour.
-            log::error!("{peer_id} peer doesn't exist");
-            Ok(())
-        }
+
+        let peer = self
+            .peers
+            .remove(peer_id)
+            .ok_or(P2pError::PeerError(PeerError::PeerDoesntExist))?;
+
+        peer.tx.send(Event::Disconnect).map_err(P2pError::from)
     }
 
-    /// Sends a request to the remote peer.
-    async fn send_request(
+    /// Sends a request to the remote peer. Might fail if the peer is already disconnected.
+    fn send_request(
         &mut self,
         request_id: RequestId,
         peer_id: PeerId,
@@ -197,64 +199,58 @@ where
             .ok_or(P2pError::PeerError(PeerError::PeerDoesntExist))?;
 
         let request = self.request_mgr.make_request(request_id, request)?;
-        peer.tx.send(Event::SendMessage(request)).await.map_err(P2pError::from)?;
-
-        Ok(())
+        peer.tx.send(Event::SendMessage(request)).map_err(P2pError::from)
     }
 
-    /// Send response to a request
-    async fn send_response(
+    /// Send response to a request. Might fail if the peer is already disconnected.
+    fn send_response(
         &mut self,
         request_id: RequestId,
         response: message::Response,
     ) -> crate::Result<()> {
         log::trace!("try to send response to request, request id {request_id}");
 
-        if let Some((peer_id, response)) = self.request_mgr.make_response(&request_id, response) {
-            return self
-                .peers
-                .get_mut(&peer_id)
-                .ok_or(P2pError::PeerError(PeerError::PeerDoesntExist))?
-                .tx
-                .send(Event::SendMessage(response))
-                .await
-                .map_err(P2pError::from);
-        }
+        let (peer_id, response) = self
+            .request_mgr
+            .make_response(&request_id, response)
+            .ok_or(P2pError::Other("unknown request id"))?;
 
-        log::error!("no request for request id {request_id} exist");
-        Ok(())
+        self.peers
+            .get_mut(&peer_id)
+            .ok_or(P2pError::PeerError(PeerError::PeerDoesntExist))?
+            .tx
+            .send(Event::SendMessage(response))
+            .map_err(P2pError::from)
     }
 
     /// Sends the announcement to all peers.
     ///
     /// 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<()> {
+    fn announce_data(&mut self, topic: PubSubTopic, message: Vec<u8>) -> crate::Result<()> {
         let announcement = message::Announcement::decode(&mut &message[..])?;
 
         // Send the message to peers in pseudorandom order.
-        let mut futures: Vec<_> = self
+        let mut peers: Vec<_> = self
             .peers
             .iter()
-            .filter(|(_, peer)| peer.subscriptions.contains(&topic))
-            .map(|(id, peer)| {
-                peer.tx
-                    .send(Event::SendMessage(Box::new(Message::Announcement {
-                        announcement: announcement.clone(),
-                    })))
-                    .inspect_err(move |e| {
-                        log::error!("Failed to send announcement to peer {id}: {e:?}")
-                    })
-            })
+            .filter(|(_peer_id, peer)| peer.subscriptions.contains(&topic))
             .collect();
-        futures.shuffle(&mut make_pseudo_rng());
-
-        join_all(futures).await;
+        peers.shuffle(&mut make_pseudo_rng());
+
+        for (peer_id, peer) in peers {
+            let res = peer.tx.send(Event::SendMessage(Box::new(Message::Announcement {
+                announcement: announcement.clone(),
+            })));
+            if let Err(e) = res {
+                log::error!("Failed to send announcement to peer {peer_id}: {e:?}")
+            }
+        }
 
         Ok(())
     }
 
     /// Handle incoming request
-    async fn handle_incoming_request(
+    fn handle_incoming_request(
         &mut self,
         peer_id: PeerId,
         request_id: RequestId,
@@ -272,7 +268,6 @@ where
                     request_id,
                     request: SyncRequest::HeaderListRequest(request),
                 })
-                .await
                 .map_err(P2pError::from),
             message::Request::BlockListRequest(request) => self
                 .sync_tx
@@ -281,7 +276,6 @@ where
                     request_id,
                     request: SyncRequest::BlockListRequest(request),
                 })
-                .await
                 .map_err(P2pError::from),
             message::Request::AddrListRequest(request) => self
                 .conn_tx
@@ -290,7 +284,6 @@ where
                     request_id,
                     request: PeerManagerRequest::AddrListRequest(request),
                 })
-                .await
                 .map_err(P2pError::from),
             message::Request::AnnounceAddrRequest(request) => self
                 .conn_tx
@@ -299,7 +292,6 @@ where
                     request_id,
                     request: PeerManagerRequest::AnnounceAddrRequest(request),
                 })
-                .await
                 .map_err(P2pError::from),
             message::Request::PingRequest(request) => self
                 .conn_tx
@@ -308,13 +300,12 @@ where
                     request_id,
                     request: PeerManagerRequest::PingRequest(request),
                 })
-                .await
                 .map_err(P2pError::from),
         }
     }
 
     /// Handle incoming response
-    async fn handle_incoming_response(
+    fn handle_incoming_response(
         &mut self,
         peer_id: PeerId,
         request_id: RequestId,
@@ -330,7 +321,6 @@ where
                     request_id,
                     response: SyncResponse::HeaderListResponse(response),
                 })
-                .await
                 .map_err(P2pError::from),
             message::Response::BlockListResponse(response) => self
                 .sync_tx
@@ -339,7 +329,6 @@ where
                     request_id,
                     response: SyncResponse::BlockListResponse(response),
                 })
-                .await
                 .map_err(P2pError::from),
             message::Response::AddrListResponse(response) => self
                 .conn_tx
@@ -348,7 +337,6 @@ where
                     request_id,
                     response: PeerManagerResponse::AddrListResponse(response),
                 })
-                .await
                 .map_err(P2pError::from),
             message::Response::AnnounceAddrResponse(response) => self
                 .conn_tx
@@ -357,7 +345,6 @@ where
                     request_id,
                     response: PeerManagerResponse::AnnounceAddrResponse(response),
                 })
-                .await
                 .map_err(P2pError::from),
             message::Response::PingResponse(response) => self
                 .conn_tx
@@ -366,12 +353,11 @@ where
                     request_id,
                     response: PeerManagerResponse::PingResponse(response),
                 })
-                .await
                 .map_err(P2pError::from),
         }
     }
 
-    async fn handle_announcement(
+    fn handle_announcement(
         &mut self,
         peer_id: PeerId,
         announcement: Announcement,
@@ -386,7 +372,6 @@ where
                         ANNOUNCEMENT_MAX_SIZE,
                     )),
                 })
-                .await
                 .map_err(P2pError::from)?;
         }
 
@@ -395,7 +380,6 @@ where
                 peer_id,
                 announcement: Box::new(announcement),
             })
-            .await
             .map_err(P2pError::from)
     }
 
@@ -403,6 +387,22 @@ where
     pub async fn run(&mut self) -> crate::Result<()> {
         loop {
             tokio::select! {
+                // Select from the channels in the specified order
+                biased;
+
+                // Handle commands.
+                command = self.cmd_rx.recv() => {
+                    self.handle_command(command.ok_or(P2pError::ChannelClosed)?).await?;
+                },
+                // Process pending commands
+                callback = self.command_queue.select_next_some(), if !self.command_queue.is_empty() => {
+                    callback(self)?;
+                },
+                // Handle peer events.
+                event = self.peer_chan.1.recv() => {
+                    let (peer, event) = event.ok_or(P2pError::ChannelClosed)?;
+                    self.handle_peer_event(peer, event)?;
+                },
                 // Accept a new peer connection.
                 res = self.socket.accept() => {
                     let (stream, address) = res.map_err(|_| P2pError::Other("accept() failed"))?;
@@ -414,18 +414,6 @@ where
                         address,
                     )?;
                 }
-                // Handle peer events.
-                event = self.peer_chan.1.recv() => {
-                    let (peer, event) = event.ok_or(P2pError::ChannelClosed)?;
-                    self.handle_peer_event(peer, event).await?;
-                },
-                // Handle commands.
-                command = self.cmd_rx.recv() => {
-                    self.handle_command(command.ok_or(P2pError::ChannelClosed)?).await?;
-                },
-                callback = self.command_queue.select_next_some(), if !self.command_queue.is_empty() => {
-                    callback(self).await?;
-                },
             }
         }
     }
@@ -442,7 +430,7 @@ where
         peer_role: PeerRole,
         address: T::Address,
     ) -> crate::Result<()> {
-        let (tx, rx) = mpsc::channel(16);
+        let (tx, rx) = mpsc::unbounded_channel();
 
         let receiver_address = Some(address.as_peer_address());
 
@@ -474,13 +462,12 @@ where
             if let Err(err) = run_res {
                 log::error!("peer {remote_peer_id} failed: {err}");
             }
-            peer.destroy().await;
         });
 
         Ok(())
     }
 
-    async fn is_connection_from_self(
+    fn is_connection_from_self(
         &mut self,
         peer_role: PeerRole,
         incoming_nonce: HandshakeNonce,
@@ -513,7 +500,6 @@ where
                         address: outbound_pending.address,
                         error: P2pError::DialError(DialError::AttemptToDialSelf),
                     })
-                    .await
                     .map_err(P2pError::from)?;
 
                 // Nothing else to do, just drop inbound connection
@@ -524,7 +510,7 @@ where
         Ok(false)
     }
 
-    async fn handle_peer_event(&mut self, peer_id: PeerId, event: PeerEvent) -> crate::Result<()> {
+    fn handle_peer_event(&mut self, peer_id: PeerId, event: PeerEvent) -> crate::Result<()> {
         match event {
             PeerEvent::PeerInfoReceived {
                 network,
@@ -543,7 +529,7 @@ where
                     None => return Ok(()),
                 };
 
-                if self.is_connection_from_self(peer_role, handshake_nonce).await? {
+                if self.is_connection_from_self(peer_role, handshake_nonce)? {
                     return Ok(());
                 }
 
@@ -561,13 +547,12 @@ where
                                 },
                                 receiver_address,
                             })
-                            .await
                             .map_err(P2pError::from)?;
                     }
                     PeerRole::Inbound => {
                         self.conn_tx
                             .send(ConnectivityEvent::InboundAccepted {
-                                address: address.clone(),
+                                address,
                                 peer_info: PeerInfo {
                                     peer_id,
                                     network,
@@ -577,7 +562,6 @@ where
                                 },
                                 receiver_address,
                             })
-                            .await
                             .map_err(P2pError::from)?;
                     }
                 }
@@ -586,7 +570,7 @@ where
                 let _ = self.request_mgr.register_peer(peer_id);
             }
             PeerEvent::MessageReceived { message } => {
-                self.handle_message(peer_id, message).await?;
+                self.handle_message(peer_id, message)?;
             }
             PeerEvent::ConnectionClosed => {
                 self.pending.remove(&peer_id);
@@ -598,14 +582,13 @@ where
                 // But doing so will break some unit tests.
                 self.conn_tx
                     .send(ConnectivityEvent::ConnectionClosed { peer_id })
-                    .await
                     .map_err(P2pError::from)?;
             }
         }
         Ok(())
     }
 
-    async fn handle_message(&mut self, peer_id: PeerId, message: Message) -> crate::Result<()> {
+    fn handle_message(&mut self, peer_id: PeerId, message: Message) -> crate::Result<()> {
         match message {
             Message::Handshake(_) => {
                 log::error!("peer {peer_id} sent handshaking message");
@@ -614,33 +597,29 @@ where
                 request_id,
                 request,
             } => {
-                self.handle_incoming_request(peer_id, request_id, request).await?;
+                self.handle_incoming_request(peer_id, request_id, request)?;
             }
             Message::Response {
                 request_id,
                 response,
             } => {
-                self.handle_incoming_response(peer_id, request_id, response).await?;
+                self.handle_incoming_response(peer_id, request_id, response)?;
             }
             Message::Announcement { announcement } => {
-                self.handle_announcement(peer_id, announcement).await?;
+                self.handle_announcement(peer_id, announcement)?;
             }
         }
         Ok(())
     }
 
     async fn handle_command(&mut self, command: Command<T>) -> crate::Result<()> {
-        // All handlings are separated to two parts, fast and slow:
-        // - "Slow", potentially blocking part (can't take reference to '&mut self' because they are run concurrently).
-        // - "Fast", non-blocking part (take mutable reference to self because they are run sequentially).
+        // All handlings are separated to two parts:
+        // - Async (can't take mutable reference to self because they are run concurrently).
+        // - Sync (take mutable reference to self because they are run sequentially).
         // Because the second part depends on result of the first part boxed closures are used.
-        // So the first future will return a closure that takes a reference to self and returns one more future.
 
         let backend_task: BackendTask<T> = match command {
-            Command::Connect { address, response } => {
-                // For now we always respond with success
-                response.send(Ok(())).map_err(|_| P2pError::ChannelClosed)?;
-
+            Command::Connect { address } => {
                 let connection_fut = timeout(
                     *self.p2p_config.outbound_connection_timeout,
                     self.transport.connect(address.clone()),
@@ -651,20 +630,17 @@ where
                         DialError::ConnectionRefusedOrTimedOut,
                     )));
 
-                    boxed_cb(move |this| {
-                        async move { this.handle_connect_res(address, connection_res).await }
-                            .boxed()
-                    })
+                    boxed_cb(move |this| this.handle_connect_res(address, connection_res))
                 }
                 .boxed()
             }
-            Command::Disconnect { peer_id, response } => async move {
+            Command::Disconnect { peer_id } => async move {
                 boxed_cb(move |this: &mut Self| {
-                    async move {
-                        let res = this.disconnect_peer(&peer_id).await;
-                        response.send(res).map_err(|_| P2pError::ChannelClosed)
+                    let res = this.disconnect_peer(&peer_id);
+                    if let Err(e) = res {
+                        log::debug!("Failed to disconnect peer {peer_id}: {e}")
                     }
-                    .boxed()
+                    Ok(())
                 })
             }
             .boxed(),
@@ -674,14 +650,11 @@ where
                 message,
             } => async move {
                 boxed_cb(move |this| {
-                    async move {
-                        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(())
+                    let res = this.send_request(request_id, peer_id, message);
+                    if let Err(e) = res {
+                        log::debug!("Failed to send request to peer {peer_id}: {e}")
                     }
-                    .boxed()
+                    Ok(())
                 })
             }
             .boxed(),
@@ -690,27 +663,21 @@ where
                 message,
             } => async move {
                 boxed_cb(move |this| {
-                    async move {
-                        let res = this.send_response(request_id, message).await;
-                        if let Err(e) = res {
-                            log::debug!("Failed to send response to peer: {e}")
-                        }
-                        Ok(())
+                    let res = this.send_response(request_id, message);
+                    if let Err(e) = res {
+                        log::debug!("Failed to send response to peer: {e}")
                     }
-                    .boxed()
+                    Ok(())
                 })
             }
             .boxed(),
             Command::AnnounceData { topic, message } => async move {
                 boxed_cb(move |this| {
-                    async move {
-                        let res = this.announce_data(topic, message).await;
-                        if let Err(e) = res {
-                            log::error!("Failed to send announce data: {e}")
-                        }
-                        Ok(())
+                    let res = this.announce_data(topic, message);
+                    if let Err(e) = res {
+                        log::error!("Failed to send announce data: {e}")
                     }
-                    .boxed()
+                    Ok(())
                 })
             }
             .boxed(),
@@ -726,11 +693,12 @@ where
 
 type BackendTask<T> = BoxFuture<'static, BackendTaskCallback<T>>;
 
-type BackendTaskCallback<T> = Box<dyn FnOnce(&mut Backend<T>) -> BackendTaskFut + Send>;
-
-type BackendTaskFut<'a> = BoxFuture<'a, crate::Result<()>>;
+type BackendTaskCallback<T> = Box<dyn FnOnce(&mut Backend<T>) -> crate::Result<()> + Send>;
 
-fn boxed_cb<T: TransportSocket, F: FnOnce(&mut Backend<T>) -> BackendTaskFut + Send + 'static>(
+fn boxed_cb<
+    T: TransportSocket,
+    F: FnOnce(&mut Backend<T>) -> crate::Result<()> + Send + 'static,
+>(
     f: F,
 ) -> BackendTaskCallback<T> {
     Box::new(f)
diff --git a/p2p/src/net/default_backend/mod.rs b/p2p/src/net/default_backend/mod.rs
index 0e63bb170..5f648e337 100644
--- a/p2p/src/net/default_backend/mod.rs
+++ b/p2p/src/net/default_backend/mod.rs
@@ -23,7 +23,7 @@ pub mod types;
 use std::{marker::PhantomData, sync::Arc};
 
 use async_trait::async_trait;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::mpsc;
 
 use logging::log;
 use serialization::Encode;
@@ -52,10 +52,10 @@ pub struct ConnectivityHandle<S: NetworkingService, T: TransportSocket> {
     local_addresses: Vec<S::Address>,
 
     /// TX channel for sending commands to default_backend backend
-    cmd_tx: mpsc::Sender<types::Command<T>>,
+    cmd_tx: mpsc::UnboundedSender<types::Command<T>>,
 
     /// RX channel for receiving connectivity events from default_backend backend
-    conn_rx: mpsc::Receiver<types::ConnectivityEvent<T>>,
+    conn_rx: mpsc::UnboundedReceiver<types::ConnectivityEvent<T>>,
 
     _marker: PhantomData<fn() -> S>,
 }
@@ -63,8 +63,8 @@ pub struct ConnectivityHandle<S: NetworkingService, T: TransportSocket> {
 impl<S: NetworkingService, T: TransportSocket> ConnectivityHandle<S, T> {
     pub fn new(
         local_addresses: Vec<S::Address>,
-        cmd_tx: mpsc::Sender<types::Command<T>>,
-        conn_rx: mpsc::Receiver<types::ConnectivityEvent<T>>,
+        cmd_tx: mpsc::UnboundedSender<types::Command<T>>,
+        conn_rx: mpsc::UnboundedReceiver<types::ConnectivityEvent<T>>,
     ) -> Self {
         Self {
             local_addresses,
@@ -81,10 +81,10 @@ where
     T: TransportSocket,
 {
     /// TX channel for sending commands to default_backend backend
-    _cmd_tx: mpsc::Sender<types::Command<T>>,
+    _cmd_tx: mpsc::UnboundedSender<types::Command<T>>,
 
     /// RX channel for receiving pubsub events from default_backend backend
-    _pubsub_rx: mpsc::Receiver<types::PubSubEvent<T>>,
+    _pubsub_rx: mpsc::UnboundedReceiver<types::PubSubEvent<T>>,
 
     _marker: PhantomData<fn() -> S>,
 }
@@ -96,10 +96,11 @@ where
     T: TransportSocket,
 {
     /// TX channel for sending commands to default_backend backend
-    cmd_tx: mpsc::Sender<types::Command<T>>,
+    cmd_tx: mpsc::UnboundedSender<types::Command<T>>,
 
     /// RX channel for receiving syncing events
-    sync_rx: mpsc::Receiver<types::SyncingEvent>,
+    sync_rx: mpsc::UnboundedReceiver<types::SyncingEvent>,
+
     _marker: PhantomData<fn() -> S>,
 }
 
@@ -119,9 +120,9 @@ impl<T: TransportSocket> NetworkingService for DefaultNetworkingService<T> {
         chain_config: Arc<common::chain::ChainConfig>,
         p2p_config: Arc<config::P2pConfig>,
     ) -> crate::Result<(Self::ConnectivityHandle, Self::SyncingMessagingHandle)> {
-        let (cmd_tx, cmd_rx) = mpsc::channel(16);
-        let (conn_tx, conn_rx) = mpsc::channel(16);
-        let (sync_tx, sync_rx) = mpsc::channel(16);
+        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
+        let (conn_tx, conn_rx) = mpsc::unbounded_channel();
+        let (sync_tx, sync_rx) = mpsc::unbounded_channel();
         let socket = transport.bind(bind_addresses).await?;
         let local_addresses = socket.local_addresses().expect("to have bind address available");
 
@@ -158,56 +159,38 @@ where
     S: NetworkingService<Address = T::Address, PeerId = PeerId, PeerRequestId = RequestId> + Send,
     T: TransportSocket,
 {
-    async fn connect(&mut self, address: S::Address) -> crate::Result<()> {
+    fn connect(&mut self, address: S::Address) -> crate::Result<()> {
         log::debug!(
             "try to establish outbound connection, address {:?}",
             address
         );
 
-        let (tx, rx) = oneshot::channel();
-        self.cmd_tx
-            .send(types::Command::Connect {
-                address,
-                response: tx,
-            })
-            .await?;
-
-        rx.await?
+        self.cmd_tx.send(types::Command::Connect { address }).map_err(P2pError::from)
     }
 
-    async fn disconnect(&mut self, peer_id: S::PeerId) -> crate::Result<()> {
+    fn disconnect(&mut self, peer_id: S::PeerId) -> crate::Result<()> {
         log::debug!("close connection with remote, {peer_id}");
 
-        let (tx, rx) = oneshot::channel();
-        self.cmd_tx
-            .send(types::Command::Disconnect {
-                peer_id,
-                response: tx,
-            })
-            .await?;
-
-        rx.await?
+        self.cmd_tx.send(types::Command::Disconnect { peer_id }).map_err(P2pError::from)
     }
 
-    async fn send_request(
+    fn send_request(
         &mut self,
         peer_id: S::PeerId,
         request: PeerManagerRequest,
     ) -> crate::Result<S::PeerRequestId> {
         let request_id = RequestId::new();
 
-        self.cmd_tx
-            .send(types::Command::SendRequest {
-                peer_id,
-                request_id,
-                message: request.into(),
-            })
-            .await?;
+        self.cmd_tx.send(types::Command::SendRequest {
+            peer_id,
+            request_id,
+            message: request.into(),
+        })?;
 
         Ok(request_id)
     }
 
-    async fn send_response(
+    fn send_response(
         &mut self,
         request_id: S::PeerRequestId,
         response: PeerManagerResponse,
@@ -217,8 +200,7 @@ where
                 request_id,
                 message: response.into(),
             })
-            .await?;
-        Ok(())
+            .map_err(P2pError::from)
     }
 
     fn local_addresses(&self) -> &[S::Address] {
@@ -282,42 +264,35 @@ where
     S: NetworkingService<PeerId = PeerId, PeerRequestId = RequestId> + Send,
     T: TransportSocket,
 {
-    async fn send_request(
+    fn send_request(
         &mut self,
         peer_id: S::PeerId,
         request: SyncRequest,
     ) -> crate::Result<S::PeerRequestId> {
         let request_id = RequestId::new();
 
-        self.cmd_tx
-            .send(types::Command::SendRequest {
-                peer_id,
-                request_id,
-                message: request.into(),
-            })
-            .await?;
+        self.cmd_tx.send(types::Command::SendRequest {
+            peer_id,
+            request_id,
+            message: request.into(),
+        })?;
 
         Ok(request_id)
     }
 
-    async fn send_response(
+    fn send_response(
         &mut self,
         request_id: S::PeerRequestId,
         response: SyncResponse,
     ) -> crate::Result<()> {
-        self.cmd_tx
-            .send(types::Command::SendResponse {
-                request_id,
-                message: response.into(),
-            })
-            .await?;
+        self.cmd_tx.send(types::Command::SendResponse {
+            request_id,
+            message: response.into(),
+        })?;
         Ok(())
     }
 
-    async fn make_announcement(
-        &mut self,
-        announcement: message::Announcement,
-    ) -> crate::Result<()> {
+    fn make_announcement(&mut self, announcement: message::Announcement) -> crate::Result<()> {
         let message = announcement.encode();
         if message.len() > ANNOUNCEMENT_MAX_SIZE {
             return Err(P2pError::PublishError(PublishError::MessageTooLarge(
@@ -330,9 +305,9 @@ where
             message::Announcement::Block(_) => PubSubTopic::Blocks,
         };
 
-        self.cmd_tx.send(types::Command::AnnounceData { topic, message }).await?;
-
-        Ok(())
+        self.cmd_tx
+            .send(types::Command::AnnounceData { topic, message })
+            .map_err(P2pError::from)
     }
 
     async fn poll_next(&mut self) -> crate::Result<SyncingEvent<S>> {
@@ -405,7 +380,7 @@ mod tests {
         .unwrap();
 
         let addr = conn2.local_addresses();
-        assert_eq!(conn1.connect(addr[0].clone()).await, Ok(()));
+        assert_eq!(conn1.connect(addr[0].clone()), Ok(()));
 
         if let Ok(ConnectivityEvent::OutboundAccepted {
             address,
@@ -468,7 +443,8 @@ mod tests {
         .unwrap();
 
         let bind_address = conn2.local_addresses();
-        let (_res1, res2) = tokio::join!(conn1.connect(bind_address[0].clone()), conn2.poll_next());
+        conn1.connect(bind_address[0].clone()).unwrap();
+        let res2 = conn2.poll_next().await;
         match res2.unwrap() {
             ConnectivityEvent::InboundAccepted {
                 address: _,
@@ -526,10 +502,8 @@ mod tests {
         .await
         .unwrap();
 
-        let (_res1, res2) = tokio::join!(
-            conn1.connect(conn2.local_addresses()[0].clone()),
-            conn2.poll_next()
-        );
+        conn1.connect(conn2.local_addresses()[0].clone()).unwrap();
+        let res2 = conn2.poll_next().await;
 
         match res2.unwrap() {
             ConnectivityEvent::InboundAccepted {
@@ -537,7 +511,7 @@ mod tests {
                 peer_info,
                 receiver_address: _,
             } => {
-                assert_eq!(conn2.disconnect(peer_info.peer_id).await, Ok(()));
+                assert_eq!(conn2.disconnect(peer_info.peer_id), Ok(()));
             }
             _ => panic!("invalid event received, expected incoming connection"),
         }
@@ -586,7 +560,7 @@ mod tests {
 
         // Try connect to self
         let addr = conn1.local_addresses();
-        assert_eq!(conn1.connect(addr[0].clone()).await, Ok(()));
+        assert_eq!(conn1.connect(addr[0].clone()), Ok(()));
 
         // ConnectionError should be reported
         if let Ok(ConnectivityEvent::ConnectionError { address, error }) = conn1.poll_next().await {
@@ -608,7 +582,7 @@ mod tests {
 
         // Check that we can still connect normally after
         let addr = conn2.local_addresses();
-        assert_eq!(conn1.connect(addr[0].clone()).await, Ok(()));
+        assert_eq!(conn1.connect(addr[0].clone()), Ok(()));
         if let Ok(ConnectivityEvent::OutboundAccepted {
             address,
             peer_info,
diff --git a/p2p/src/net/default_backend/peer.rs b/p2p/src/net/default_backend/peer.rs
index ad34d5921..320d77107 100644
--- a/p2p/src/net/default_backend/peer.rs
+++ b/p2p/src/net/default_backend/peer.rs
@@ -71,10 +71,10 @@ pub struct Peer<T: TransportSocket> {
     receiver_address: Option<PeerAddress>,
 
     /// TX channel for communicating with backend
-    tx: mpsc::Sender<(PeerId, PeerEvent)>,
+    tx: mpsc::UnboundedSender<(PeerId, PeerEvent)>,
 
     /// RX channel for receiving commands from backend
-    rx: mpsc::Receiver<Event>,
+    rx: mpsc::UnboundedReceiver<Event>,
 }
 
 impl<T> Peer<T>
@@ -89,8 +89,8 @@ where
         p2p_config: Arc<P2pConfig>,
         socket: T::Stream,
         receiver_address: Option<PeerAddress>,
-        tx: mpsc::Sender<(PeerId, PeerEvent)>,
-        rx: mpsc::Receiver<Event>,
+        tx: mpsc::UnboundedSender<(PeerId, PeerEvent)>,
+        rx: mpsc::UnboundedReceiver<Event>,
     ) -> Self {
         let socket = BufferedTranscoder::new(socket);
 
@@ -134,7 +134,6 @@ where
                             handshake_nonce,
                         },
                     ))
-                    .await
                     .map_err(P2pError::from)?;
 
                 self.socket
@@ -180,7 +179,6 @@ where
                             handshake_nonce,
                         },
                     ))
-                    .await
                     .map_err(P2pError::from)?;
             }
         }
@@ -188,10 +186,6 @@ where
         Ok(())
     }
 
-    pub async fn destroy(self) {
-        let _ = self.tx.send((self.peer_id, types::PeerEvent::ConnectionClosed)).await;
-    }
-
     pub async fn run(&mut self) -> crate::Result<()> {
         // handshake with remote peer and send peer's info to backend
         let handshake_res = timeout(PEER_HANDSHAKE_TIMEOUT, self.handshake()).await;
@@ -209,6 +203,9 @@ where
 
         loop {
             tokio::select! {
+                // Sending messages should have higher priority
+                biased;
+
                 event = self.rx.recv() => match event.ok_or(P2pError::ChannelClosed)? {
                     Event::Disconnect => return Ok(()),
                     Event::SendMessage(message) => self.socket.send(*message).await?,
@@ -226,7 +223,6 @@ where
                                     message
                                 },
                             ))
-                            .await
                             .map_err(P2pError::from)?;
                     }
                 }
@@ -235,6 +231,12 @@ where
     }
 }
 
+impl<T: TransportSocket> Drop for Peer<T> {
+    fn drop(&mut self) {
+        let _ = self.tx.send((self.peer_id, types::PeerEvent::ConnectionClosed));
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -264,8 +266,8 @@ mod tests {
         let (socket1, socket2) = get_two_connected_sockets::<A, T>().await;
         let chain_config = Arc::new(common::chain::config::create_mainnet());
         let p2p_config = Arc::new(P2pConfig::default());
-        let (tx1, mut rx1) = mpsc::channel(16);
-        let (_tx2, rx2) = mpsc::channel(16);
+        let (tx1, mut rx1) = mpsc::unbounded_channel();
+        let (_tx2, rx2) = mpsc::unbounded_channel();
         let peer_id2 = PeerId::new();
 
         let mut peer = Peer::<T>::new(
@@ -337,8 +339,8 @@ mod tests {
         let (socket1, socket2) = get_two_connected_sockets::<A, T>().await;
         let chain_config = Arc::new(common::chain::config::create_mainnet());
         let p2p_config = Arc::new(P2pConfig::default());
-        let (tx1, mut rx1) = mpsc::channel(16);
-        let (_tx2, rx2) = mpsc::channel(16);
+        let (tx1, mut rx1) = mpsc::unbounded_channel();
+        let (_tx2, rx2) = mpsc::unbounded_channel();
         let peer_id3 = PeerId::new();
 
         let mut peer = Peer::<T>::new(
@@ -414,8 +416,8 @@ mod tests {
         let (socket1, socket2) = get_two_connected_sockets::<A, T>().await;
         let chain_config = Arc::new(common::chain::config::create_mainnet());
         let p2p_config = Arc::new(P2pConfig::default());
-        let (tx1, _rx1) = mpsc::channel(16);
-        let (_tx2, rx2) = mpsc::channel(16);
+        let (tx1, _rx1) = mpsc::unbounded_channel();
+        let (_tx2, rx2) = mpsc::unbounded_channel();
         let peer_id3 = PeerId::new();
 
         let mut peer = Peer::<T>::new(
@@ -472,8 +474,8 @@ mod tests {
         let (socket1, socket2) = get_two_connected_sockets::<A, T>().await;
         let chain_config = Arc::new(common::chain::config::create_mainnet());
         let p2p_config = Arc::new(P2pConfig::default());
-        let (tx1, _rx1) = mpsc::channel(16);
-        let (_tx2, rx2) = mpsc::channel(16);
+        let (tx1, _rx1) = mpsc::unbounded_channel();
+        let (_tx2, rx2) = mpsc::unbounded_channel();
         let peer_id2 = PeerId::new();
 
         let mut peer = Peer::<T>::new(
diff --git a/p2p/src/net/default_backend/types.rs b/p2p/src/net/default_backend/types.rs
index cde8872d8..bf4a9d910 100644
--- a/p2p/src/net/default_backend/types.rs
+++ b/p2p/src/net/default_backend/types.rs
@@ -20,8 +20,6 @@ use std::{
     sync::atomic::{AtomicU64, Ordering},
 };
 
-use tokio::sync::oneshot;
-
 use common::primitives::semver::SemVer;
 use serialization::{Decode, Encode};
 
@@ -39,11 +37,9 @@ use crate::{
 pub enum Command<T: TransportSocket> {
     Connect {
         address: T::Address,
-        response: oneshot::Sender<crate::Result<()>>,
     },
     Disconnect {
         peer_id: PeerId,
-        response: oneshot::Sender<crate::Result<()>>,
     },
     SendRequest {
         peer_id: PeerId,
diff --git a/p2p/src/net/mod.rs b/p2p/src/net/mod.rs
index 5f41bc1fd..7d25ca429 100644
--- a/p2p/src/net/mod.rs
+++ b/p2p/src/net/mod.rs
@@ -108,20 +108,20 @@ where
     ///
     /// # Arguments
     /// `address` - socket address of the peer
-    async fn connect(&mut self, address: T::Address) -> crate::Result<()>;
+    fn connect(&mut self, address: T::Address) -> crate::Result<()>;
 
     /// Disconnect active connection
     ///
     /// # Arguments
     /// `peer_id` - Peer ID of the remote node
-    async fn disconnect(&mut self, peer_id: T::PeerId) -> crate::Result<()>;
+    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(
+    fn send_request(
         &mut self,
         peer_id: T::PeerId,
         request: PeerManagerRequest,
@@ -132,7 +132,7 @@ where
     /// # Arguments
     /// * `request_id` - ID of the request this is a response to
     /// * `response` - Response to be sent
-    async fn send_response(
+    fn send_response(
         &mut self,
         request_id: T::PeerRequestId,
         response: PeerManagerResponse,
@@ -162,7 +162,7 @@ where
     /// # Arguments
     /// * `peer_id` - Unique ID of the peer the request is sent to
     /// * `request` - Request to be sent
-    async fn send_request(
+    fn send_request(
         &mut self,
         peer_id: T::PeerId,
         request: SyncRequest,
@@ -173,14 +173,14 @@ where
     /// # Arguments
     /// * `request_id` - ID of the request this is a response to
     /// * `response` - Response to be sent
-    async fn send_response(
+    fn send_response(
         &mut self,
         request_id: T::PeerRequestId,
         response: SyncResponse,
     ) -> crate::Result<()>;
 
     /// Publishes an announcement on the network.
-    async fn make_announcement(&mut self, announcement: Announcement) -> crate::Result<()>;
+    fn make_announcement(&mut self, announcement: Announcement) -> crate::Result<()>;
 
     /// Poll syncing-related event from the networking service
     async fn poll_next(&mut self) -> crate::Result<types::SyncingEvent<T>>;
diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs
index 2cc7f5969..7789f96f0 100644
--- a/p2p/src/peer_manager/mod.rs
+++ b/p2p/src/peer_manager/mod.rs
@@ -182,7 +182,7 @@ where
     /// *receiver_address* is this host socket address as seen and reported by remote peer.
     /// This should work for hosts with public IPs and for hosts behind NAT with port forwarding (same port is assumed).
     /// This won't work for majority of nodes but that should be accepted.
-    async fn handle_outbound_receiver_address(
+    fn handle_outbound_receiver_address(
         &mut self,
         peer_id: T::PeerId,
         receiver_address: PeerAddress,
@@ -218,27 +218,25 @@ where
             .collect::<Vec<_>>();
 
         for address in discovered_own_addresses {
-            self.send_announced_address(peer_id, address).await?;
+            self.send_announced_address(peer_id, address)?;
         }
 
         Ok(())
     }
 
-    async fn send_announced_address(
+    fn send_announced_address(
         &mut self,
         peer_id: T::PeerId,
         address: T::Address,
     ) -> crate::Result<()> {
         let peer_addresses = self.announced_addresses.entry(peer_id).or_default();
         if !peer_addresses.contains(&address) {
-            self.peer_connectivity_handle
-                .send_request(
-                    peer_id,
-                    PeerManagerRequest::AnnounceAddrRequest(AnnounceAddrRequest {
-                        address: address.as_peer_address(),
-                    }),
-                )
-                .await?;
+            self.peer_connectivity_handle.send_request(
+                peer_id,
+                PeerManagerRequest::AnnounceAddrRequest(AnnounceAddrRequest {
+                    address: address.as_peer_address(),
+                }),
+            )?;
             peer_addresses.insert(address);
         }
         Ok(())
@@ -249,7 +247,7 @@ where
     /// The event is received from the networking backend and it's either a result of an incoming
     /// connection from a remote peer or a response to an outbound connection that was initiated
     /// by the node as result of the peer manager maintenance.
-    async fn accept_connection(
+    fn accept_connection(
         &mut self,
         address: T::Address,
         role: Role,
@@ -282,16 +280,14 @@ where
         );
 
         if let (Some(receiver_address), Role::Outbound) = (receiver_address, role) {
-            self.handle_outbound_receiver_address(peer_id, receiver_address).await?;
+            self.handle_outbound_receiver_address(peer_id, receiver_address)?;
         }
 
         if role == Role::Outbound {
-            self.peer_connectivity_handle
-                .send_request(
-                    peer_id,
-                    PeerManagerRequest::AddrListRequest(AddrListRequest {}),
-                )
-                .await?;
+            self.peer_connectivity_handle.send_request(
+                peer_id,
+                PeerManagerRequest::AddrListRequest(AddrListRequest {}),
+            )?;
         }
 
         log::info!(
@@ -326,7 +322,7 @@ where
     /// This function verifies that neither address the nor the peer ID are on the
     /// list of banned IPs/peer IDs. It also checks that the maximum number of
     /// connections `PeerManager` is configured to have has not been reached.
-    async fn accept_inbound_connection(
+    fn accept_inbound_connection(
         &mut self,
         address: T::Address,
         info: net::types::PeerInfo<T::PeerId>,
@@ -353,7 +349,7 @@ where
             return Err(P2pError::PeerError(PeerError::TooManyPeers));
         }
 
-        self.accept_connection(address, Role::Inbound, info, receiver_address).await
+        self.accept_connection(address, Role::Inbound, info, receiver_address)
     }
 
     /// The connection to a remote peer is reported as closed.
@@ -391,7 +387,7 @@ where
     ///
     /// If peer is banned, it is removed from the connected peers
     /// and its address is marked as banned.
-    async fn adjust_peer_score(&mut self, peer_id: T::PeerId, score: u32) -> crate::Result<()> {
+    fn adjust_peer_score(&mut self, peer_id: T::PeerId, score: u32) -> crate::Result<()> {
         log::debug!("adjusting score for peer {peer_id}, adjustment {score}");
 
         let peer = match self.peers.get_mut(&peer_id) {
@@ -403,7 +399,7 @@ where
 
         if peer.score >= *self.p2p_config.ban_threshold {
             self.peerdb.ban_peer(&peer.address)?;
-            self.disconnect(peer_id, None).await?;
+            self.disconnect(peer_id, None)?;
         }
 
         Ok(())
@@ -431,7 +427,7 @@ where
     /// This function doesn't block on the call but sends a command to the
     /// networking backend which then reports at some point in the future
     /// whether the connection failed or succeeded.
-    async fn try_connect(&mut self, address: T::Address) -> crate::Result<()> {
+    fn try_connect(&mut self, address: T::Address) -> crate::Result<()> {
         ensure!(
             !self.pending_connects.contains_key(&address),
             P2pError::PeerError(PeerError::Pending(address.to_string())),
@@ -448,18 +444,18 @@ where
             P2pError::PeerError(PeerError::BannedAddress(address.to_string())),
         );
 
-        self.peer_connectivity_handle.connect(address).await
+        self.peer_connectivity_handle.connect(address)
     }
 
     /// Establish an outbound connection
-    async fn connect(
+    fn connect(
         &mut self,
         address: T::Address,
         response: Option<oneshot::Sender<crate::Result<()>>>,
     ) -> crate::Result<()> {
         log::debug!("try to establish outbound connection to peer at address {address:?}");
 
-        let res = self.try_connect(address.clone()).await;
+        let res = self.try_connect(address.clone());
 
         match res {
             Ok(()) => {
@@ -475,7 +471,7 @@ where
         Ok(())
     }
 
-    async fn try_disconnect(&mut self, peer_id: T::PeerId) -> crate::Result<()> {
+    fn try_disconnect(&mut self, peer_id: T::PeerId) -> crate::Result<()> {
         ensure!(
             !self.pending_disconnects.contains_key(&peer_id),
             P2pError::PeerError(PeerError::Pending(peer_id.to_string())),
@@ -486,7 +482,7 @@ where
             P2pError::PeerError(PeerError::PeerDisconnected),
         );
 
-        self.peer_connectivity_handle.disconnect(peer_id).await
+        self.peer_connectivity_handle.disconnect(peer_id)
     }
 
     /// Disconnect an existing connection (inbound or outbound)
@@ -494,14 +490,14 @@ where
     /// The decision to close the connection is made either by the user via RPC
     /// or by the [`PeerManager::heartbeat()`] function which has decided to cull
     /// this connection in favor of another potential connection.
-    async fn disconnect(
+    fn disconnect(
         &mut self,
         peer_id: T::PeerId,
         response: Option<oneshot::Sender<crate::Result<()>>>,
     ) -> crate::Result<()> {
         log::debug!("disconnect peer {peer_id}");
 
-        let res = self.try_disconnect(peer_id).await;
+        let res = self.try_disconnect(peer_id);
 
         match res {
             Ok(()) => {
@@ -538,7 +534,7 @@ where
     /// the number of desired connections and there are available peers, the function tries to
     /// establish new connections. After that it updates the peer scores and discards any records
     /// that no longer need to be stored.
-    async fn heartbeat(&mut self) -> crate::Result<()> {
+    fn heartbeat(&mut self) -> crate::Result<()> {
         let count = std::cmp::min(
             self.peerdb.available_addresses_count(),
             MAX_ACTIVE_CONNECTIONS
@@ -549,7 +545,7 @@ where
         let addresses = self.peerdb.random_known_addresses(count);
 
         for address in addresses {
-            self.connect(address, None).await?;
+            self.connect(address, None)?;
         }
 
         // TODO: update peer scores
@@ -557,7 +553,7 @@ where
         Ok(())
     }
 
-    async fn handle_incoming_request(
+    fn handle_incoming_request(
         &mut self,
         peer_id: T::PeerId,
         request_id: T::PeerRequestId,
@@ -573,12 +569,10 @@ where
                     .filter(|address| self.is_peer_address_valid(address))
                     .collect();
 
-                self.peer_connectivity_handle
-                    .send_response(
-                        request_id,
-                        PeerManagerResponse::AddrListResponse(AddrListResponse { addresses }),
-                    )
-                    .await
+                self.peer_connectivity_handle.send_response(
+                    request_id,
+                    PeerManagerResponse::AddrListResponse(AddrListResponse { addresses }),
+                )
             }
             PeerManagerRequest::AnnounceAddrRequest(AnnounceAddrRequest { address }) => {
                 // TODO: Rate limit announce address requests to prevent DoS attacks.
@@ -594,18 +588,16 @@ where
 
                     let peer_ids = self.random_peer_ids(ANNOUNCED_RESEND_COUNT);
                     for new_peer_id in peer_ids {
-                        self.send_announced_address(new_peer_id, address.clone()).await?;
+                        self.send_announced_address(new_peer_id, address.clone())?;
                     }
                 }
                 Ok(())
             }
             PeerManagerRequest::PingRequest(PingRequest { nonce }) => {
-                self.peer_connectivity_handle
-                    .send_response(
-                        request_id,
-                        PeerManagerResponse::PingResponse(PingResponse { nonce }),
-                    )
-                    .await
+                self.peer_connectivity_handle.send_response(
+                    request_id,
+                    PeerManagerResponse::PingResponse(PingResponse { nonce }),
+                )
             }
         }
     }
@@ -654,7 +646,7 @@ where
     /// # Arguments
     /// `peer_id` - peer ID of the remote peer, if available
     /// `result` - result of the operation that was performed
-    pub async fn handle_result(
+    pub fn handle_result(
         &mut self,
         peer_id: Option<T::PeerId>,
         result: crate::Result<()>,
@@ -670,7 +662,7 @@ where
 
                 if let Some(peer_id) = peer_id {
                     log::info!("adjust peer score for peer {peer_id}, reason {err}");
-                    return self.adjust_peer_score(peer_id, err.ban_score()).await;
+                    return self.adjust_peer_score(peer_id, err.ban_score());
                 }
 
                 Ok(())
@@ -681,19 +673,19 @@ where
     /// Handle control event.
     ///
     /// Handle events from an outside controller (rpc, for example) that sets/gets values for PeerManager
-    async fn handle_control_event(&mut self, event: PeerManagerEvent<T>) -> crate::Result<()> {
+    fn handle_control_event(&mut self, event: PeerManagerEvent<T>) -> crate::Result<()> {
         match event {
             PeerManagerEvent::Connect(address, response) => {
-                self.connect(address, Some(response)).await?;
+                self.connect(address, Some(response))?;
             }
             PeerManagerEvent::Disconnect(peer_id, response) => {
-                self.disconnect(peer_id, Some(response)).await?;
+                self.disconnect(peer_id, Some(response))?;
             }
             PeerManagerEvent::AdjustPeerScore(peer_id, score, response) => {
                 log::debug!("adjust peer {peer_id} score: {score}");
 
                 response
-                    .send(self.adjust_peer_score(peer_id, score).await)
+                    .send(self.adjust_peer_score(peer_id, score))
                     .map_err(|_| P2pError::ChannelClosed)?;
             }
             PeerManagerEvent::GetPeerCount(response) => {
@@ -718,7 +710,7 @@ where
     }
 
     /// Handle connectivity event.
-    async fn handle_connectivity_event_result(
+    fn handle_connectivity_event_result(
         &mut self,
         event_res: crate::Result<ConnectivityEvent<T>>,
     ) -> crate::Result<()> {
@@ -729,7 +721,7 @@ where
                     request_id,
                     request,
                 } => {
-                    self.handle_incoming_request(peer_id, request_id, request).await?;
+                    self.handle_incoming_request(peer_id, request_id, request)?;
                 }
                 net::types::ConnectivityEvent::Response {
                     peer_id,
@@ -745,17 +737,16 @@ where
                 } => {
                     let peer_id = peer_info.peer_id;
 
-                    match self.accept_inbound_connection(address, peer_info, receiver_address).await
-                    {
+                    match self.accept_inbound_connection(address, peer_info, receiver_address) {
                         Ok(_) => {}
                         Err(P2pError::ChannelClosed) => return Err(P2pError::ChannelClosed),
                         Err(P2pError::PeerError(err)) => {
                             log::warn!("peer error for peer {peer_id}: {err}");
-                            self.peer_connectivity_handle.disconnect(peer_id).await?;
+                            self.peer_connectivity_handle.disconnect(peer_id)?;
                         }
                         Err(P2pError::ProtocolError(err)) => {
                             log::warn!("peer error for peer {peer_id}: {err}");
-                            self.adjust_peer_score(peer_id, err.ban_score()).await?;
+                            self.adjust_peer_score(peer_id, err.ban_score())?;
                         }
                         Err(err) => {
                             log::error!("unknown error for peer {peer_id}: {err}");
@@ -768,15 +759,13 @@ where
                     receiver_address,
                 } => {
                     let peer_id = peer_info.peer_id;
-                    let res = self
-                        .accept_connection(
-                            address.clone(),
-                            Role::Outbound,
-                            peer_info,
-                            receiver_address,
-                        )
-                        .await;
-                    self.handle_result(Some(peer_id), res).await?;
+                    let res = self.accept_connection(
+                        address.clone(),
+                        Role::Outbound,
+                        peer_info,
+                        receiver_address,
+                    );
+                    self.handle_result(Some(peer_id), res)?;
 
                     match self.pending_connects.remove(&address) {
                         Some(Some(channel)) => {
@@ -788,20 +777,20 @@ where
                 }
                 net::types::ConnectivityEvent::ConnectionClosed { peer_id } => {
                     let res = self.connection_closed(peer_id);
-                    self.handle_result(Some(peer_id), res).await?;
+                    self.handle_result(Some(peer_id), res)?;
                 }
                 net::types::ConnectivityEvent::ConnectionError { address, error } => {
                     let res = self.handle_outbound_error(address, error);
-                    self.handle_result(None, res).await?;
+                    self.handle_result(None, res)?;
                 }
                 net::types::ConnectivityEvent::Misbehaved { peer_id, error } => {
-                    let res = self.adjust_peer_score(peer_id, error.ban_score()).await;
-                    self.handle_result(Some(peer_id), res).await?;
+                    let res = self.adjust_peer_score(peer_id, error.ban_score());
+                    self.handle_result(Some(peer_id), res)?;
                 }
             },
             Err(err) => {
                 log::error!("failed to read network event: {err:?}");
-                self.handle_result(None, Err(err)).await?;
+                self.handle_result(None, Err(err))?;
             }
         }
 
@@ -823,7 +812,7 @@ where
     /// It can be used to distribute data in the gossip protocol
     /// (for example, to relay announced addresses to a small group of peers).
     pub fn random_peer_ids(&self, count: usize) -> Vec<T::PeerId> {
-        // TODO: Optimise this
+        // TODO: Optimize this
         let all_peer_ids = self.peers.keys().cloned().collect::<Vec<_>>();
         all_peer_ids
             .choose_multiple(&mut make_pseudo_rng(), count)
@@ -837,7 +826,7 @@ where
     }
 
     /// Sends ping requests and disconnects peers that do not respond in time
-    async fn ping_check(&mut self) -> crate::Result<()> {
+    fn ping_check(&mut self) -> crate::Result<()> {
         let now = Instant::now();
         let mut dead_peers = Vec::new();
         for (peer_id, peer) in self.peers.iter_mut() {
@@ -853,12 +842,10 @@ where
                 }
                 None => {
                     let nonce = make_pseudo_rng().gen();
-                    self.peer_connectivity_handle
-                        .send_request(
-                            *peer_id,
-                            PeerManagerRequest::PingRequest(PingRequest { nonce }),
-                        )
-                        .await?;
+                    self.peer_connectivity_handle.send_request(
+                        *peer_id,
+                        PeerManagerRequest::PingRequest(PingRequest { nonce }),
+                    )?;
                     peer.sent_ping = Some(SentPing {
                         nonce,
                         timestamp: now,
@@ -868,7 +855,7 @@ where
         }
 
         for peer_id in dead_peers {
-            self.disconnect(peer_id, None).await?;
+            self.disconnect(peer_id, None)?;
         }
 
         Ok(())
@@ -901,13 +888,13 @@ where
         loop {
             tokio::select! {
                 event = self.rx_peer_manager.recv() => {
-                    self.handle_control_event(event.ok_or(P2pError::ChannelClosed)?).await?;
+                    self.handle_control_event(event.ok_or(P2pError::ChannelClosed)?)?;
                 },
                 event_res = self.peer_connectivity_handle.poll_next() => {
-                    self.handle_connectivity_event_result(event_res).await?;
+                    self.handle_connectivity_event_result(event_res)?;
                 },
                 _event = ping_check_interval.tick(), if ping_check_enabled => {
-                    self.ping_check().await?;
+                    self.ping_check()?;
                 }
                 _event = tokio::time::sleep(PEER_MGR_HEARTBEAT_INTERVAL_MAX) => {}
             }
@@ -915,7 +902,7 @@ where
             // finally update peer manager state
             let now = tokio::time::Instant::now();
             if now.duration_since(self.last_heartbeat) > PEER_MGR_HEARTBEAT_INTERVAL_MIN {
-                self.heartbeat().await?;
+                self.heartbeat()?;
                 self.last_heartbeat = now;
             }
         }
diff --git a/p2p/src/sync/mod.rs b/p2p/src/sync/mod.rs
index 7bb55d4e3..4b529411f 100644
--- a/p2p/src/sync/mod.rs
+++ b/p2p/src/sync/mod.rs
@@ -129,7 +129,6 @@ where
             peer_id,
             SyncRequest::HeaderListRequest(message::HeaderListRequest::new(locator.clone())),
         )
-        .await
         .map(|_| {
             self.peers.insert(
                 peer_id,
@@ -154,7 +153,7 @@ where
 
         // TODO: check if remote has already asked for these headers?
         let headers = self.chainstate_handle.call(move |this| this.get_headers(locator)).await??;
-        self.send_header_response(request_id, headers).await
+        self.send_header_response(request_id, headers)
     }
 
     /// Process block request
@@ -181,7 +180,7 @@ where
             self.chainstate_handle.call(move |this| this.get_block(block_id)).await?;
 
         match block_result {
-            Ok(Some(block)) => self.send_block_response(request_id, vec![block]).await,
+            Ok(Some(block)) => self.send_block_response(request_id, vec![block]),
             Ok(None) => {
                 // TODO: check if remote has already asked for these headers?
                 Err(P2pError::ProtocolError(ProtocolError::InvalidMessage))
@@ -254,7 +253,7 @@ where
         headers: Vec<BlockHeader>,
     ) -> crate::Result<()> {
         match self.validate_header_response(&peer_id, headers).await {
-            Ok(Some(header)) => self.send_block_request(peer_id, header.get_id()).await,
+            Ok(Some(header)) => self.send_block_request(peer_id, header.get_id()),
             Ok(None) => {
                 self.peers
                     .get_mut(&peer_id)
@@ -315,11 +314,11 @@ where
         );
 
         match self.validate_block_response(&peer_id, blocks).await {
-            Ok(Some(next_block)) => self.send_block_request(peer_id, next_block.get_id()).await,
+            Ok(Some(next_block)) => self.send_block_request(peer_id, next_block.get_id()),
             Ok(None) => {
                 // last block from peer received, ask if peer knows of any new headers
                 let locator = self.chainstate_handle.call(|this| this.get_locator()).await??;
-                self.send_header_request(peer_id, locator).await
+                self.send_header_request(peer_id, locator)
             }
             Err(err) => Err(err),
         }
@@ -500,7 +499,7 @@ where
 
                     match self.chainstate_handle.call(move |this| this.get_block(block_id)).await?? {
                         Some(block) => {
-                            let _ = self.peer_sync_handle.make_announcement(Announcement::Block(block)).await.log_err();
+                            let _ = self.peer_sync_handle.make_announcement(Announcement::Block(block)).log_err();
                         }
                         None => log::error!("CRITICAL: best block not available"),
                     }
diff --git a/p2p/src/sync/request.rs b/p2p/src/sync/request.rs
index 537088e2a..120d99445 100644
--- a/p2p/src/sync/request.rs
+++ b/p2p/src/sync/request.rs
@@ -64,12 +64,8 @@ where
     }
 
     /// Sends a request to the given peer.
-    pub async fn send_request(
-        &mut self,
-        peer_id: T::PeerId,
-        request: SyncRequest,
-    ) -> crate::Result<()> {
-        self.peer_sync_handle.send_request(peer_id, request).await.map(|_| ())
+    pub fn send_request(&mut self, peer_id: T::PeerId, request: SyncRequest) -> crate::Result<()> {
+        self.peer_sync_handle.send_request(peer_id, request).map(|_| ())
     }
 
     /// Send block request to remote peer
@@ -81,7 +77,7 @@ where
     /// * `peer_id` - peer ID of the remote node
     /// * `block_id` - ID of the block that is requested
     /// * `retry_count` - how many times the request has been resent
-    pub async fn send_block_request(
+    pub fn send_block_request(
         &mut self,
         peer_id: T::PeerId,
         block_id: Id<Block>,
@@ -95,7 +91,7 @@ where
 
         // send request to remote peer and start tracking its progress
         let wanted_blocks = self.make_block_request(vec![block_id]);
-        self.send_request(peer_id, wanted_blocks).await?;
+        self.send_request(peer_id, wanted_blocks)?;
 
         self.peers
             .get_mut(&peer_id)
@@ -113,7 +109,7 @@ where
     /// * `peer_id` - peer ID of the remote node
     /// * `locator` - local locator object
     /// * `retry_count` - how many times the request has been resent
-    pub async fn send_header_request(
+    pub fn send_header_request(
         &mut self,
         peer_id: T::PeerId,
         locator: Locator,
@@ -127,7 +123,7 @@ where
 
         // send header request and start tracking its progress
         let wanted_headers = self.make_header_request(locator.clone());
-        self.send_request(peer_id, wanted_headers).await?;
+        self.send_request(peer_id, wanted_headers)?;
 
         self.peers
             .get_mut(&peer_id)
@@ -144,7 +140,7 @@ where
     /// # Arguments
     /// * `request_id` - ID of the request that this is a response to
     /// * `headers` - headers that the remote requested
-    pub async fn send_header_response(
+    pub fn send_header_response(
         &mut self,
         request_id: T::PeerRequestId,
         headers: Vec<BlockHeader>,
@@ -153,7 +149,7 @@ where
 
         // TODO: save sent header IDs somewhere and validate future requests against those?
         let message = self.make_header_response(headers);
-        self.peer_sync_handle.send_response(request_id, message).await
+        self.peer_sync_handle.send_response(request_id, message)
     }
 
     /// Send header response to remote peer
@@ -164,7 +160,7 @@ where
     /// # Arguments
     /// * `request_id` - ID of the request that this is a response to
     /// * `headers` - headers that the remote requested
-    pub async fn send_block_response(
+    pub fn send_block_response(
         &mut self,
         request_id: T::PeerRequestId,
         blocks: Vec<Block>,
@@ -173,6 +169,6 @@ where
 
         // TODO: save sent block IDs somewhere and validate future requests against those?
         let message = self.make_block_response(blocks);
-        self.peer_sync_handle.send_response(request_id, message).await
+        self.peer_sync_handle.send_response(request_id, message)
     }
 }


Test patch:
diff --git a/p2p/backend-test-suite/src/ban.rs b/p2p/backend-test-suite/src/ban.rs
index 77b0cbadf..a5713e700 100644
--- a/p2p/backend-test-suite/src/ban.rs
+++ b/p2p/backend-test-suite/src/ban.rs
@@ -98,10 +98,9 @@ where
                 request_id,
                 SyncResponse::HeaderListResponse(HeaderListResponse::new(Vec::new())),
             )
-            .await
             .unwrap();
 
-        sync2.make_announcement(Announcement::Block(blocks[2].clone())).await.unwrap();
+        sync2.make_announcement(Announcement::Block(blocks[2].clone())).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 45c052011..f61b24030 100644
--- a/p2p/backend-test-suite/src/block_announcement.rs
+++ b/p2p/backend-test-suite/src/block_announcement.rs
@@ -85,7 +85,6 @@ where
             )
             .unwrap(),
         ))
-        .await
         .unwrap();
 
     // Poll an event from the network for server2.
@@ -108,7 +107,6 @@ where
             )
             .unwrap(),
         ))
-        .await
         .unwrap();
 
     let block = match sync1.poll_next().await.unwrap() {
@@ -170,7 +168,6 @@ where
             )
             .unwrap(),
         ))
-        .await
         .unwrap();
 }
 
@@ -227,7 +224,7 @@ where
     let encoded_size = message.encode().len();
 
     assert_eq!(
-        sync1.make_announcement(message).await,
+        sync1.make_announcement(message),
         Err(P2pError::PublishError(PublishError::MessageTooLarge(
             encoded_size,
             ANNOUNCEMENT_MAX_SIZE
diff --git a/p2p/backend-test-suite/src/connect.rs b/p2p/backend-test-suite/src/connect.rs
index 08bafba91..86a67b38d 100644
--- a/p2p/backend-test-suite/src/connect.rs
+++ b/p2p/backend-test-suite/src/connect.rs
@@ -101,7 +101,6 @@ where
     .unwrap();
 
     let conn_addr = service1.local_addresses().to_vec();
-    let (res1, res2) = tokio::join!(service1.poll_next(), service2.connect(conn_addr[0].clone()));
-    res1.unwrap();
-    res2.unwrap();
+    service2.connect(conn_addr[0].clone()).unwrap();
+    service1.poll_next().await.unwrap();
 }
diff --git a/p2p/backend-test-suite/src/sync.rs b/p2p/backend-test-suite/src/sync.rs
index e1732c0c4..8bf08f3bb 100644
--- a/p2p/backend-test-suite/src/sync.rs
+++ b/p2p/backend-test-suite/src/sync.rs
@@ -163,7 +163,6 @@ where
                         request_id,
                         SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
-                    .await
                     .unwrap()
             }
             SyncingEvent::Request {
@@ -184,7 +183,6 @@ where
                         request_id,
                         SyncResponse::BlockListResponse(BlockListResponse::new(blocks)),
                     )
-                    .await
                     .unwrap();
             }
             SyncingEvent::Response {
@@ -260,7 +258,6 @@ where
                         request_id,
                         SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
-                    .await
                     .unwrap()
             }
             SyncingEvent::Response {
@@ -282,7 +279,6 @@ where
                             peer_id,
                             SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                         )
-                        .await
                         .unwrap();
                 } else {
                     // all blocks have been downloaded
@@ -307,7 +303,6 @@ where
                         peer_id,
                         SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                     )
-                    .await
                     .unwrap();
             }
             msg => panic!("invalid message received: {msg:?}"),
@@ -385,7 +380,6 @@ where
                         request_id,
                         SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
-                    .await
                     .unwrap()
             }
             SyncingEvent::Request {
@@ -406,7 +400,6 @@ where
                         request_id,
                         SyncResponse::BlockListResponse(BlockListResponse::new(blocks)),
                     )
-                    .await
                     .unwrap();
             }
             SyncingEvent::Response {
@@ -428,7 +421,6 @@ where
                             peer_id,
                             SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                         )
-                        .await
                         .unwrap();
                 }
             }
@@ -449,7 +441,6 @@ where
                         peer_id,
                         SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                     )
-                    .await
                     .unwrap();
             }
             msg => panic!("invalid message received: {msg:?}"),
@@ -529,7 +520,6 @@ where
                         request_id,
                         SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
-                    .await
                     .unwrap()
             }
             SyncingEvent::Request {
@@ -550,7 +540,6 @@ where
                         request_id,
                         SyncResponse::BlockListResponse(BlockListResponse::new(blocks)),
                     )
-                    .await
                     .unwrap();
             }
             SyncingEvent::Response {
@@ -572,7 +561,6 @@ where
                             peer_id,
                             SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                         )
-                        .await
                         .unwrap();
                 }
             }
@@ -593,7 +581,6 @@ where
                         peer_id,
                         SyncRequest::BlockListRequest(BlockListRequest::new(vec![header])),
                     )
-                    .await
                     .unwrap();
             }
             msg => panic!("invalid message received: {msg:?}"),
@@ -688,9 +675,9 @@ where
                 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()
+                    mgr2.handle_mut().send_response(request_id, msg).unwrap()
                 } else {
-                    mgr3.handle_mut().send_response(request_id, msg).await.unwrap()
+                    mgr3.handle_mut().send_response(request_id, msg).unwrap()
                 }
             }
             SyncingEvent::Request {
@@ -708,9 +695,9 @@ where
                     .unwrap()]));
 
                 if dest_peer_id == peer_info21.peer_id {
-                    mgr2.handle_mut().send_response(request_id, msg).await.unwrap();
+                    mgr2.handle_mut().send_response(request_id, msg).unwrap();
                 } else {
-                    mgr3.handle_mut().send_response(request_id, msg).await.unwrap();
+                    mgr3.handle_mut().send_response(request_id, msg).unwrap();
                 }
             }
             SyncingEvent::Response {
@@ -827,9 +814,9 @@ where
                 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()
+                    mgr2.handle_mut().send_response(request_id, msg).unwrap()
                 } else {
-                    mgr3.handle_mut().send_response(request_id, msg).await.unwrap()
+                    mgr3.handle_mut().send_response(request_id, msg).unwrap()
                 }
             }
             SyncingEvent::Request {
@@ -847,9 +834,9 @@ where
                     .unwrap()]));
 
                 if dest_peer_id == peer_info21.peer_id {
-                    mgr2.handle_mut().send_response(request_id, msg).await.unwrap();
+                    mgr2.handle_mut().send_response(request_id, msg).unwrap();
                 } else {
-                    mgr3.handle_mut().send_response(request_id, msg).await.unwrap();
+                    mgr3.handle_mut().send_response(request_id, msg).unwrap();
                 }
             }
             SyncingEvent::Response {
@@ -964,9 +951,9 @@ where
                 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()
+                    mgr2.handle_mut().send_response(request_id, msg).unwrap()
                 } else {
-                    mgr3.handle_mut().send_response(request_id, msg).await.unwrap()
+                    mgr3.handle_mut().send_response(request_id, msg).unwrap()
                 }
 
                 if gethdr_received.insert(dest_peer_id) {
@@ -994,9 +981,9 @@ where
                     .unwrap()]));
 
                 if dest_peer_id == peer_info21.peer_id {
-                    mgr2.handle_mut().send_response(request_id, msg).await.unwrap();
+                    mgr2.handle_mut().send_response(request_id, msg).unwrap();
                 } else {
-                    mgr3.handle_mut().send_response(request_id, msg).await.unwrap();
+                    mgr3.handle_mut().send_response(request_id, msg).unwrap();
                 }
             }
             SyncingEvent::Response {
@@ -1061,7 +1048,7 @@ where
     assert!(!mgr1_handle.call(|c| c.is_initial_block_download()).await.unwrap().unwrap());
 
     mgr1.unregister_peer(peer_info2.peer_id);
-    assert_eq!(conn1.disconnect(peer_info2.peer_id).await, Ok(()));
+    assert_eq!(conn1.disconnect(peer_info2.peer_id), Ok(()));
 
     let event = get_connectivity_event::<N>(&mut conn2).await;
     assert!(std::matches!(
@@ -1103,7 +1090,6 @@ where
                         request_id,
                         SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
                     )
-                    .await
                     .unwrap()
             }
             SyncingEvent::Request {
@@ -1124,7 +1110,6 @@ where
                         request_id,
                         SyncResponse::BlockListResponse(BlockListResponse::new(blocks)),
                     )
-                    .await
                     .unwrap();
             }
             SyncingEvent::Response {
@@ -1267,12 +1252,10 @@ where
                 .await
                 .unwrap()
                 .unwrap();
-            mgr.handle_mut()
-                .send_response(
-                    request_id,
-                    SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
-                )
-                .await
+            mgr.handle_mut().send_response(
+                request_id,
+                SyncResponse::HeaderListResponse(HeaderListResponse::new(headers)),
+            )
         }
         _ => panic!("invalid message"),
     }
diff --git a/p2p/src/peer_manager/tests/ban.rs b/p2p/src/peer_manager/tests/ban.rs
index 4825c18fd..11a87832f 100644
--- a/p2p/src/peer_manager/tests/ban.rs
+++ b/p2p/src/peer_manager/tests/ban.rs
@@ -61,9 +61,9 @@ where
     )
     .await;
     let peer_id = peer_info.peer_id;
-    pm2.accept_inbound_connection(address, peer_info, None).await.unwrap();
+    pm2.accept_inbound_connection(address, peer_info, None).unwrap();
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
+    assert_eq!(pm2.adjust_peer_score(peer_id, 1000), Ok(()));
     let addr1 = pm1.peer_connectivity_handle.local_addresses()[0].clone().as_bannable();
     assert!(pm2.peerdb.is_address_banned(&addr1).unwrap());
     let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
@@ -108,9 +108,9 @@ where
     )
     .await;
     let peer_id = peer_info.peer_id;
-    pm2.accept_inbound_connection(address, peer_info, None).await.unwrap();
+    pm2.accept_inbound_connection(address, peer_info, None).unwrap();
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
+    assert_eq!(pm2.adjust_peer_score(peer_id, 1000), Ok(()));
     let addr1 = pm1.peer_connectivity_handle.local_addresses()[0].clone().as_bannable();
     assert!(pm2.peerdb.is_address_banned(&addr1).unwrap());
     let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
@@ -163,14 +163,14 @@ where
     )
     .await;
     let peer_id = peer_info1.peer_id;
-    pm2.accept_inbound_connection(address, peer_info1, None).await.unwrap();
+    pm2.accept_inbound_connection(address, peer_info1, None).unwrap();
 
     let remote_addr = pm1.peer_connectivity_handle.local_addresses()[0].clone();
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 10).await, Ok(()));
+    assert_eq!(pm2.adjust_peer_score(peer_id, 10), Ok(()));
     assert!(!pm2.peerdb.is_address_banned(&remote_addr.as_bannable()).unwrap());
 
-    assert_eq!(pm2.adjust_peer_score(peer_id, 90).await, Ok(()));
+    assert_eq!(pm2.adjust_peer_score(peer_id, 90), Ok(()));
     assert!(pm2.peerdb.is_address_banned(&remote_addr.as_bannable()).unwrap());
 
     let event = get_connectivity_event::<T>(&mut pm2.peer_connectivity_handle).await;
@@ -178,10 +178,10 @@ where
         Ok(net::types::ConnectivityEvent::ConnectionClosed { .. }) => {}
         _ => panic!("unexpected event: {event:?}"),
     }
-    pm2.handle_connectivity_event_result(event).await.unwrap();
+    pm2.handle_connectivity_event_result(event).unwrap();
 
     let (tx, rx) = oneshot::channel();
-    pm2.connect(remote_addr, Some(tx)).await.unwrap();
+    pm2.connect(remote_addr, Some(tx)).unwrap();
     let res = rx.await.unwrap();
     match res {
         Err(P2pError::PeerError(PeerError::BannedAddress(_))) => {}
@@ -220,65 +220,53 @@ where
         make_peer_manager::<S>(A::make_transport(), A::make_address(), Arc::clone(&config)).await;
 
     // invalid magic bytes
-    let res = peer_manager
-        .accept_connection(
-            B::new(),
-            Role::Outbound,
-            net::types::PeerInfo::<S::PeerId> {
-                peer_id,
-                network: [1, 2, 3, 4],
-                version: SemVer::new(0, 1, 0),
-                agent: None,
-                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
-                    .into_iter()
-                    .collect(),
-            },
-            None,
-        )
-        .await;
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
+    let res = peer_manager.accept_connection(
+        B::new(),
+        Role::Outbound,
+        net::types::PeerInfo::<S::PeerId> {
+            peer_id,
+            network: [1, 2, 3, 4],
+            version: SemVer::new(0, 1, 0),
+            agent: None,
+            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
+        },
+        None,
+    );
+    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
     assert!(!peer_manager.is_peer_connected(&peer_id));
 
     // invalid version
-    let res = peer_manager
-        .accept_connection(
-            B::new(),
-            Role::Outbound,
-            net::types::PeerInfo::<S::PeerId> {
-                peer_id,
-                network: *config.magic_bytes(),
-                version: SemVer::new(1, 1, 1),
-                agent: None,
-                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
-                    .into_iter()
-                    .collect(),
-            },
-            None,
-        )
-        .await;
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
+    let res = peer_manager.accept_connection(
+        B::new(),
+        Role::Outbound,
+        net::types::PeerInfo::<S::PeerId> {
+            peer_id,
+            network: *config.magic_bytes(),
+            version: SemVer::new(1, 1, 1),
+            agent: None,
+            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
+        },
+        None,
+    );
+    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
     assert!(!peer_manager.is_peer_connected(&peer_id));
 
     // valid connection
     let address = B::new();
-    let res = peer_manager
-        .accept_connection(
-            address.clone(),
-            Role::Outbound,
-            net::types::PeerInfo::<S::PeerId> {
-                peer_id,
-                network: *config.magic_bytes(),
-                version: SemVer::new(0, 1, 0),
-                agent: None,
-                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
-                    .into_iter()
-                    .collect(),
-            },
-            None,
-        )
-        .await;
+    let res = peer_manager.accept_connection(
+        address.clone(),
+        Role::Outbound,
+        net::types::PeerInfo::<S::PeerId> {
+            peer_id,
+            network: *config.magic_bytes(),
+            version: SemVer::new(0, 1, 0),
+            agent: None,
+            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
+        },
+        None,
+    );
     assert!(res.is_ok());
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
+    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
     assert!(peer_manager.is_peer_connected(&peer_id));
     assert!(!peer_manager.peerdb.is_address_banned(&address.as_bannable()).unwrap());
 }
@@ -325,62 +313,50 @@ where
         make_peer_manager::<S>(A::make_transport(), A::make_address(), Arc::clone(&config)).await;
 
     // invalid magic bytes
-    let res = peer_manager
-        .accept_inbound_connection(
-            B::new(),
-            net::types::PeerInfo::<S::PeerId> {
-                peer_id,
-                network: [1, 2, 3, 4],
-                version: SemVer::new(0, 1, 0),
-                agent: None,
-                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
-                    .into_iter()
-                    .collect(),
-            },
-            None,
-        )
-        .await;
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
+    let res = peer_manager.accept_inbound_connection(
+        B::new(),
+        net::types::PeerInfo::<S::PeerId> {
+            peer_id,
+            network: [1, 2, 3, 4],
+            version: SemVer::new(0, 1, 0),
+            agent: None,
+            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
+        },
+        None,
+    );
+    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
     assert!(!peer_manager.is_peer_connected(&peer_id));
 
     // invalid version
-    let res = peer_manager
-        .accept_inbound_connection(
-            B::new(),
-            net::types::PeerInfo::<S::PeerId> {
-                peer_id,
-                network: *config.magic_bytes(),
-                version: SemVer::new(1, 1, 1),
-                agent: None,
-                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
-                    .into_iter()
-                    .collect(),
-            },
-            None,
-        )
-        .await;
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
+    let res = peer_manager.accept_inbound_connection(
+        B::new(),
+        net::types::PeerInfo::<S::PeerId> {
+            peer_id,
+            network: *config.magic_bytes(),
+            version: SemVer::new(1, 1, 1),
+            agent: None,
+            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
+        },
+        None,
+    );
+    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
     assert!(!peer_manager.is_peer_connected(&peer_id));
 
     // valid connection
     let address = B::new();
-    let res = peer_manager
-        .accept_inbound_connection(
-            address.clone(),
-            net::types::PeerInfo::<S::PeerId> {
-                peer_id,
-                network: *config.magic_bytes(),
-                version: SemVer::new(0, 1, 0),
-                agent: None,
-                subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions]
-                    .into_iter()
-                    .collect(),
-            },
-            None,
-        )
-        .await;
+    let res = peer_manager.accept_inbound_connection(
+        address.clone(),
+        net::types::PeerInfo::<S::PeerId> {
+            peer_id,
+            network: *config.magic_bytes(),
+            version: SemVer::new(0, 1, 0),
+            agent: None,
+            subscriptions: [PubSubTopic::Blocks, PubSubTopic::Transactions].into_iter().collect(),
+        },
+        None,
+    );
     assert!(res.is_ok());
-    assert_eq!(peer_manager.handle_result(Some(peer_id), res).await, Ok(()));
+    assert_eq!(peer_manager.handle_result(Some(peer_id), res), Ok(()));
     assert!(peer_manager.is_peer_connected(&peer_id));
     assert!(!peer_manager.peerdb.is_address_banned(&address.as_bannable()).unwrap());
 }
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index 19ecc1a3a..e9df49a46 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -60,7 +60,7 @@ async fn test_peer_manager_connect<T: NetworkingService>(
     let config = Arc::new(config::create_mainnet());
     let mut peer_manager = make_peer_manager::<T>(transport, bind_addr, config).await;
 
-    peer_manager.try_connect(remote_addr).await.unwrap();
+    peer_manager.try_connect(remote_addr).unwrap();
 
     assert!(matches!(
         peer_manager.peer_connectivity_handle.poll_next().await,
@@ -124,7 +124,7 @@ where
 
     // "discover" the other networking service
     pm1.peerdb.peer_discovered(&addr).unwrap();
-    pm1.heartbeat().await.unwrap();
+    pm1.heartbeat().unwrap();
 
     assert_eq!(pm1.pending_connects.len(), 1);
     assert!(std::matches!(
@@ -260,7 +260,7 @@ where
     )
     .await;
     assert_eq!(
-        pm2.accept_inbound_connection(address, peer_info, None).await,
+        pm2.accept_inbound_connection(address, peer_info, None),
         Ok(())
     );
 }
@@ -315,7 +315,7 @@ where
     .await;
 
     assert_eq!(
-        pm2.accept_inbound_connection(address, peer_info, None).await,
+        pm2.accept_inbound_connection(address, peer_info, None),
         Err(P2pError::ProtocolError(ProtocolError::DifferentNetwork(
             [1, 2, 3, 4],
             *config::create_mainnet().magic_bytes(),
@@ -379,7 +379,7 @@ where
     .await;
 
     assert_eq!(
-        pm2.peer_connectivity_handle.disconnect(peer_info.peer_id).await,
+        pm2.peer_connectivity_handle.disconnect(peer_info.peer_id),
         Ok(())
     );
     assert!(std::matches!(
@@ -419,7 +419,7 @@ where
     let mut pm2 = make_peer_manager::<T>(A::make_transport(), addr2, Arc::clone(&config)).await;
 
     for peer in peers.into_iter() {
-        pm1.accept_connection(peer.0, Role::Inbound, peer.1, None).await.unwrap();
+        pm1.accept_connection(peer.0, Role::Inbound, peer.1, None).unwrap();
     }
     assert_eq!(
         pm1.active_peer_count(),
@@ -541,7 +541,7 @@ where
     .unwrap();
 
     // This will fail immediately because it is trying to connect to the closed port
-    conn.connect(addr2).await.expect("dial to succeed");
+    conn.connect(addr2).expect("dial to succeed");
 
     match timeout(Duration::from_secs(1), conn.poll_next()).await {
         Ok(res) => assert!(std::matches!(
diff --git a/p2p/src/peer_manager/tests/ping.rs b/p2p/src/peer_manager/tests/ping.rs
index 6091af9e1..46a209927 100644
--- a/p2p/src/peer_manager/tests/ping.rs
+++ b/p2p/src/peer_manager/tests/ping.rs
@@ -42,8 +42,8 @@ async fn ping_timeout() {
     let ping_check_period = *p2p_config.ping_check_period;
     let ping_timeout = *p2p_config.ping_timeout;
 
-    let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::channel(10);
-    let (conn_tx, conn_rx) = tokio::sync::mpsc::channel(10);
+    let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel();
+    let (conn_tx, conn_rx) = tokio::sync::mpsc::unbounded_channel();
     let (_peer_tx, peer_rx) =
         tokio::sync::mpsc::unbounded_channel::<PeerManagerEvent<TestNetworkingService>>();
     let time_getter = P2pTestTimeGetter::new();
@@ -82,7 +82,6 @@ async fn ping_timeout() {
             },
             receiver_address: None,
         })
-        .await
         .unwrap();
 
     // Receive ping requests and send responses normally
@@ -102,7 +101,6 @@ async fn ping_timeout() {
                         request_id,
                         response: PeerManagerResponse::PingResponse(PingResponse { nonce }),
                     })
-                    .await
                     .unwrap();
             }
             _ => panic!("unexpected event: {event:?}"),
@@ -126,9 +124,8 @@ async fn ping_timeout() {
     // PeerManager should ask backend to close connection
     let event = cmd_rx.recv().await.unwrap();
     match event {
-        Command::Disconnect { peer_id, response } => {
-            response.send(Ok(())).unwrap();
-            conn_tx.send(ConnectivityEvent::ConnectionClosed { peer_id }).await.unwrap();
+        Command::Disconnect { peer_id } => {
+            conn_tx.send(ConnectivityEvent::ConnectionClosed { peer_id }).unwrap();
         }
         _ => panic!("unexpected event: {event:?}"),
     }
diff --git a/p2p/src/sync/tests/request_response.rs b/p2p/src/sync/tests/request_response.rs
index 7ef81922d..b34f2e80f 100644
--- a/p2p/src/sync/tests/request_response.rs
+++ b/p2p/src/sync/tests/request_response.rs
@@ -56,7 +56,6 @@ where
             peer_info2.peer_id,
             SyncRequest::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
         )
-        .await
         .unwrap();
 
     if let Ok(SyncingEvent::Request {
@@ -75,7 +74,6 @@ where
                 request_id,
                 SyncResponse::HeaderListResponse(HeaderListResponse::new(vec![])),
             )
-            .await
             .unwrap();
     } else {
         panic!("invalid data received");
@@ -123,7 +121,6 @@ where
             peer_info2.peer_id,
             SyncRequest::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
         )
-        .await
         .unwrap();
     request_ids.insert(id);
 
@@ -133,7 +130,6 @@ where
             peer_info2.peer_id,
             SyncRequest::HeaderListRequest(HeaderListRequest::new(Locator::new(vec![]))),
         )
-        .await
         .unwrap();
     request_ids.insert(id);
 
@@ -148,7 +144,6 @@ where
                             request_id,
                             SyncResponse::HeaderListResponse(HeaderListResponse::new(vec![])),
                         )
-                        .await
                         .unwrap();
                 }
                 _ => panic!("invalid event: {event:?}"),
diff --git a/p2p/src/testing_utils.rs b/p2p/src/testing_utils.rs
index 93a32c0d3..44add2d7c 100644
--- a/p2p/src/testing_utils.rs
+++ b/p2p/src/testing_utils.rs
@@ -161,7 +161,7 @@ where
     T::ConnectivityHandle: ConnectivityService<T>,
 {
     let addr = conn2.local_addresses();
-    conn1.connect(addr[0].clone()).await.expect("dial to succeed");
+    conn1.connect(addr[0].clone()).expect("dial to succeed");
 
     let (address, peer_info1) = match timeout(Duration::from_secs(5), conn2.poll_next()).await {
         Ok(event) => match event.unwrap() {
diff --git a/p2p/tests/block_announcement.rs b/p2p/tests/block_announcement.rs
index 9a88db287..c63262fbe 100644
--- a/p2p/tests/block_announcement.rs
+++ b/p2p/tests/block_announcement.rs
@@ -92,7 +92,6 @@ where
             )
             .unwrap(),
         ))
-        .await
         .unwrap();
 
     // Verify that all peers received the message even though they weren't directly connected.