Repository: mintlayer/mintlayer-core
Pull request #668 (2023-02-02 08:05:54)

Update channel type used in `PeerManagerEvent`
I think using the `oneshot` channel in `PeerManagerEvent` is not very safe.
It's easy to imagine a situation where the receiver is dropped before the event is processed.
For example here:

```
#[async_trait::async_trait]
impl<T> P2pInterface for P2p<T>
where
    T: NetworkingService,
{
    async fn connect(&mut self, addr: String) -> crate::Result<()> {
        let (tx, rx) = oneshot_nofail::channel();
        let addr = addr
            .parse::<T::Address>()
            .map_err(|_| P2pError::ConversionError(ConversionError::InvalidAddress(addr)))?;
        self.tx_peer_manager
            .send(PeerManagerEvent::Connect(addr, tx))
            .map_err(|_| P2pError::ChannelClosed)?;
        rx.await.map_err(P2pError::from)?
    }
...    
```

Here `connect` is async and so can be cancelled at any `await` point (I think we are not doing this now, but this could easily change later).
And yet, sending to a closed channel in `PeerManager` is considered a fatal error and `PeerManager` would stop working.
One fix would be to review all `oneshot` usages and fix them accordingly. However, I decided to write a small wrapper for the `oneshot` channel that ignores send errors, as this should be safer.
It would be nice if you could suggest a better name (than `oneshot_nofail`) for the new wrapper.


Patch:
diff --git a/p2p/src/event.rs b/p2p/src/event.rs
index 7bd37ef6b..f009a7975 100644
--- a/p2p/src/event.rs
+++ b/p2p/src/event.rs
@@ -13,31 +13,29 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use tokio::sync::oneshot;
-
 use common::chain::block::Block;
 
-use crate::{interface::types::ConnectedPeer, net::NetworkingService};
+use crate::{interface::types::ConnectedPeer, net::NetworkingService, utils::oneshot_nofail};
 
 #[derive(Debug)]
 pub enum PeerManagerEvent<T: NetworkingService> {
     /// Try to establish connection with a remote peer
-    Connect(T::Address, oneshot::Sender<crate::Result<()>>),
+    Connect(T::Address, oneshot_nofail::Sender<crate::Result<()>>),
 
     /// Disconnect node using peer ID
-    Disconnect(T::PeerId, oneshot::Sender<crate::Result<()>>),
+    Disconnect(T::PeerId, oneshot_nofail::Sender<crate::Result<()>>),
 
     /// Get the total number of peers local node has a connection with
-    GetPeerCount(oneshot::Sender<usize>),
+    GetPeerCount(oneshot_nofail::Sender<usize>),
 
     /// Get the bind address of the local node
-    GetBindAddresses(oneshot::Sender<Vec<String>>),
+    GetBindAddresses(oneshot_nofail::Sender<Vec<String>>),
 
     /// Get peer IDs and addresses of connected peers
-    GetConnectedPeers(oneshot::Sender<Vec<ConnectedPeer>>),
+    GetConnectedPeers(oneshot_nofail::Sender<Vec<ConnectedPeer>>),
 
     /// Adjust peer score
-    AdjustPeerScore(T::PeerId, u32, oneshot::Sender<crate::Result<()>>),
+    AdjustPeerScore(T::PeerId, u32, oneshot_nofail::Sender<crate::Result<()>>),
 }
 
 #[derive(Debug)]
diff --git a/p2p/src/interface/p2p_interface_impl.rs b/p2p/src/interface/p2p_interface_impl.rs
index ffe4a4581..b4cedfdda 100644
--- a/p2p/src/interface/p2p_interface_impl.rs
+++ b/p2p/src/interface/p2p_interface_impl.rs
@@ -13,14 +13,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::str::FromStr;
-
-use tokio::sync::oneshot;
-
 use crate::{
     error::{ConversionError, P2pError},
     event::PeerManagerEvent,
     net::NetworkingService,
+    utils::oneshot_nofail,
     P2p,
 };
 
