test(gossipsub): flaky tests (#1451)

tests/pubsub/testgossipsubgossip.nim

@@ -306,15 +306,13 @@ suite "GossipSub Gossip Protocol":
 
     # When node 0 sends a large message
     let largeMsg = newSeq[byte](1000)
-    check (await nodes[0].publish(topic, largeMsg)) == 1
-    await waitForHeartbeat()
+    tryPublish await nodes[0].publish(topic, largeMsg), 1
 
     # Only node 2 should have received the iDontWant message
-    let receivedIDontWants = messages[].mapIt(it[].len)
-    check:
-      receivedIDontWants[0] == 0
-      receivedIDontWants[1] == 0
-      receivedIDontWants[2] == 1
+    checkUntilTimeout:
+      messages[].mapIt(it[].len)[2] == 1
+      messages[].mapIt(it[].len)[1] == 0
+      messages[].mapIt(it[].len)[0] == 0
 
   asyncTest "e2e - GossipSub peer exchange":
     # A, B & C are subscribed to something

tests/pubsub/testgossipsubheartbeat.nim

@@ -169,14 +169,17 @@ suite "GossipSub Heartbeat":
       node.subscribe(topic, voidTopicHandler)
     await waitForHeartbeat(heartbeatInterval)
 
-    # When Node0 sends a message to the topic
     let node0 = nodes[0]
+    checkUntilTimeout:
+      node0.gossipsub.hasKey(topic)
+
+    # When Node0 sends a message to the topic
     tryPublish await node0.publish(topic, newSeq[byte](10000)), 3
 
     # Then Node0 fanout peers are populated
-    let maxFanoutPeers = node0.parameters.d
     checkUntilTimeout:
-      node0.fanout.hasKey(topic) and node0.fanout[topic].len == maxFanoutPeers
+      node0.fanout.hasKey(topic)
+      node0.fanout[topic].len > 0
 
     # And after heartbeat Node0 fanout peers are dropped (because fanoutTTL < heartbeatInterval)
     checkUntilTimeout:
@@ -254,7 +257,7 @@ suite "GossipSub Heartbeat":
       tryPublish await nodes[0].publish(topic, newSeq[byte](1000)), 1
 
     # Then Node1 receives 5 iDontWant messages from Node0
-    checkUntilTimeout:
+    checkUntilTimeoutCustom(3.seconds, 50.milliseconds):
       peer.iDontWants[0].len == msgCount
 
     for i in 0 ..< historyLength:

tests/pubsub/testgossipsubmeshmanagement.nim

@@ -9,8 +9,9 @@
 
 {.used.}
 
-import utils
 import chronicles
+import std/[sequtils]
+import utils
 import ../../libp2p/protocols/pubsub/[gossipsub, mcache, peertable]
 import ../helpers
 
@@ -401,28 +402,24 @@ suite "GossipSub Mesh Management":
     await waitForHeartbeat()
 
     # Then all nodes should be subscribed to the topics initially
-    for i in 0 ..< numberOfNodes:
-      let node = nodes[i]
-      for j in 0 ..< topics.len:
-        let topic = topics[j]
-        checkUntilTimeout:
-          node.topics.contains(topic)
-          node.gossipsub[topic].len() == numberOfNodes - 1
-          node.mesh[topic].len() == numberOfNodes - 1
+    for i in 0 ..< topics.len:
+      let topic = topics[i]
+      checkUntilTimeout:
+        nodes.allIt(it.topics.contains(topic))
+        nodes.allIt(it.gossipsub.getOrDefault(topic).len() == numberOfNodes - 1)
+        nodes.allIt(it.mesh.getOrDefault(topic).len() == numberOfNodes - 1)
 
     # When they unsubscribe from all topics
     for topic in topics:
       unsubscribeAllNodes(nodes, topic, voidTopicHandler)
 
     # Then topics should be removed from mesh and gossipsub
-    for i in 0 ..< numberOfNodes:
-      let node = nodes[i]
-      for j in 0 ..< topics.len:
-        let topic = topics[j]
-        checkUntilTimeout:
-          topic notin node.topics
-          topic notin node.mesh
-          topic notin node.gossipsub
+    for i in 0 ..< topics.len:
+      let topic = topics[i]
+      checkUntilTimeout:
+        nodes.allIt(not it.topics.contains(topic))
+        nodes.allIt(topic notin it.gossipsub)
+        nodes.allIt(topic notin it.mesh)
 
   asyncTest "Unsubscribe backoff":
     const