fix: wait for gossipsub heartbeat before sending message (#94)

Only wait for the sending peer to see the remote in their topic subscriber
list but also wait a little bit longer for gossipsub to rebalance the
mesh to increase the chances of the publisher actually sending a pubsub
message to the subscriber.

package.json

@@ -137,6 +137,7 @@
     "@libp2p/interface-peer-id": "^2.0.1",
     "@libp2p/interface-peer-info": "^1.0.7",
     "@multiformats/multiaddr": "^11.4.0",
+    "delay": "^5.0.0",
     "it-all": "^2.0.0",
     "it-first": "^2.0.0",
     "it-handshake": "^4.1.2",

src/pubsub/floodsub.ts

@@ -4,7 +4,7 @@ import { expect } from 'aegir/chai'
 import type { Daemon, DaemonFactory, NodeType, SpawnOptions } from '../index.js'
 import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import first from 'it-first'
-import { waitForBothSubscribed } from './utils.js'
+import { waitForSubscribed } from './utils.js'
 
 export function floodsubTests (factory: DaemonFactory): void {
   const nodeTypes: NodeType[] = ['js', 'go']
@@ -61,7 +61,7 @@ function runFloodsubTests (factory: DaemonFactory, optionsA: SpawnOptions, optio
       }
 
       const publisher = async (): Promise<void> => {
-        await waitForBothSubscribed(topic, peerA, peerB)
+        await waitForSubscribed(topic, peerA, peerB)
         await peerA.client.pubsub.publish(topic, data)
       }
 

src/pubsub/gossipsub.ts

@@ -4,7 +4,7 @@ import { expect } from 'aegir/chai'
 import type { Daemon, DaemonFactory, NodeType, SpawnOptions } from '../index.js'
 import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import first from 'it-first'
-import { waitForBothSubscribed } from './utils.js'
+import { waitForSubscribed } from './utils.js'
 
 export function gossipsubTests (factory: DaemonFactory): void {
   const nodeTypes: NodeType[] = ['js', 'go']
@@ -61,7 +61,7 @@ function runGossipsubTests (factory: DaemonFactory, optionsA: SpawnOptions, opti
       }
 
       const publisher = async (): Promise<void> => {
-        await waitForBothSubscribed(topic, peerA, peerB)
+        await waitForSubscribed(topic, peerA, peerB)
         await peerA.client.pubsub.publish(topic, data)
       }
 

src/pubsub/hybrid.ts

@@ -4,7 +4,7 @@ import { expect } from 'aegir/chai'
 import type { Daemon, DaemonFactory, NodeType, SpawnOptions } from '../index.js'
 import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import first from 'it-first'
-import { waitForBothSubscribed } from './utils.js'
+import { waitForSubscribed } from './utils.js'
 
 export function hybridTests (factory: DaemonFactory): void {
   const nodeTypes: NodeType[] = ['js', 'go']
@@ -61,7 +61,7 @@ function runHybridTests (factory: DaemonFactory, optionsA: SpawnOptions, options
       }
 
       const publisher = async (): Promise<void> => {
-        await waitForBothSubscribed(topic, peerA, peerB)
+        await waitForSubscribed(topic, peerA, peerB)
         await peerA.client.pubsub.publish(topic, data)
       }
 

src/pubsub/utils.ts

@@ -1,26 +1,22 @@
 import pWaitFor from 'p-wait-for'
-import type { Daemon } from '..'
+import delay from 'delay'
+import type { Daemon } from '../index.js'
 
-export async function waitForBothSubscribed (topic: string, a: Daemon, b: Daemon): Promise<void> {
-  await a.client.pubsub.subscribe(topic)
-  await b.client.pubsub.subscribe(topic)
-
-  const idA = await a.client.identify()
+/**
+ * Wait for daemon a to see daemon b in it's subscriber list
+ * for the passed topic
+ */
+export async function waitForSubscribed (topic: string, a: Daemon, b: Daemon): Promise<void> {
   const idB = await b.client.identify()
 
   // wait for subscription stream
-  await Promise.all([
-    await pWaitFor(async () => {
-      const peers = await a.client.pubsub.getSubscribers(topic)
-      return peers.map(p => p.toString()).includes(idB.peerId.toString())
-    }, {
-      interval: 500
-    }),
-    await pWaitFor(async () => {
-      const peers = await b.client.pubsub.getSubscribers(topic)
-      return peers.map(p => p.toString()).includes(idA.peerId.toString())
-    }, {
-      interval: 500
-    })
-  ])
+  await pWaitFor(async () => {
+    const peers = await a.client.pubsub.getSubscribers(topic)
+    return peers.map(p => p.toString()).includes(idB.peerId.toString())
+  }, {
+    interval: 500
+  })
+
+  // wait for the gossipsub heartbeat to rebalance the mesh
+  await delay(2000)
 }