feat(gossipsub): switch internal async-channel,

protocols/gossipsub/CHANGELOG.md

@@ -1,4 +1,7 @@
-## 0.49.0
+## 0.49
+
+- 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)
 
 - Fix a race condition for messages published which are already in the network.
   See [PR 5928](https://github.com/libp2p/rust-libp2p/pull/5928)

protocols/gossipsub/Cargo.toml

@@ -24,14 +24,16 @@ fnv = "1.0.7"
 futures = { workspace = true }
 futures-timer = "3.0.2"
 getrandom = { workspace = true }
-hashlink = { workspace = true}
+hashlink = { workspace = true }
 hex_fmt = "0.3.0"
 web-time = { workspace = true }
-libp2p-core = { workspace = true }
-libp2p-identity = { workspace = true, features = ["rand"] }
-libp2p-swarm = { workspace = true }
+# Libp2p crates, updated to use crates.io versions so that we can use this gossipsub fork with
+# crates.io libp2p
+libp2p-core = "0.43"
+libp2p-identity = { version = "0.2", features = ["rand"] }
+libp2p-swarm = "0.46"
 quick-protobuf = "0.8"
-quick-protobuf-codec = { workspace = true }
+quick-protobuf-codec = "0.3.1"
 rand = "0.8"
 regex = "1.10.5"
 serde = { version = "1", optional = true, features = ["derive"] }
@@ -42,8 +44,7 @@ tracing = { workspace = true }
 prometheus-client = { workspace = true }
 
 [dev-dependencies]
-libp2p-core = { workspace = true }
-libp2p-swarm-test = { path = "../../swarm-test" }
+libp2p-swarm-test = "0.5.0"
 quickcheck = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 tokio = { workspace = true, features = ["rt", "rt-multi-thread", "time", "macros"] }

protocols/gossipsub/src/behaviour.rs

@@ -19,10 +19,12 @@
 // DEALINGS IN THE SOFTWARE.
 
 use std::{
-    cmp::{max, Ordering, Ordering::Equal},
+    cmp::{
+        max,
+        Ordering::{self, Equal},
+    },
     collections::{BTreeSet, HashMap, HashSet, VecDeque},
-    fmt,
-    fmt::Debug,
+    fmt::{self, Debug},
     net::IpAddr,
     task::{Context, Poll},
     time::Duration,
@@ -57,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,
@@ -750,6 +752,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()),
                 },
@@ -1359,6 +1362,7 @@ where
                     self.send_message(
                         *peer_id,
                         RpcOut::Forward {
+                            message_id: id.clone(),
                             message: msg,
                             timeout: Delay::new(self.config.forward_queue_duration()),
                         },
@@ -2057,9 +2061,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());
             }
         }
 
@@ -2724,6 +2727,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()),
                     },
@@ -2846,34 +2850,20 @@ where
             return false;
         };
 
-        // 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 {
@@ -3109,16 +3099,18 @@ where
             .or_insert(PeerConnections {
                 kind: PeerKind::Floodsub,
                 connections: vec![],
-                sender: Sender::new(self.config.connection_handler_queue_len()),
                 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(),
         ))
     }
 
@@ -3136,16 +3128,18 @@ where
             .or_insert(PeerConnections {
                 kind: PeerKind::Floodsub,
                 connections: vec![],
-                sender: Sender::new(self.config.connection_handler_queue_len()),
                 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(),
         ))
     }
 
@@ -3186,37 +3180,22 @@ where
                     }
                 }
             }
-            HandlerEvent::MessageDropped(rpc) => {
+            HandlerEvent::MessagesDropped(rpcs) => {
                 // Account for this in the scoring logic
                 if let Some((peer_score, _, _)) = &mut self.peer_score {
                     peer_score.failed_message_slow_peer(&propagation_source);
                 }
 
                 // 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.
+                failed_messages.timeout += rpcs.len();
                 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);
+                    for rpc in rpcs {
+                        if let RpcOut::Publish { message, .. } | RpcOut::Forward { message, .. } =
+                            rpc
+                        {
                             metrics.timeout_msg_dropped(&message.topic);
                         }
-                        _ => {}
                     }
                 }
             }
@@ -3319,6 +3298,16 @@ where
                             if let Some(metrics) = self.metrics.as_mut() {
                                 metrics.register_idontwant(message_ids.len());
                             }
+
+                            // Remove messages from the queue.
+                            peer.messages.retain_mut(|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.