[SEC] gossipsub - handleIHAVE/handleIWANT recommendations & notes (#535)

Fixes #400

libp2p/protocols/pubsub/gossipsub/behavior.nim

@@ -199,12 +199,14 @@ proc handleIHave*(g: GossipSub,
     # TODO review deduplicate algorithm
     # * https://github.com/nim-lang/Nim/blob/5f46474555ee93306cce55342e81130c1da79a42/lib/pure/collections/sequtils.nim#L184
     # * it's probably not efficient and might give preference to the first dupe
-    var deIhaves = ihaves.deduplicate()
+    let deIhaves = ihaves.deduplicate()
-    for ihave in deIhaves.mitems:
+    for ihave in deIhaves:
       trace "peer sent ihave",
         peer, topic = ihave.topicID, msgs = ihave.messageIDs
       if ihave.topicID in g.mesh:
-        for m in ihave.messageIDs:
+        # also avoid duplicates here!
+        let deIhavesMsgs = ihave.messageIDs.deduplicate()
+        for m in deIhavesMsgs:
           let msgId = m & g.randomBytes
           if msgId notin g.seen:
             if peer.iHaveBudget > 0:
@@ -225,9 +227,10 @@ proc handleIWant*(g: GossipSub,
   elif peer.iWantBudget <= 0:
     trace "iwant: ignoring out of budget peer", peer, score = peer.score
   else:
-    var deIwants = iwants.deduplicate()
+    let deIwants = iwants.deduplicate()
     for iwant in deIwants:
-      for mid in iwant.messageIDs:
+      let deIwantsMsgs = iwant.messageIDs.deduplicate()
+      for mid in deIwantsMsgs:
         trace "peer sent iwant", peer, messageID = mid
         let msg = g.mcache.get(mid)
         if msg.isSome:

tests/pubsub/testgossipinternal.nim

@@ -630,3 +630,78 @@ suite "GossipSub internal":
 
     await allFuturesThrowing(conns.mapIt(it.close()))
     await gossipSub.switch.stop()
+
+  asyncTest "handleIHave/Iwant tests":
+    let gossipSub = TestGossipSub.init(newStandardSwitch())
+
+    proc handler(peer: PubSubPeer, msg: RPCMsg) {.async.} =
+      check false
+
+    let topic = "foobar"
+    var conns = newSeq[Connection]()
+    gossipSub.gossipsub[topic] = initHashSet[PubSubPeer]()
+    gossipSub.mesh[topic] = initHashSet[PubSubPeer]()
+    for i in 0..<30:
+      let conn = newBufferStream(noop)
+      conns &= conn
+      let peerInfo = randomPeerInfo()
+      conn.peerInfo = peerInfo
+      let peer = gossipSub.getPubSubPeer(peerInfo.peerId)
+      gossipSub.onNewPeer(peer)
+      peer.handler = handler
+      gossipSub.grafted(peer, topic)
+      gossipSub.peers[peerInfo.peerId] = peer
+      gossipSub.mesh[topic].incl(peer)
+
+    block:
+      # should ignore no budget peer
+      let conn = newBufferStream(noop)
+      conns &= conn
+      let peerInfo = randomPeerInfo()
+      conn.peerInfo = peerInfo
+      let peer = gossipSub.getPubSubPeer(peerInfo.peerId)
+      let id = @[0'u8, 1, 2, 3]
+      let msg = ControlIHave(
+        topicID: topic,
+        messageIDs: @[id, id, id]
+      )
+      # gossipSub.initPeerStats(peer)
+      let iwants = gossipSub.handleIHave(peer, @[msg])
+      check: iwants.messageIDs.len == 0
+
+    block:
+      # given duplicate ihave should generate only one iwant
+      let conn = newBufferStream(noop)
+      conns &= conn
+      let peerInfo = randomPeerInfo()
+      conn.peerInfo = peerInfo
+      let peer = gossipSub.getPubSubPeer(peerInfo.peerId)
+      let id = @[0'u8, 1, 2, 3]
+      let msg = ControlIHave(
+        topicID: topic,
+        messageIDs: @[id, id, id]
+      )
+      gossipSub.initPeerStats(peer)
+      let iwants = gossipSub.handleIHave(peer, @[msg])
+      check: iwants.messageIDs.len == 1
+
+    block:
+      # given duplicate iwant should generate only one message
+      let conn = newBufferStream(noop)
+      conns &= conn
+      let peerInfo = randomPeerInfo()
+      conn.peerInfo = peerInfo
+      let peer = gossipSub.getPubSubPeer(peerInfo.peerId)
+      let id = @[0'u8, 1, 2, 3]
+      gossipSub.mcache.put(id, Message())
+      let msg = ControlIWant(
+        messageIDs: @[id, id, id]
+      )
+      gossipSub.initPeerStats(peer)
+      let genmsg = gossipSub.handleIWant(peer, @[msg])
+      check: genmsg.len == 1
+
+    check gossipSub.mcache.msgs.len == 1
+
+    await allFuturesThrowing(conns.mapIt(it.close()))
+    await gossipSub.switch.stop()