[SEC] gossipsub - rebalanceMesh may prune up to D_lo on oversubscription (#531)

Fixes #403

libp2p/protocols/pubsub/peertable.nim

@@ -41,6 +41,3 @@ func peers*(table: PeerTable, topic: string): int =
     table[topic].len
   else:
     0
-
-func getPeers*(table: Table[string, HashSet[string]], topic: string): HashSet[string] =
-  table.getOrDefault(topic, initHashSet[string]())

tests/pubsub/testgossipinternal.nim

@@ -509,3 +509,59 @@ suite "GossipSub internal":
 
     await conn.close()
     await gossipSub.switch.stop()
+
+  asyncTest "rebalanceMesh Degree Hi - audit scenario":
+    let gossipSub = TestGossipSub.init(newStandardSwitch())
+    gossipSub.parameters.dScore = 4
+    gossipSub.parameters.d = 6
+    gossipSub.parameters.dOut = 3
+    gossipSub.parameters.dHigh = 12
+    gossipSub.parameters.dLow = 4
+
+    let topic = "foobar"
+    gossipSub.mesh[topic] = initHashSet[PubSubPeer]()
+    gossipSub.topicParams[topic] = TopicParams.init()
+
+    var conns = newSeq[Connection]()
+    gossipSub.gossipsub[topic] = initHashSet[PubSubPeer]()
+    for i in 0..<6:
+      let conn = newBufferStream(noop)
+      conn.dir = Direction.In
+      conns &= conn
+      let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA, rng[]).get())
+      conn.peerInfo = peerInfo
+      let peer = gossipSub.getPubSubPeer(peerInfo.peerId)
+      peer.score = 40.0
+      peer.sendConn = conn
+      gossipSub.onNewPeer(peer)
+      gossipSub.grafted(peer, topic)
+      gossipSub.peers[peerInfo.peerId] = peer
+      gossipSub.mesh[topic].incl(peer)
+
+    for i in 0..<7:
+      let conn = newBufferStream(noop)
+      conn.dir = Direction.Out
+      conns &= conn
+      let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA, rng[]).get())
+      conn.peerInfo = peerInfo
+      let peer = gossipSub.getPubSubPeer(peerInfo.peerId)
+      peer.score = 10.0
+      peer.sendConn = conn
+      gossipSub.onNewPeer(peer)
+      gossipSub.grafted(peer, topic)
+      gossipSub.peers[peerInfo.peerId] = peer
+      gossipSub.mesh[topic].incl(peer)
+
+    check gossipSub.mesh[topic].len == 13
+    gossipSub.rebalanceMesh(topic)
+    # ensure we are above dlow
+    check gossipSub.mesh[topic].len > gossipSub.parameters.dLow
+    var outbound = 0
+    for peer in gossipSub.mesh[topic]:
+      if peer.sendConn.dir == Direction.Out:
+        inc outbound
+    # ensure we give priority and keep at least dOut outbound peers
+    check outbound >= gossipSub.parameters.dOut
+
+    await allFuturesThrowing(conns.mapIt(it.close()))
+    await gossipSub.switch.stop()