fix: add timeout to protocol notifications + log improvements

- protocol notifications now have a set "safety" timeout.
- add log for inbound comms pipeline concurrency usage

comms/dht/src/store_forward/store.rs

@@ -226,7 +226,7 @@ where S: Service<DecryptedDhtMessage, Response = (), Error = PipelineError> + Se
         }
 
         if message.dht_header.message_type.is_saf_message() {
-            log_not_eligible("it is a SAF message");
+            log_not_eligible("it is a SAF protocol message");
             return Ok(None);
         }
 

comms/src/bounded_executor.rs

@@ -40,13 +40,15 @@ pub struct TrySpawnError;
 pub struct BoundedExecutor {
     inner: runtime::Handle,
     semaphore: Arc<Semaphore>,
+    max_available: usize,
 }
 
 impl BoundedExecutor {
     pub fn new(executor: runtime::Handle, num_permits: usize) -> Self {
         Self {
             inner: executor,
             semaphore: Arc::new(Semaphore::new(num_permits)),
+            max_available: num_permits,
         }
     }
 
@@ -70,12 +72,18 @@ impl BoundedExecutor {
         self.num_available() > 0
     }
 
-    /// Returns the number tasks that can be spawned on this executor without blocking.
+    /// Returns the remaining number of tasks that can be spawned on this executor without waiting.
     #[inline]
     pub fn num_available(&self) -> usize {
         self.semaphore.available_permits()
     }
 
+    /// Returns the maximum number of concurrent tasks that can be spawned on this executor without waiting.
+    #[inline]
+    pub fn max_available(&self) -> usize {
+        self.max_available
+    }
+
     pub fn try_spawn<F>(&self, future: F) -> Result<JoinHandle<F::Output>, TrySpawnError>
     where
         F: Future + Send + 'static,

comms/src/connection_manager/manager.rs

@@ -392,15 +392,25 @@ where
                     node_id.short_str(),
                     proto_str
                 );
-                if let Err(err) = self
+                let notify_fut = self
                     .protocols
-                    .notify(&protocol, ProtocolEvent::NewInboundSubstream(*node_id, stream))
-                    .await
-                {
-                    error!(
-                        target: LOG_TARGET,
-                        "Error sending NewSubstream notification for protocol '{}' because '{:?}'", proto_str, err
-                    );
+                    .notify(&protocol, ProtocolEvent::NewInboundSubstream(*node_id, stream));
+                match time::timeout(Duration::from_secs(10), notify_fut).await {
+                    Ok(Err(err)) => {
+                        error!(
+                            target: LOG_TARGET,
+                            "Error sending NewSubstream notification for protocol '{}' because '{:?}'", proto_str, err
+                        );
+                    },
+                    Err(err) => {
+                        error!(
+                            target: LOG_TARGET,
+                            "Error sending NewSubstream notification for protocol '{}' because {}", proto_str, err
+                        );
+                    },
+                    _ => {
+                        debug!(target: LOG_TARGET, "Protocol notification for '{}' sent", proto_str);
+                    },
                 }
             },
 

comms/src/pipeline/inbound.rs

@@ -71,6 +71,18 @@ where
                 return;
             }
             let service = self.service.clone();
+
+            let num_available = self.executor.num_available();
+            let max_available = self.executor.max_available();
+            // Only emit this message if there is any concurrent usage
+            if num_available < max_available {
+                debug!(
+                    target: LOG_TARGET,
+                    "Inbound pipeline usage: {}/{}",
+                    max_available - num_available,
+                    max_available
+                );
+            }
             // Call the service in it's own spawned task
             self.executor
                 .spawn(async move {
@@ -80,6 +92,7 @@ where
                 })
                 .await;
         }
+        info!(target: LOG_TARGET, "Inbound pipeline terminated: the stream completed");
     }
 }
 

comms/src/protocol/messaging/extension.rs

