address age's review

Author: jxs

protocols/gossipsub/Cargo.toml

@@ -13,6 +13,7 @@ categories = ["network-programming", "asynchronous"]
 [features]
 wasm-bindgen = ["getrandom/js", "futures-timer/wasm-bindgen"]
 metrics = ["prometheus-client"]
+test-extension = []
 
 [dependencies]
 async-channel = "2.3.1"

protocols/gossipsub/src/behaviour.rs

@@ -1540,6 +1540,7 @@ where
 
         peer.extensions = Some(extensions);
 
+        #[cfg(feature = "test-extension")]
         if extensions.test_extension.unwrap_or(false)
             && matches!(peer.kind, PeerKind::Gossipsubv1_3)
         {
@@ -3164,19 +3165,11 @@ where
         });
         // Add the new connection
         connected_peer.connections.push(connection_id);
-        let receiver = connected_peer.sender.new_receiver();
 
-        if connected_peer.connections.len() <= 1 {
-            // If this is the first connection send extensions message.
-            self.send_message(
-                peer_id,
-                RpcOut::Extensions(Extensions {
-                    test_extension: Some(true),
-                }),
-            );
-        }
-
-        Ok(Handler::new(self.config.protocol_config(), receiver))
+        Ok(Handler::new(
+            self.config.protocol_config(),
+            connected_peer.sender.new_receiver(),
+        ))
     }
 
     fn handle_established_outbound_connection(
@@ -3200,19 +3193,11 @@ where
         });
         // Add the new connection
         connected_peer.connections.push(connection_id);
-        let receiver = connected_peer.sender.new_receiver();
-
-        if connected_peer.connections.len() <= 1 {
-            // If this is the first connection send extensions message.
-            self.send_message(
-                peer_id,
-                RpcOut::Extensions(Extensions {
-                    test_extension: Some(true),
-                }),
-            );
-        }
 
-        Ok(Handler::new(self.config.protocol_config(), receiver))
+        Ok(Handler::new(
+            self.config.protocol_config(),
+            connected_peer.sender.new_receiver(),
+        ))
     }
 
     fn on_connection_handler_event(

protocols/gossipsub/src/handler.rs

@@ -39,7 +39,7 @@ use crate::{
     protocol::{GossipsubCodec, ProtocolConfig},
     rpc::Receiver,
     rpc_proto::proto,
-    types::{PeerKind, RawMessage, RpcIn, RpcOut},
+    types::{Extensions, PeerKind, RawMessage, RpcIn, RpcOut},
     ValidationError,
 };
 
@@ -211,7 +211,23 @@ impl EnabledHandler {
             self.outbound_substream.is_none(),
             "Established an outbound substream with one already available"
         );
-        self.outbound_substream = Some(OutboundSubstreamState::WaitingOutput(substream));
+
+        // For gossipsub 1.3 peers, immediately send Extensions message.
+        if matches!(peer_kind, PeerKind::Gossipsubv1_3) {
+            let test_extension = if cfg!(feature = "test-extension") {
+                Some(true)
+            } else {
+                None
+            };
+
+            let extensions_rpc = RpcOut::Extensions(Extensions { test_extension });
+            self.outbound_substream = Some(OutboundSubstreamState::PendingSend(
+                substream,
+                extensions_rpc.into_protobuf(),
+            ));
+        } else {
+            self.outbound_substream = Some(OutboundSubstreamState::WaitingOutput(substream));
+        }
     }
 
     fn poll(

protocols/gossipsub/src/protocol.rs

@@ -45,7 +45,7 @@ pub(crate) const SIGNING_PREFIX: &[u8] = b"libp2p-pubsub:";
 
 pub(crate) const GOSSIPSUB_1_3_0_PROTOCOL: ProtocolId = ProtocolId {
     protocol: StreamProtocol::new("/meshsub/1.3.0"),
-    kind: PeerKind::Gossipsubv1_2,
+    kind: PeerKind::Gossipsubv1_3,
 };
 
 pub(crate) const GOSSIPSUB_1_2_0_PROTOCOL: ProtocolId = ProtocolId {