feat(gossipsub): send IDONTWANT before publishing a new message

This allows us to possible save some bandwidth on the network.
See libp2p/go-libp2p-pubsub#610 (comment) for more info

Pull-Request: libp2p#6017.

Author: jxs

protocols/gossipsub/CHANGELOG.md

@@ -1,4 +1,7 @@
 ## 0.49.0
+- Send IDONTWANT before Publishing a new message.
+  See [PR 6017](https://github.com/libp2p/rust-libp2p/pull/6017)
+
 - Improve log messaging by renaming `message` to `message_id`. This change makes it log coherent to
   the field and also improve duplication of `message` field.
   See [PR 5972](https://github.com/libp2p/rust-libp2p/pull/5972)

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,
@@ -747,6 +749,19 @@ where
         let mut publish_failed = true;
         for peer_id in recipient_peers.iter() {
             tracing::trace!(peer=%peer_id, "Sending message to peer");
+            // If enabled, Send first an IDONTWANT so that if we are slower than forwarders
+            // publishing the original message we don't receive it back.
+            if raw_message.raw_protobuf_len() > self.config.idontwant_message_size_threshold()
+                && self.config.idontwant_on_publish()
+            {
+                self.send_message(
+                    *peer_id,
+                    RpcOut::IDontWant(IDontWant {
+                        message_ids: vec![msg_id.clone()],
+                    }),
+                );
+            }
+
             if self.send_message(
                 *peer_id,
                 RpcOut::Publish {
@@ -766,13 +781,6 @@ where
             return Err(PublishError::AllQueuesFull(recipient_peers.len()));
         }
 
-        // Broadcast IDONTWANT messages
-        if raw_message.raw_protobuf_len() > self.config.idontwant_message_size_threshold()
-            && self.config.idontwant_on_publish()
-        {
-            self.send_idontwant(&raw_message, &msg_id, raw_message.source.as_ref());
-        }
-
         tracing::debug!(message_id=%msg_id, "Published message");
 
         if let Some(metrics) = self.metrics.as_mut() {
@@ -1766,7 +1774,26 @@ where
 
         // Broadcast IDONTWANT messages
         if raw_message.raw_protobuf_len() > self.config.idontwant_message_size_threshold() {
-            self.send_idontwant(&raw_message, &msg_id, Some(propagation_source));
+            let recipient_peers = self
+                .mesh
+                .get(&message.topic)
+                .map(|mesh| mesh.iter())
+                .unwrap_or_default()
+                .copied()
+                .chain(self.gossip_promises.peers_for_message(&msg_id))
+                .filter(|peer_id| {
+                    peer_id != propagation_source && Some(peer_id) != message.source.as_ref()
+                })
+                .collect::<Vec<PeerId>>();
+
+            for peer_id in recipient_peers {
+                self.send_message(
+                    peer_id,
+                    RpcOut::IDontWant(IDontWant {
+                        message_ids: vec![msg_id.clone()],
+                    }),
+                );
+            }
         }
 
         // Check the validity of the message
@@ -2633,49 +2660,6 @@ where
         }
     }
 
-    /// Helper function which sends an IDONTWANT message to mesh\[topic\] peers.
-    fn send_idontwant(
-        &mut self,
-        message: &RawMessage,
-        msg_id: &MessageId,
-        propagation_source: Option<&PeerId>,
-    ) {
-        let Some(mesh_peers) = self.mesh.get(&message.topic) else {
-            return;
-        };
-
-        let iwant_peers = self.gossip_promises.peers_for_message(msg_id);
-
-        let recipient_peers: Vec<PeerId> = mesh_peers
-            .iter()
-            .chain(iwant_peers.iter())
-            .filter(|&peer_id| {
-                Some(peer_id) != propagation_source && Some(peer_id) != message.source.as_ref()
-            })
-            .cloned()
-            .collect();
-
-        for peer_id in recipient_peers {
-            let Some(peer) = self.connected_peers.get_mut(&peer_id) else {
-                tracing::error!(peer = %peer_id,
-                    "Could not IDONTWANT, peer doesn't exist in connected peer list");
-                continue;
-            };
-
-            // Only gossipsub 1.2 peers support IDONTWANT.
-            if peer.kind != PeerKind::Gossipsubv1_2 {
-                continue;
-            }
-
-            self.send_message(
-                peer_id,
-                RpcOut::IDontWant(IDontWant {
-                    message_ids: vec![msg_id.clone()],
-                }),
-            );
-        }
-    }
-
     /// Helper function which forwards a message to mesh\[topic\] peers.
     ///
     /// Returns true if at least one peer was messaged.
@@ -2867,6 +2851,11 @@ where
             return false;
         };
 
+        if !matches!(peer.kind, PeerKind::Gossipsubv1_2) && matches!(rpc, RpcOut::IDontWant(..)) {
+            tracing::trace!(peer=%peer_id, "Won't send IDONTWANT message for message to peer as it doesn't support Gossipsub v1.2");
+            return false;
+        }
+
         // Try sending the message to the connection handler.
         match peer.sender.send_message(rpc) {
             Ok(()) => true,