feat(gossipsub): switch internal async-channel,

protocols/gossipsub/CHANGELOG.md

@@ -1,4 +1,7 @@
 ## 0.49.0
+- switch the internal `async-channel` used to dispatch messages from `NetworkBehaviour` to the `ConnectionHandler`
+  with an internal priority queue. See [PR XXXX](https://github.com/libp2p/rust-libp2p/pull/XXXX)
+
 - Send IDONTWANT before Publishing a new message.
   See [PR 6017](https://github.com/libp2p/rust-libp2p/pull/6017)
 

protocols/gossipsub/src/behaviour.rs

@@ -59,7 +59,7 @@ use crate::{
     metrics::{Churn, Config as MetricsConfig, Inclusion, Metrics, Penalty},
     peer_score::{PeerScore, PeerScoreParams, PeerScoreThresholds, RejectReason},
     protocol::SIGNING_PREFIX,
-    rpc::Sender,
+    queue::Queue,
     rpc_proto::proto,
     subscription_filter::{AllowAllSubscriptionFilter, TopicSubscriptionFilter},
     time_cache::DuplicateCache,
@@ -765,6 +765,7 @@ where
             if self.send_message(
                 *peer_id,
                 RpcOut::Publish {
+                    message_id: msg_id.clone(),
                     message: raw_message.clone(),
                     timeout: Delay::new(self.config.publish_queue_duration()),
                 },
@@ -1360,6 +1361,7 @@ where
                     self.send_message(
                         *peer_id,
                         RpcOut::Forward {
+                            message_id: id.clone(),
                             message: msg,
                             timeout: Delay::new(self.config.forward_queue_duration()),
                         },
@@ -2083,9 +2085,8 @@ where
         // before we add all the gossip from this heartbeat in order to gain a true measure of
         // steady-state size of the queues.
         if let Some(m) = &mut self.metrics {
-            for sender_queue in self.connected_peers.values().map(|v| &v.sender) {
-                m.observe_priority_queue_size(sender_queue.priority_queue_len());
-                m.observe_non_priority_queue_size(sender_queue.non_priority_queue_len());
+            for sender_queue in self.connected_peers.values().map(|v| &v.messages) {
+                m.observe_priority_queue_size(sender_queue.len());
             }
         }
 
@@ -2729,6 +2730,7 @@ where
                 self.send_message(
                     *peer_id,
                     RpcOut::Forward {
+                        message_id: msg_id.clone(),
                         message: message.clone(),
                         timeout: Delay::new(self.config.forward_queue_duration()),
                     },
@@ -2857,33 +2859,20 @@ where
         }
 
         // Try sending the message to the connection handler.
-        match peer.sender.send_message(rpc) {
+        if rpc.high_priority() {
+            peer.messages.push(rpc);
+            return true;
+        }
+
+        match peer.messages.try_push(rpc) {
             Ok(()) => true,
             Err(rpc) => {
                 // Sending failed because the channel is full.
                 tracing::warn!(peer=%peer_id, "Send Queue full. Could not send {:?}.", rpc);
 
                 // Update failed message counter.
                 let failed_messages = self.failed_messages.entry(peer_id).or_default();
-                match rpc {
-                    RpcOut::Publish { .. } => {
-                        failed_messages.priority += 1;
-                        failed_messages.publish += 1;
-                    }
-                    RpcOut::Forward { .. } => {
-                        failed_messages.non_priority += 1;
-                        failed_messages.forward += 1;
-                    }
-                    RpcOut::IWant(_) | RpcOut::IHave(_) | RpcOut::IDontWant(_) => {
-                        failed_messages.non_priority += 1;
-                    }
-                    RpcOut::Graft(_)
-                    | RpcOut::Prune(_)
-                    | RpcOut::Subscribe(_)
-                    | RpcOut::Unsubscribe(_) => {
-                        unreachable!("Channel for highpriority control messages is unbounded and should always be open.")
-                    }
-                }
+                failed_messages.queue_full += 1;
 
                 // Update peer score.
                 if let Some((peer_score, ..)) = &mut self.peer_score {
@@ -3105,18 +3094,20 @@ where
         // occur.
         let connected_peer = self.connected_peers.entry(peer_id).or_insert(PeerDetails {
             kind: PeerKind::Floodsub,
-            connections: vec![],
             outbound: false,
-            sender: Sender::new(self.config.connection_handler_queue_len()),
+            connections: vec![],
             topics: Default::default(),
             dont_send: LinkedHashMap::new(),
+            messages: Queue::new(self.config.connection_handler_queue_len()),
         });
         // Add the new connection
         connected_peer.connections.push(connection_id);
 
+        // This clones a reference to the Queue so any new handlers reference the same underlying queue.
+        // No data is actually cloned here.
         Ok(Handler::new(
             self.config.protocol_config(),
-            connected_peer.sender.new_receiver(),
+            connected_peer.messages.clone(),
         ))
     }
 
@@ -3130,20 +3121,22 @@ where
     ) -> Result<THandler<Self>, ConnectionDenied> {
         let connected_peer = self.connected_peers.entry(peer_id).or_insert(PeerDetails {
             kind: PeerKind::Floodsub,
-            connections: vec![],
             // Diverging from the go implementation we only want to consider a peer as outbound peer
             // if its first connection is outbound.
             outbound: !self.px_peers.contains(&peer_id),
-            sender: Sender::new(self.config.connection_handler_queue_len()),
+            connections: vec![],
             topics: Default::default(),
             dont_send: LinkedHashMap::new(),
+            messages: Queue::new(self.config.connection_handler_queue_len()),
         });
         // Add the new connection
         connected_peer.connections.push(connection_id);
 
+        // This clones a reference to the Queue so any new handlers reference the same underlying queue.
+        // No data is actually cloned here.
         Ok(Handler::new(
             self.config.protocol_config(),
-            connected_peer.sender.new_receiver(),
+            connected_peer.messages.clone(),
         ))
     }
 
@@ -3184,7 +3177,7 @@ where
                     }
                 }
             }
-            HandlerEvent::MessageDropped(rpc) => {
+            HandlerEvent::MessagesDropped(rpc) => {
                 // Account for this in the scoring logic
                 if let Some((peer_score, _, _)) = &mut self.peer_score {
                     peer_score.failed_message_slow_peer(&propagation_source);
@@ -3193,28 +3186,9 @@ where
                 // Keep track of expired messages for the application layer.
                 let failed_messages = self.failed_messages.entry(propagation_source).or_default();
                 failed_messages.timeout += 1;
-                match rpc {
-                    RpcOut::Publish { .. } => {
-                        failed_messages.publish += 1;
-                    }
-                    RpcOut::Forward { .. } => {
-                        failed_messages.forward += 1;
-                    }
-                    _ => {}
-                }
-
-                // Record metrics on the failure.
                 if let Some(metrics) = self.metrics.as_mut() {
-                    match rpc {
-                        RpcOut::Publish { message, .. } => {
-                            metrics.publish_msg_dropped(&message.topic);
-                            metrics.timeout_msg_dropped(&message.topic);
-                        }
-                        RpcOut::Forward { message, .. } => {
-                            metrics.forward_msg_dropped(&message.topic);
-                            metrics.timeout_msg_dropped(&message.topic);
-                        }
-                        _ => {}
+                    if let RpcOut::Publish { message, .. } | RpcOut::Forward { message, .. } = rpc {
+                        metrics.timeout_msg_dropped(&message.topic);
                     }
                 }
             }
@@ -3317,6 +3291,16 @@ where
                             if let Some(metrics) = self.metrics.as_mut() {
                                 metrics.register_idontwant(message_ids.len());
                             }
+
+                            // Remove messages from the queue.
+                            peer.messages.retain(|rpc| match rpc {
+                                RpcOut::Publish { message_id, .. }
+                                | RpcOut::Forward { message_id, .. } => {
+                                    !message_ids.contains(message_id)
+                                }
+                                _ => true,
+                            });
+
                             for message_id in message_ids {
                                 peer.dont_send.insert(message_id, Instant::now());
                                 // Don't exceed capacity.