@@ -31,12 +28,8 @@ impl<T> P2pInterface for P2p<T>
 where
     T: NetworkingService,
 {
-    async fn connect(&mut self, addr: String) -> crate::Result<()>
-    where
-        <T as NetworkingService>::Address: FromStr,
-        <T as NetworkingService>::PeerId: FromStr,
-    {
-        let (tx, rx) = oneshot::channel();
+    async fn connect(&mut self, addr: String) -> crate::Result<()> {
+        let (tx, rx) = oneshot_nofail::channel();
         let addr = addr
             .parse::<T::Address>()
             .map_err(|_| P2pError::ConversionError(ConversionError::InvalidAddress(addr)))?;
@@ -46,11 +39,8 @@ where
         rx.await.map_err(P2pError::from)?
     }
 
-    async fn disconnect(&mut self, peer_id: String) -> crate::Result<()>
-    where
-        <T as NetworkingService>::PeerId: FromStr,
-    {
-        let (tx, rx) = oneshot::channel();
+    async fn disconnect(&mut self, peer_id: String) -> crate::Result<()> {
+        let (tx, rx) = oneshot_nofail::channel();
         let peer_id = peer_id
             .parse::<T::PeerId>()
             .map_err(|_| P2pError::ConversionError(ConversionError::InvalidPeerId(peer_id)))?;
@@ -62,7 +52,7 @@ where
     }
 
     async fn get_peer_count(&self) -> crate::Result<usize> {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = oneshot_nofail::channel();
         self.tx_peer_manager
             .send(PeerManagerEvent::GetPeerCount(tx))
             .map_err(P2pError::from)?;
@@ -70,7 +60,7 @@ where
     }
 
     async fn get_bind_addresses(&self) -> crate::Result<Vec<String>> {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = oneshot_nofail::channel();
         self.tx_peer_manager
             .send(PeerManagerEvent::GetBindAddresses(tx))
             .map_err(P2pError::from)?;
@@ -78,7 +68,7 @@ where
     }
 
     async fn get_connected_peers(&self) -> crate::Result<Vec<ConnectedPeer>> {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = oneshot_nofail::channel();
         self.tx_peer_manager
             .send(PeerManagerEvent::GetConnectedPeers(tx))
             .map_err(P2pError::from)?;
diff --git a/p2p/src/lib.rs b/p2p/src/lib.rs
index 26887ff80..c41000333 100644
--- a/p2p/src/lib.rs
+++ b/p2p/src/lib.rs
@@ -26,6 +26,7 @@ pub mod sync;
 #[cfg(feature = "testing_utils")]
 pub mod testing_utils;
 pub mod types;
+pub mod utils;
 
 use std::sync::Arc;
 
diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs
index 7789f96f0..c4f2b1597 100644
--- a/p2p/src/peer_manager/mod.rs
+++ b/p2p/src/peer_manager/mod.rs
@@ -30,10 +30,7 @@ use std::{
 };
 
 use crypto::random::{make_pseudo_rng, Rng, SliceRandom};
-use tokio::{
-    sync::{mpsc, oneshot},
-    time::Instant,
-};
+use tokio::{sync::mpsc, time::Instant};
 
 use chainstate::ban_score::BanScore;
 use common::{chain::ChainConfig, primitives::semver::SemVer, time_getter::TimeGetter};
@@ -57,6 +54,7 @@ use crate::{
         AsBannableAddress, ConnectivityService, NetworkingService,
     },
     types::peer_address::{PeerAddress, PeerAddressIp4, PeerAddressIp6},
+    utils::oneshot_nofail,
 };
 
 use self::{
@@ -99,10 +97,10 @@ where
     tx_sync: mpsc::UnboundedSender<SyncControlEvent<T>>,
 
     /// Hashmap of pending outbound connections
-    pending_connects: HashMap<T::Address, Option<oneshot::Sender<crate::Result<()>>>>,
+    pending_connects: HashMap<T::Address, Option<oneshot_nofail::Sender<crate::Result<()>>>>,
 
     /// Hashmap of pending disconnect requests
-    pending_disconnects: HashMap<T::PeerId, Option<oneshot::Sender<crate::Result<()>>>>,
+    pending_disconnects: HashMap<T::PeerId, Option<oneshot_nofail::Sender<crate::Result<()>>>>,
 
     /// Set of active peers
     peers: BTreeMap<T::PeerId, PeerContext<T>>,
@@ -368,7 +366,7 @@ where
             self.tx_sync.send(SyncControlEvent::Disconnected(peer_id))?;
 
             if let Some(Some(response)) = self.pending_disconnects.remove(&peer_id) {
-                response.send(Ok(())).map_err(|_| P2pError::ChannelClosed)?;
+                response.send(Ok(()));
             }
 
             self.peerdb.peer_disconnected(peer.address);
@@ -415,7 +413,7 @@ where
     /// update its own records.
     fn handle_outbound_error(&mut self, address: T::Address, error: P2pError) -> crate::Result<()> {
         if let Some(Some(channel)) = self.pending_connects.remove(&address) {
-            channel.send(Err(error)).map_err(|_| P2pError::ChannelClosed)?;
+            channel.send(Err(error));
         }
 
         self.peerdb.report_outbound_failure(address);
@@ -451,7 +449,7 @@ where
     fn connect(
         &mut self,
         address: T::Address,
-        response: Option<oneshot::Sender<crate::Result<()>>>,
+        response: Option<oneshot_nofail::Sender<crate::Result<()>>>,
     ) -> crate::Result<()> {
         log::debug!("try to establish outbound connection to peer at address {address:?}");
 
@@ -463,7 +461,7 @@ where
             }
             Err(e) => {
                 if let Some(response) = response {
-                    response.send(Err(e)).map_err(|_| P2pError::ChannelClosed)?;
+                    response.send(Err(e));
                 }
             }
         }
@@ -493,7 +491,7 @@ where
     fn disconnect(
         &mut self,
         peer_id: T::PeerId,
-        response: Option<oneshot::Sender<crate::Result<()>>>,
+        response: Option<oneshot_nofail::Sender<crate::Result<()>>>,
     ) -> crate::Result<()> {
         log::debug!("disconnect peer {peer_id}");
 
@@ -505,7 +503,7 @@ where
             }
             Err(e) => {
                 if let Some(response) = response {
-                    response.send(Err(e)).map_err(|_| P2pError::ChannelClosed)?;
+                    response.send(Err(e));
                 }
             }
         }
@@ -684,12 +682,10 @@ where
             PeerManagerEvent::AdjustPeerScore(peer_id, score, response) => {
                 log::debug!("adjust peer {peer_id} score: {score}");
 
-                response
-                    .send(self.adjust_peer_score(peer_id, score))
-                    .map_err(|_| P2pError::ChannelClosed)?;
+                response.send(self.adjust_peer_score(peer_id, score));
             }
             PeerManagerEvent::GetPeerCount(response) => {
-                response.send(self.active_peer_count()).map_err(|_| P2pError::ChannelClosed)?;
+                response.send(self.active_peer_count());
             }
             PeerManagerEvent::GetBindAddresses(response) => {
                 let addr = self
@@ -698,11 +694,11 @@ where
                     .iter()
                     .map(|addr| addr.to_string())
                     .collect();
-                response.send(addr).map_err(|_| P2pError::ChannelClosed)?;
+                response.send(addr);
             }
             PeerManagerEvent::GetConnectedPeers(response) => {
                 let peers = self.get_connected_peers();
-                response.send(peers).map_err(|_| P2pError::ChannelClosed)?
+                response.send(peers);
             }
         }
 
@@ -769,7 +765,7 @@ where
 
                     match self.pending_connects.remove(&address) {
                         Some(Some(channel)) => {
-                            channel.send(Ok(())).map_err(|_| P2pError::ChannelClosed)?
+                            channel.send(Ok(()));
                         }
                         Some(None) => {}
                         None => log::error!("connection accepted but it's not pending?"),
diff --git a/p2p/src/sync/mod.rs b/p2p/src/sync/mod.rs
index 4b529411f..31b6510f5 100644
--- a/p2p/src/sync/mod.rs
+++ b/p2p/src/sync/mod.rs
@@ -22,7 +22,7 @@ mod request;
 
 use std::{collections::HashMap, sync::Arc};
 
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::mpsc;
 use void::Void;
 
 use chainstate::{ban_score::BanScore, chainstate_interface, BlockError, ChainstateError, Locator};
@@ -42,6 +42,7 @@ use crate::{
     event::{PeerManagerEvent, SyncControlEvent},
     message::{self, Announcement, SyncRequest},
     net::{types::SyncingEvent, NetworkingService, SyncingMessagingService},
+    utils::oneshot_nofail,
 };
 
 // TODO: from config? global constant?
@@ -379,7 +380,7 @@ where
             Err(P2pError::ProtocolError(err)) => {
                 log::error!("Peer {peer_id} committed a protocol error: {err}");
 
-                let (tx, rx) = oneshot::channel();
+                let (tx, rx) = oneshot_nofail::channel();
                 self.tx_peer_manager
                     .send(PeerManagerEvent::AdjustPeerScore(
                         peer_id,
@@ -392,7 +393,7 @@ where
             Err(P2pError::ChainstateError(err)) => match err {
                 ChainstateError::ProcessBlockError(err) => {
                     if err.ban_score() > 0 {
-                        let (tx, rx) = oneshot::channel();
+                        let (tx, rx) = oneshot_nofail::channel();
                         self.tx_peer_manager
                             .send(PeerManagerEvent::AdjustPeerScore(
                                 peer_id,
@@ -419,7 +420,7 @@ where
 
                 if err.ban_score() > 0 {
                     // TODO: better abstraction over channels
-                    let (tx, rx) = oneshot::channel();
+                    let (tx, rx) = oneshot_nofail::channel();
                     self.tx_peer_manager
                         .send(PeerManagerEvent::AdjustPeerScore(
                             peer_id,
@@ -563,7 +564,7 @@ where
 
         if score > 0 {
             // TODO: better abstraction over channels
-            let (tx, rx) = oneshot::channel();
+            let (tx, rx) = oneshot_nofail::channel();
             self.tx_peer_manager
                 .send(PeerManagerEvent::AdjustPeerScore(peer_id, score, tx))
                 .map_err(P2pError::from)?;
diff --git a/p2p/src/utils/mod.rs b/p2p/src/utils/mod.rs
new file mode 100644
index 000000000..0e6a3a94b
--- /dev/null
+++ b/p2p/src/utils/mod.rs
@@ -0,0 +1,16 @@
+// Copyright (c) 2023 RBB S.r.l
+// opensource@mintlayer.org
+// SPDX-License-Identifier: MIT
+// Licensed under the MIT License;
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod oneshot_nofail;
diff --git a/p2p/src/utils/oneshot_nofail.rs b/p2p/src/utils/oneshot_nofail.rs
new file mode 100644
index 000000000..6f201f673
--- /dev/null
+++ b/p2p/src/utils/oneshot_nofail.rs
@@ -0,0 +1,53 @@
+// Copyright (c) 2023 RBB S.r.l
+// opensource@mintlayer.org
+// SPDX-License-Identifier: MIT
+// Licensed under the MIT License;
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Simple wrapper for the `tokio::sync::oneshot` channel
+//! that does not return an error if the receiver is disconnected.
+//!
+//! The wrapper could be used when sending to a closed channel is not considered an error
+//! (for example, when the async receiver was canceled for some reason).
+
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use tokio::sync::oneshot;
+
+#[derive(Debug)]
+pub struct Sender<T>(oneshot::Sender<T>);
+
+impl<T> Sender<T> {
+    pub fn send(self, t: T) {
+        let _ = self.0.send(t);
+    }
+}
+
+#[derive(Debug)]
+pub struct Receiver<T>(oneshot::Receiver<T>);
+
+impl<T> Future for Receiver<T> {
+    type Output = Result<T, oneshot::error::RecvError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut self.0).poll(cx)
+    }
+}
+
+pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
+    let (sender, receiver) = oneshot::channel();
+    (Sender(sender), Receiver(receiver))
+}


Test patch:
diff --git a/p2p/src/peer_manager/tests/ban.rs b/p2p/src/peer_manager/tests/ban.rs
index 11a87832f..cc27e0b0e 100644
--- a/p2p/src/peer_manager/tests/ban.rs
+++ b/p2p/src/peer_manager/tests/ban.rs
@@ -22,9 +22,9 @@ use crate::{
         TestTcpAddressMaker, TestTransportChannel, TestTransportMaker, TestTransportNoise,
         TestTransportTcp,
     },
+    utils::oneshot_nofail,
 };
 use common::{chain::config, primitives::semver::SemVer};
-use tokio::sync::oneshot;
 
 use crate::{
     error::{P2pError, PeerError},
@@ -180,7 +180,7 @@ where
     }
     pm2.handle_connectivity_event_result(event).unwrap();
 
-    let (tx, rx) = oneshot::channel();
+    let (tx, rx) = oneshot_nofail::channel();
     pm2.connect(remote_addr, Some(tx)).unwrap();
     let res = rx.await.unwrap();
     match res {
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index e9df49a46..17daa1dbb 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -19,7 +19,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use tokio::{sync::oneshot, time::timeout};
+use tokio::time::timeout;
 
 use crate::{
     config::P2pConfig,
@@ -29,6 +29,7 @@ use crate::{
         connect_services, get_connectivity_event, peerdb_inmemory_store, P2pTestTimeGetter,
         TestTransportChannel, TestTransportMaker, TestTransportNoise, TestTransportTcp,
     },
+    utils::oneshot_nofail,
 };
 use common::chain::config;
 
@@ -623,7 +624,7 @@ async fn connection_timeout_rpc_notified<T>(
         peer_manager.run().await.unwrap();
     });
 
-    let (rtx, rrx) = oneshot::channel();
+    let (rtx, rrx) = oneshot_nofail::channel();
     tx.send(PeerManagerEvent::Connect(addr2, rtx)).unwrap();
 
     match timeout(*p2p_config.outbound_connection_timeout, rrx).await {
@@ -697,7 +698,7 @@ where
     .await;
 
     // Get the first peer manager's bind address
-    let (rtx, rrx) = oneshot::channel();
+    let (rtx, rrx) = oneshot_nofail::channel();
     tx1.send(PeerManagerEvent::GetBindAddresses(rtx)).unwrap();
     let bind_addresses = timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap();
     assert_eq!(bind_addresses.len(), 1);
@@ -728,7 +729,7 @@ where
     loop {
         tokio::time::sleep(Duration::from_millis(10)).await;
         time_getter.advance_time(Duration::from_secs(1)).await;
-        let (rtx, rrx) = oneshot::channel();
+        let (rtx, rrx) = oneshot_nofail::channel();
         tx1.send(PeerManagerEvent::GetConnectedPeers(rtx)).unwrap();
         let connected_peers = timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap();
         if connected_peers.len() == 1 {
@@ -791,7 +792,7 @@ where
     .await;
 
     // Get the first peer manager's bind address
-    let (rtx, rrx) = oneshot::channel();
+    let (rtx, rrx) = oneshot_nofail::channel();
     tx1.send(PeerManagerEvent::GetBindAddresses(rtx)).unwrap();
 
     let bind_addresses = timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap();
diff --git a/p2p/src/peer_manager/tests/mod.rs b/p2p/src/peer_manager/tests/mod.rs
index 66a966ed9..676c87aa1 100644
--- a/p2p/src/peer_manager/tests/mod.rs
+++ b/p2p/src/peer_manager/tests/mod.rs
@@ -20,10 +20,7 @@ mod ping;
 use std::{sync::Arc, time::Duration};
 
 use common::time_getter::TimeGetter;
-use tokio::{
-    sync::{mpsc::UnboundedSender, oneshot},
-    time::timeout,
-};
+use tokio::{sync::mpsc::UnboundedSender, time::timeout};
 
 use crate::{
     event::PeerManagerEvent,
@@ -31,6 +28,7 @@ use crate::{
     net::{ConnectivityService, NetworkingService},
     peer_manager::PeerManager,
     testing_utils::peerdb_inmemory_store,
+    utils::oneshot_nofail,
     P2pConfig,
 };
 
@@ -120,7 +118,7 @@ where
 async fn get_connected_peers<T: NetworkingService + std::fmt::Debug>(
     tx: &UnboundedSender<PeerManagerEvent<T>>,
 ) -> Vec<ConnectedPeer> {
-    let (rtx, rrx) = oneshot::channel();
+    let (rtx, rrx) = oneshot_nofail::channel();
     tx.send(PeerManagerEvent::GetConnectedPeers(rtx)).unwrap();
     timeout(Duration::from_secs(1), rrx).await.unwrap().unwrap()
 }