Get negotiated Gossip protocol version from either inbound or outbound stream

src/main/kotlin/io/libp2p/pubsub/gossip/GossipRouter.kt

@@ -38,10 +38,15 @@ fun P2PService.PeerHandler.getIP(): String? =
     streamHandler.stream.connection.remoteAddress().getStringComponent(Protocol.IP4)
 
 fun P2PService.PeerHandler.isOutbound() = streamHandler.stream.connection.isInitiator
-fun P2PService.PeerHandler.getOutboundProtocol() = getOutboundHandler()?.stream?.getProtocol()?.getNow(null)
-    ?: throw InternalErrorException("Outbound gossip stream not initialized or protocol is missing")
 
-fun P2PService.PeerHandler.getOutboundGossipProtocol() = PubsubProtocol.fromProtocol(getOutboundProtocol())
+fun P2PService.PeerHandler.getPeerProtocol(): PubsubProtocol {
+    fun P2PService.StreamHandler.getProtocol(): String? = stream.getProtocol().getNow(null)
+    val proto =
+        getOutboundHandler()?.getProtocol()
+            ?: getInboundHandler()?.getProtocol()
+            ?: throw InternalErrorException("Couldn't get peer gossip protocol")
+    return PubsubProtocol.fromProtocol(proto)
+}
 
 /**
  * Router implementing this protocol: https://github.com/libp2p/specs/tree/master/pubsub/gossipsub
@@ -111,7 +116,11 @@ open class GossipRouter @JvmOverloads constructor(
         score.notifyUnseenMessage(peer, msg)
     }
 
-    override fun notifySeenMessage(peer: PeerHandler, msg: PubsubMessage, validationResult: Optional<ValidationResult>) {
+    override fun notifySeenMessage(
+        peer: PeerHandler,
+        msg: PubsubMessage,
+        validationResult: Optional<ValidationResult>
+    ) {
         score.notifySeenMessage(peer, msg, validationResult)
         if (validationResult.isPresent && validationResult.get() != ValidationResult.Invalid) {
             notifyAnyValidMessage(peer, msg)
@@ -449,7 +458,7 @@ open class GossipRouter @JvmOverloads constructor(
 
     private fun enqueuePrune(peer: PeerHandler, topic: Topic) {
         val pruneBuilder = Rpc.ControlPrune.newBuilder().setTopicID(topic)
-        if (peer.getOutboundGossipProtocol() == PubsubProtocol.Gossip_V_1_1 && this.protocol == PubsubProtocol.Gossip_V_1_1) {
+        if (peer.getPeerProtocol() == PubsubProtocol.Gossip_V_1_1 && this.protocol == PubsubProtocol.Gossip_V_1_1) {
             // add v1.1 specific fields
             pruneBuilder.backoff = params.pruneBackoff.seconds
             (getTopicPeers(topic) - peer)

src/test/kotlin/io/libp2p/pubsub/gossip/GossipPubsubRouterTest.kt

@@ -176,4 +176,48 @@ class GossipPubsubRouterTest : PubsubRouterTest({
             Assertions.assertFalse(testLogAppender.hasAnyWarns())
         }
     }
+
+    @Test
+    fun `test that no warn when trying to respond prune without outbound gossip stream`() {
+        // when remote gossip makes connection and immediately send GRAFT
+        // the situation when we fail to send PRUNE (as not outbound stream yet)
+        // shouldn't be treated as internal error and no WARN logs should be printed
+        val fuzz = DeterministicFuzz()
+
+        val router1 = fuzz.createTestRouter(MockRouter())
+
+        // when isDirect the gossip router should reply with PRUNE to GRAFT
+        // this would reproduce the case
+        val gossipScoreParams = GossipScoreParams(GossipPeerScoreParams(isDirect = { true }))
+
+        val router2 = fuzz.createTestRouter(GossipRouter(scoreParams = gossipScoreParams))
+        val mockRouter = router1.router as MockRouter
+
+        router2.router.subscribe("topic1")
+        router1.connect(router2, LogLevel.INFO, LogLevel.INFO)
+
+        TestLogAppender().install().use { testLogAppender ->
+            val msg1 = Rpc.RPC.newBuilder()
+                .setControl(
+                    Rpc.ControlMessage.newBuilder().addGraft(
+                        Rpc.ControlGraft.newBuilder().setTopicID("topic1")
+                    )
+                ).build()
+
+            mockRouter.sendToSingle(msg1)
+
+            Assertions.assertFalse(testLogAppender.hasAnyWarns())
+        }
+
+        router2.connect(router1, LogLevel.INFO, LogLevel.INFO)
+
+        val msg1 = Rpc.RPC.newBuilder()
+            .setControl(
+                Rpc.ControlMessage.newBuilder().addIhave(
+                    Rpc.ControlIHave.newBuilder().addMessageIDs("messageId".toByteArray().toProtobuf())
+                )
+            ).build()
+
+        mockRouter.sendToSingle(msg1)
+    }
 }