@@ -26,7 +26,7 @@ use crate::{
     message::InboundMessage,
     pipeline,
     protocol::{
-        messaging::{consts, protocol::MESSAGING_PROTOCOL, MessagingEventSender},
+        messaging::{protocol::MESSAGING_PROTOCOL, MessagingEventSender},
         ProtocolExtension,
         ProtocolExtensionContext,
         ProtocolExtensionError,
@@ -38,6 +38,18 @@ use futures::channel::mpsc;
 use std::fmt;
 use tower::Service;
 
+/// Buffer size for inbound messages from _all_ peers. This should be large enough to buffer quite a few incoming
+/// messages before creating backpressure on peers speaking the messaging protocol.
+pub const INBOUND_MESSAGE_BUFFER_SIZE: usize = 100;
+/// Buffer size notifications that a peer wants to speak /tari/messaging. This buffer is used for all peers, but a low
+/// value is ok because this events happen once (or less) per connecting peer. For e.g. a value of 10 would allow 10
+/// peers to concurrently request to speak /tari/messaging.
+pub const MESSAGING_PROTOCOL_EVENTS_BUFFER_SIZE: usize = 30;
+
+/// Buffer size for requests to the messaging protocol. All outbound messages will be sent along this channel. Some
+/// buffering may be required if the node needs to send many messages out at the same time.
+pub const MESSAGING_REQUEST_BUFFER_SIZE: usize = 50;
+
 pub struct MessagingProtocolExtension<TInPipe, TOutPipe, TOutReq> {
     event_tx: MessagingEventSender,
     pipeline: pipeline::Config<TInPipe, TOutPipe, TOutReq>,
@@ -60,11 +72,11 @@ where
     TOutReq: Send + 'static,
 {
     fn install(self: Box<Self>, context: &mut ProtocolExtensionContext) -> Result<(), ProtocolExtensionError> {
-        let (proto_tx, proto_rx) = mpsc::channel(consts::MESSAGING_PROTOCOL_EVENTS_BUFFER_SIZE);
+        let (proto_tx, proto_rx) = mpsc::channel(MESSAGING_PROTOCOL_EVENTS_BUFFER_SIZE);
         context.add_protocol(&[MESSAGING_PROTOCOL.clone()], proto_tx);
 
-        let (messaging_request_tx, messaging_request_rx) = mpsc::channel(consts::MESSAGING_REQUEST_BUFFER_SIZE);
-        let (inbound_message_tx, inbound_message_rx) = mpsc::channel(consts::INBOUND_MESSAGE_BUFFER_SIZE);
+        let (messaging_request_tx, messaging_request_rx) = mpsc::channel(MESSAGING_REQUEST_BUFFER_SIZE);
+        let (inbound_message_tx, inbound_message_rx) = mpsc::channel(INBOUND_MESSAGE_BUFFER_SIZE);
 
         let messaging = MessagingProtocol::new(
             Default::default(),

comms/src/protocol/messaging/mod.rs

@@ -26,8 +26,6 @@ pub use config::MessagingConfig;
 mod extension;
 pub use extension::MessagingProtocolExtension;
 
-mod consts;
-
 mod error;
 mod inbound;
 mod outbound;

comms/src/protocol/rpc/server/router.rs

@@ -208,7 +208,7 @@ where
     <B::Service as Service<Request<Bytes>>>::Future: Send + 'static,
 {
     fn install(self: Box<Self>, context: &mut ProtocolExtensionContext) -> Result<(), ProtocolExtensionError> {
-        let (proto_notif_tx, proto_notif_rx) = mpsc::channel(10);
+        let (proto_notif_tx, proto_notif_rx) = mpsc::channel(20);
         context.add_protocol(&self.protocol_names, proto_notif_tx);
         let rpc_context = RpcCommsBackend::new(context.peer_manager(), context.connectivity());
         task::spawn(self.serve(proto_notif_rx, rpc_context));

integration_tests/features/WalletBaseNodeSwitch.feature

@@ -1,6 +1,5 @@
 Feature: Wallet Base Node Switch
 
-    @doit
     Scenario: As a user I want to change base node for a wallet
         Given I have a base node Node1 connected to all seed nodes
         And I have a base node Node2 connected to all seed nodes