Add tests for publish extra nodes

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

@@ -125,6 +125,7 @@ class GossipV1_1Tests : GossipTestsBase() {
                 super.initChannelWithHandler(streamHandler, handler)
             }
         }
+
         val test = TwoRoutersTest(mockRouterFactory = { exec, _, _ -> MalformedMockRouter(exec) })
         val mockRouter = test.router2.router as MalformedMockRouter
 
@@ -1126,34 +1127,99 @@ class GossipV1_1Tests : GossipTestsBase() {
         // 2 heartbeats - the topic should be GRAFTed
         test.fuzz.timeController.addTime(2.seconds)
 
-        fun createPruneMessage(peersCount: Int): Rpc.RPC {
-            val peerInfos = List(peersCount) {
-                Rpc.PeerInfo.newBuilder()
-                    .setPeerID(PeerId.random().bytes.toProtobuf())
-                    .setSignedPeerRecord(ByteString.EMPTY)
-                    .build()
-            }
-            return Rpc.RPC.newBuilder().setControl(
-                Rpc.ControlMessage.newBuilder().addPrune(
-                    Rpc.ControlPrune.newBuilder()
-                        .setTopicID(topic)
-                        .addAllPeers(peerInfos)
-                )
-            ).build()
-        }
-
         test.mockRouter.sendToSingle(
-            createPruneMessage(test.gossipRouter.params.maxPeersAcceptedInPruneMsg + 1)
+            createPruneMessage(topic, test.gossipRouter.params.maxPeersAcceptedInPruneMsg + 1)
         )
 
         // prune message should be dropped because too many peers
         assertEquals(1, test.gossipRouter.mesh[topic]!!.size)
 
         test.mockRouter.sendToSingle(
-            createPruneMessage(test.gossipRouter.params.maxPeersAcceptedInPruneMsg)
+            createPruneMessage(topic, test.gossipRouter.params.maxPeersAcceptedInPruneMsg)
         )
 
         // prune message should now be processed
         assertEquals(0, test.gossipRouter.mesh[topic]!!.size)
     }
+
+    @Test
+    fun `when a peer leaves the mesh it should still be considered for publishing`() {
+        val test = TwoRoutersTest()
+        val topic = "topic1"
+
+        test.mockRouter.subscribe(topic)
+        test.gossipRouter.subscribe(topic)
+
+        // 2 heartbeats - the topic should be GRAFTed
+        test.fuzz.timeController.addTime(2.seconds)
+
+        assertTrue((test.gossipRouter.mesh[topic]?.size ?: 0) == 1)
+
+        // remote peer leaves the mesh
+        test.mockRouter.sendToSingle(createPruneMessage(topic))
+        test.fuzz.timeController.addTime(1.seconds)
+
+        assertTrue((test.gossipRouter.mesh[topic]?.size ?: 0) == 0)
+
+        val message1 = newMessage(topic, 0L, "Hello-0".toByteArray())
+        test.gossipRouter.publish(message1)
+
+        test.mockRouter.waitForMessage { it.publishCount > 0 }
+    }
+
+    @Test
+    fun `publishing should collect at least D peers if mesh is smaller`() {
+        val params = GossipParams()
+
+        val test = ManyRoutersTest(params = params, mockRouterCount = params.D)
+        val topic = "topic1"
+        test.connectAll()
+
+        test.mockRouters.forEach { it.subscribe(topic) }
+        test.gossipRouter.subscribe(topic)
+
+        // 2 heartbeats - the topic should be GRAFTed
+        test.fuzz.timeController.addTime(2.seconds)
+
+        val topicMeshRouters = test.gossipRouter.mesh[topic]!!
+        assertTrue((topicMeshRouters.size) >= params.DLow)
+
+        // leave just 2 peers in the mesh
+        topicMeshRouters.drop(2)
+            .forEach {
+                test.getMockRouter(it.peerId).sendToSingle(createPruneMessage(topic))
+            }
+        test.fuzz.timeController.addTime(1.seconds)
+
+        assertTrue((test.gossipRouter.mesh[topic]?.size ?: 0) == 2)
+
+        val message1 = newMessage(topic, 0L, "Hello-0".toByteArray())
+        test.gossipRouter.publish(message1)
+
+        val routerReceivedMessageCount =
+            test.mockRouters.count { mockRouter ->
+                mockRouter.inboundMessages.any { msg ->
+                    msg.publishCount > 0
+                }
+            }
+
+        assertTrue(routerReceivedMessageCount >= params.D)
+    }
+
+    private fun createPruneMessage(topic: String, pxPeersCount: Int = 0): Rpc.RPC {
+        val peerInfos = List(pxPeersCount) {
+            Rpc.PeerInfo.newBuilder()
+                .setPeerID(PeerId.random().bytes.toProtobuf())
+                .setSignedPeerRecord(ByteString.EMPTY)
+                .build()
+        }
+        return Rpc.RPC.newBuilder().setControl(
+            Rpc.ControlMessage.newBuilder().addPrune(
+                Rpc.ControlPrune.newBuilder()
+                    .setTopicID(topic)
+                    .setBackoff(10)
+                    .addAllPeers(peerInfos)
+            )
+        ).build()
+    }
 }