test: enable gossipsub tests (#3424)

Author: dozyio

packages/gossipsub/package.json

@@ -47,14 +47,16 @@
     "release": "aegir release --no-types",
     "build": "aegir build",
     "generate": "protons ./src/message/rpc.proto",
-    "pretest": "npm run build",
-    "pretest:e2e": "npm run build",
-    "benchmark": "yarn benchmark:files 'test/benchmark/**/*.test.ts'",
+    "benchmark": "yarn benchmark:files test/benchmark/**/*.test.ts",
     "benchmark:files": "NODE_OPTIONS='--max-old-space-size=4096 --loader=ts-node/esm' benchmark --config .benchrc.yaml --defaultBranch master",
-    "test": "aegir test -f './dist/test/*.spec.js'",
-    "test:unit": "aegir test -f './dist/test/unit/*.test.js' --target node",
-    "test:e2e": "aegir test -f './dist/test/e2e/*.spec.js'",
-    "test:browser": "npm run test -- --target browser"
+    "test": "aegir test -f dist/test/*.spec.js",
+    "test:node": "aegir test -t node -f dist/test/*.spec.js -f dist/test/unit/*.spec.js -f dist/test/e2e/*.spec.js",
+    "test:chrome": "aegir test -f dist/test/*.spec.js -t browser",
+    "test:chrome-webworker": "aegir test -f dist/test/*.spec.js -t webworker",
+    "test:firefox": "aegir test -f dist/test/*.spec.js -t browser -- --browser firefox",
+    "test:firefox-webworker": "aegir test -f dist/test/*.spec.js -t webworker -- --browser firefox",
+    "test:electron-main": "aegir test -f dist/test/*.spec.js -t electron-main",
+    "test:webkit": "aegir test -f dist/test/*.spec.js -t browser -- --browser webkit"
   },
   "repository": {
     "type": "git",

packages/gossipsub/test/2-nodes.spec.ts

@@ -27,7 +27,15 @@ async function nodesArePubSubPeers (node0: GossipSubAndComponents, node1: Gossip
         .getPeers()
         .map((p) => p.toString())
         .includes(node0.components.peerId.toString())
-      return node0SeesNode1 && node1SeesNode0
+
+      const node0StreamsReady =
+        node0.pubsub.streamsOutbound.has(node1.components.peerId.toString()) &&
+        node0.pubsub.streamsInbound.has(node1.components.peerId.toString())
+      const node1StreamsReady =
+        node1.pubsub.streamsOutbound.has(node0.components.peerId.toString()) &&
+        node1.pubsub.streamsInbound.has(node0.components.peerId.toString())
+
+      return node0SeesNode1 && node1SeesNode0 && node0StreamsReady && node1StreamsReady
     },
     {
       timeout
@@ -93,14 +101,16 @@ describe('2 nodes', () => {
     it('Subscribe to a topic', async () => {
       const topic = 'test_topic'
 
+      const subscriptionChanges = [
+        pEvent<'subscription-change', CustomEvent<SubscriptionChangeData>>(nodes[0].pubsub, 'subscription-change'),
+        pEvent<'subscription-change', CustomEvent<SubscriptionChangeData>>(nodes[1].pubsub, 'subscription-change')
+      ]
+
       nodes[0].pubsub.subscribe(topic)
       nodes[1].pubsub.subscribe(topic)
 
       // await subscription change
-      const [evt0] = await Promise.all([
-        pEvent<'subscription-change', CustomEvent<SubscriptionChangeData>>(nodes[0].pubsub, 'subscription-change'),
-        pEvent<'subscription-change', CustomEvent<SubscriptionChangeData>>(nodes[1].pubsub, 'subscription-change')
-      ])
+      const [evt0] = await Promise.all(subscriptionChanges)
 
       const { peerId: changedPeerId, subscriptions: changedSubs } = evt0.detail
 
@@ -118,14 +128,11 @@ describe('2 nodes', () => {
       expect(changedSubs[0].topic).to.equal(topic)
       expect(changedSubs[0].subscribe).to.equal(true)
 
-      // await heartbeats
-      await Promise.all([
-        pEvent(nodes[0].pubsub, 'gossipsub:heartbeat'),
-        pEvent(nodes[1].pubsub, 'gossipsub:heartbeat')
-      ])
-
-      expect((nodes[0].pubsub).mesh.get(topic)?.has(nodes[1].components.peerId.toString())).to.be.true()
-      expect((nodes[1].pubsub).mesh.get(topic)?.has(nodes[0].components.peerId.toString())).to.be.true()
+      // await mesh propagation
+      await pWaitFor(() => {
+        return (nodes[0].pubsub).mesh.get(topic)?.has(nodes[1].components.peerId.toString()) === true &&
+          (nodes[1].pubsub).mesh.get(topic)?.has(nodes[0].components.peerId.toString()) === true
+      }, { timeout: 10000 })
     })
   })
 
@@ -141,15 +148,22 @@ describe('2 nodes', () => {
       })
 
       // Create subscriptions
+      const subscriptionChanges = [
+        pEvent(nodes[0].pubsub, 'subscription-change'),
+        pEvent(nodes[1].pubsub, 'subscription-change')
+      ]
+      const heartbeats = [
+        pEvent(nodes[0].pubsub, 'gossipsub:heartbeat'),
+        pEvent(nodes[1].pubsub, 'gossipsub:heartbeat')
+      ]
+
       nodes[0].pubsub.subscribe(topic)
       nodes[1].pubsub.subscribe(topic)
 
       // await subscription change and heartbeat
       await Promise.all([
-        pEvent(nodes[0].pubsub, 'subscription-change'),
-        pEvent(nodes[1].pubsub, 'subscription-change'),
-        pEvent(nodes[0].pubsub, 'gossipsub:heartbeat'),
-        pEvent(nodes[1].pubsub, 'gossipsub:heartbeat')
+        ...subscriptionChanges,
+        ...heartbeats
       ])
     })
 
@@ -240,35 +254,40 @@ describe('2 nodes', () => {
       await connectAllPubSubNodes(nodes)
 
       // Create subscriptions
-      nodes[0].pubsub.subscribe(topic)
-      nodes[1].pubsub.subscribe(topic)
-
-      // await subscription change and heartbeat
-      await Promise.all([
+      const subscriptionChanges = [
         pEvent(nodes[0].pubsub, 'subscription-change'),
         pEvent(nodes[1].pubsub, 'subscription-change')
-      ])
-      await Promise.all([
+      ]
+      const heartbeats = [
         pEvent(nodes[0].pubsub, 'gossipsub:heartbeat'),
         pEvent(nodes[1].pubsub, 'gossipsub:heartbeat')
-      ])
+      ]
+
+      nodes[0].pubsub.subscribe(topic)
+      nodes[1].pubsub.subscribe(topic)
+
+      // await subscription change and heartbeat
+      await Promise.all([...subscriptionChanges, ...heartbeats])
     })
 
     afterEach(async () => {
       await stop(...nodes.reduce<any[]>((acc, curr) => acc.concat(curr.pubsub, ...Object.entries(curr.components)), []))
     })
 
     it('Unsubscribe from a topic', async () => {
-      nodes[0].pubsub.unsubscribe(topic)
-      expect(nodes[0].pubsub.getTopics()).to.be.empty()
-
-      const evt = await pEvent<'subscription-change', CustomEvent<SubscriptionChangeData>>(
+      const subscriptionChangePromise = pEvent<'subscription-change', CustomEvent<SubscriptionChangeData>>(
         nodes[1].pubsub,
         'subscription-change'
       )
+      const heartbeatPromise = pEvent(nodes[1].pubsub, 'gossipsub:heartbeat')
+
+      nodes[0].pubsub.unsubscribe(topic)
+      expect(nodes[0].pubsub.getTopics()).to.be.empty()
+
+      const evt = await subscriptionChangePromise
       const { peerId: changedPeerId, subscriptions: changedSubs } = evt.detail
 
-      await pEvent(nodes[1].pubsub, 'gossipsub:heartbeat')
+      await heartbeatPromise
 
       expect(nodes[1].pubsub.getPeers()).to.have.lengthOf(1)
       expect(nodes[1].pubsub.getSubscribers(topic)).to.be.empty()

packages/gossipsub/test/gossip.spec.ts

@@ -86,13 +86,10 @@ describe('gossip', () => {
   })
 
   it('should send idontwant to peers in topic', async function () {
-    // This test checks that idontwants and idontwantsCounts are correctly incrmemented
-    // - idontwantCounts should track the number of idontwant messages received from a peer for a single heartbeat
-    //   - it should increment on receive of idontwant msgs (up to limit)
-    //   - it should be emptied after heartbeat
-    // - idontwants should track the idontwant messages received from a peer along with the heartbeatId when received
-    //   - it should increment on receive of idontwant msgs (up to limit)
-    //   - it should be emptied after mcacheLength heartbeats
+    // This integration test checks IDONTWANT lifecycle behavior under network traffic:
+    // - publishing messages in a connected topic causes peers to track IDONTWANT state
+    // - retained idontwants stay bounded while entries are tracked across heartbeats
+    // - idontwantCounts are cleared at the next heartbeat
     this.timeout(10e4)
     const nodeA = nodes[0]
     const otherNodes = nodes.slice(1)
@@ -121,70 +118,99 @@ describe('gossip', () => {
       ])
       await nodeA.pubsub.publish(topic, msg)
     }
-    // track the heartbeat when each node received the last message
-
-    const ticks = otherNodes.map((n) => n.pubsub['heartbeatTicks'])
-
-    // there's no event currently implemented to await, so just wait a bit - flaky :(
-    // TODO figure out something more robust
-    await new Promise((resolve) => setTimeout(resolve, 200))
+    // wait for one heartbeat so IDONTWANT handling has happened on all peers
+    await Promise.all(otherNodes.map(async (n) => pEvent(n.pubsub, 'gossipsub:heartbeat')))
 
-    // other nodes should have received idontwant messages
-    // check that idontwants <= GossipsubIdontwantMaxMessages
+    // other nodes should have tracked idontwant messages
+    // check retained idontwants are bounded over mcacheLength heartbeats
     for (let i = 0; i < otherNodes.length; i++) {
       const node = otherNodes[i]
 
-      const currentTick = node.pubsub['heartbeatTicks']
-
-      const idontwantCounts = node.pubsub['idontwantCounts']
-      let minCount = Infinity
-      let maxCount = 0
-      for (const count of idontwantCounts.values()) {
-        minCount = Math.min(minCount, count)
-        maxCount = Math.max(maxCount, count)
-      }
-      // expect(minCount).to.be.greaterThan(0)
-      expect(maxCount).to.be.lessThanOrEqual(idontwantMaxMessages)
-
       const idontwants = node.pubsub['idontwants']
-      let minIdontwants = Infinity
       let maxIdontwants = 0
       for (const idontwant of idontwants.values()) {
-        minIdontwants = Math.min(minIdontwants, idontwant.size)
         maxIdontwants = Math.max(maxIdontwants, idontwant.size)
       }
-      // expect(minIdontwants).to.be.greaterThan(0)
-      expect(maxIdontwants).to.be.lessThanOrEqual(idontwantMaxMessages)
-
-      // sanity check that the idontwantCount matches idontwants.size
-      // only the case if there hasn't been a heartbeat
-      if (currentTick === ticks[i]) {
-        expect(minCount).to.be.equal(minIdontwants)
-        expect(maxCount).to.be.equal(maxIdontwants)
-      }
+
+      expect(maxIdontwants).to.be.lessThanOrEqual(idontwantMaxMessages * node.pubsub.opts.mcacheLength)
     }
 
     await Promise.all(otherNodes.map(async (n) => pEvent(n.pubsub, 'gossipsub:heartbeat')))
 
     // after a heartbeat
     // idontwants are still tracked
     // but idontwantCounts have been cleared
-    for (const node of nodes) {
+    for (const node of otherNodes) {
       const idontwantCounts = node.pubsub['idontwantCounts']
-      for (const count of idontwantCounts.values()) {
-        expect(count).to.be.equal(0)
-      }
+      expect(idontwantCounts.size).to.equal(0)
 
       const idontwants = node.pubsub['idontwants']
-      let minIdontwants = Infinity
       let maxIdontwants = 0
       for (const idontwant of idontwants.values()) {
-        minIdontwants = Math.min(minIdontwants, idontwant.size)
         maxIdontwants = Math.max(maxIdontwants, idontwant.size)
       }
-      // expect(minIdontwants).to.be.greaterThan(0)
-      expect(maxIdontwants).to.be.lessThanOrEqual(idontwantMaxMessages)
+      expect(maxIdontwants).to.be.lessThanOrEqual(idontwantMaxMessages * node.pubsub.opts.mcacheLength)
+    }
+  })
+
+  it('should cap idontwant tracking per peer per heartbeat', async function () {
+    // `should send idontwant to peers in topic` exercises this path indirectly, this
+    // test verifies the cap deterministically with controlled input and exact assertions.
+    // This test directly exercises handleIdontwant to verify per-heartbeat cap semantics:
+    // - idontwantCounts and idontwants stop growing at idontwantMaxMessages
+    // - counts reset on heartbeat and start again next heartbeat
+    const nodeA = nodes[0]
+    const pubsub = nodeA.pubsub as unknown as Partial<GossipSubClass> & {
+      handleIdontwant: GossipSubClass['handleIdontwant']
+      idontwantCounts: Map<string, number>
+      idontwants: Map<string, Map<string, number>>
     }
+    const peerId = 'peer-a'
+    const idontwantMaxMessages = nodeA.pubsub.opts.idontwantMaxMessages
+
+    pubsub.handleIdontwant(peerId, [{
+      messageIDs: Array.from({ length: idontwantMaxMessages * 2 }, (_, i) => uint8ArrayFromString(`msg-${i}`))
+    }])
+
+    expect(pubsub.idontwantCounts.get(peerId)).to.equal(idontwantMaxMessages)
+    expect(pubsub.idontwants.get(peerId)?.size).to.equal(idontwantMaxMessages)
+
+    pubsub.handleIdontwant(peerId, [{ messageIDs: [uint8ArrayFromString('overflow')] }])
+
+    expect(pubsub.idontwantCounts.get(peerId)).to.equal(idontwantMaxMessages)
+    expect(pubsub.idontwants.get(peerId)?.size).to.equal(idontwantMaxMessages)
+
+    await nodeA.pubsub.heartbeat()
+
+    expect(pubsub.idontwantCounts.get(peerId)).to.equal(undefined)
+
+    pubsub.handleIdontwant(peerId, [{ messageIDs: [uint8ArrayFromString('next-heartbeat')] }])
+
+    expect(pubsub.idontwantCounts.get(peerId)).to.equal(1)
+  })
+
+  it('should prune tracked idontwants after mcacheLength heartbeats', async function () {
+    const nodeA = nodes[0]
+    const pubsub = nodeA.pubsub as unknown as Partial<GossipSubClass> & {
+      handleIdontwant: GossipSubClass['handleIdontwant']
+      idontwants: Map<string, Map<string, number>>
+    }
+    const peerId = 'peer-b'
+    const mcacheLength = nodeA.pubsub.opts.mcacheLength
+
+    pubsub.handleIdontwant(peerId, [{ messageIDs: [uint8ArrayFromString('msg-to-prune')] }])
+    expect(pubsub.idontwants.get(peerId)?.size).to.equal(1)
+
+    for (let i = 0; i < mcacheLength - 1; i++) {
+      await nodeA.pubsub.heartbeat()
+    }
+
+    if (mcacheLength > 1) {
+      expect(pubsub.idontwants.get(peerId)?.size).to.equal(1)
+    }
+
+    await nodeA.pubsub.heartbeat()
+    expect(pubsub.idontwants.get(peerId)?.size).to.equal(0)
   })
 
   it('Should allow publishing to zero peers if flag is passed', async function () {

packages/gossipsub/test/unit/set.test.ts packages/gossipsub/test/unit/set.spec.tspackages/gossipsub/test/unit/set.test.ts renamed to packages/gossipsub/test/unit/set.spec.ts

@@ -12,7 +12,7 @@ describe('Set util', function () {
       { id: 'remove even numbers - need 0', ineed: 0, fn: (item) => item % 2 === 0, result: new Set([]) },
       { id: 'remove even numbers - need 1', ineed: 1, fn: (item) => item % 2 === 0, result: new Set([2]) },
       { id: 'remove even numbers - need 2', ineed: 2, fn: (item) => item % 2 === 0, result: new Set([2, 4]) },
-      { id: 'remove even numbers - need 10', ineed: 2, fn: (item) => item % 2 === 0, result: new Set([2, 4]) }
+      { id: 'remove even numbers - need 10', ineed: 10, fn: (item) => item % 2 === 0, result: new Set([2, 4]) }
     ]
 
     for (const { id, ineed, fn, result } of testCases) {