refactor: modify reconnectPeers and unittest

Author: alrevuelta

tests/v2/test_peer_manager.nim

@@ -5,6 +5,7 @@ import
   stew/shims/net as stewNet,
   testutils/unittests,
   chronicles,
+  chronos,
   json_rpc/rpcserver,
   json_rpc/rpcclient,
   eth/keys,
@@ -269,3 +270,47 @@ procSuite "Peer Manager":
       node3.peerManager.peerStore.connectedness(peerInfo2.peerId) == Connected
 
     await allFutures([node1.stop(), node2.stop(), node3.stop()])
+
+  asyncTest "Peer manager connects to all peers supporting a given protocol":
+    # Create 4 nodes
+    var nodes: seq[WakuNode]
+    for i in 0..<4:
+      let nodeKey = crypto.PrivateKey.random(Secp256k1, rng[])[]
+      let node = WakuNode.new(nodeKey, ValidIpAddress.init("0.0.0.0"), Port(60860 + i))
+      nodes &= node
+
+    # Start them
+    await allFutures(nodes.mapIt(it.start()))
+    await allFutures(nodes.mapIt(it.mountRelay()))
+
+    # Get all peer infos
+    let peerInfos = nodes.mapIt(it.switch.peerInfo.toRemotePeerInfo())
+
+    # Add all peers (but self) to node 0
+    nodes[0].peerManager.addPeer(peerInfos[1], WakuRelayCodec)
+    nodes[0].peerManager.addPeer(peerInfos[2], WakuRelayCodec)
+    nodes[0].peerManager.addPeer(peerInfos[3], WakuRelayCodec)
+
+    # Attempt to connect to all known peers supporting a given protocol
+    await nodes[0].peerManager.reconnectPeers(WakuRelayCodec, protocolMatcher(WakuRelayCodec))
+
+    check:
+      # Peerstore track all three peers
+      nodes[0].peerManager.peerStore.peers().len == 3
+
+      # All peer ids are correct
+      nodes[0].peerManager.peerStore.peers().anyIt(it.peerId == nodes[1].switch.peerInfo.peerId)
+      nodes[0].peerManager.peerStore.peers().anyIt(it.peerId == nodes[2].switch.peerInfo.peerId)
+      nodes[0].peerManager.peerStore.peers().anyIt(it.peerId == nodes[3].switch.peerInfo.peerId)
+
+      # All peers support the relay protocol
+      nodes[0].peerManager.peerStore[ProtoBook][nodes[1].switch.peerInfo.peerId].contains(WakuRelayCodec)
+      nodes[0].peerManager.peerStore[ProtoBook][nodes[2].switch.peerInfo.peerId].contains(WakuRelayCodec)
+      nodes[0].peerManager.peerStore[ProtoBook][nodes[3].switch.peerInfo.peerId].contains(WakuRelayCodec)
+
+      # All peers are connected
+      nodes[0].peerManager.peerStore[ConnectionBook][nodes[1].switch.peerInfo.peerId] == Connected
+      nodes[0].peerManager.peerStore[ConnectionBook][nodes[2].switch.peerInfo.peerId] == Connected
+      nodes[0].peerManager.peerStore[ConnectionBook][nodes[3].switch.peerInfo.peerId] == Connected
+
+    await allFutures(nodes.mapIt(it.stop()))

waku/v2/node/peer_manager/peer_manager.nim

@@ -158,7 +158,6 @@ proc new*(T: type PeerManager, switch: Switch, storage: PeerStorage = nil): Peer
 # Manager interface #
 #####################
 
-# TODO: Move to peer store and unit test
 proc addPeer*(pm: PeerManager, remotePeerInfo: RemotePeerInfo, proto: string) =
   # Adds peer to manager for the specified protocol
 
@@ -195,31 +194,26 @@ proc reconnectPeers*(pm: PeerManager,
   debug "Reconnecting peers", proto=proto
 
   for storedInfo in pm.peerStore.peers(protocolMatcher):
-    # Check that peer exists and can be connected
-    if storedInfo.peerId notin pm.peerStore[ConnectionBook] or
-       pm.peerStore[ConnectionBook][storedInfo.peerId] == CannotConnect:
+    # Check that the peer can be connected
+    if storedInfo.connection == CannotConnect:
       debug "Not reconnecting to unreachable or non-existing peer", peerId=storedInfo.peerId
       continue
 
     # Respect optional backoff period where applicable.
     let
-      disconnectTime = Moment.init(pm.peerStore[DisconnectBook][storedInfo.peerId], Second)  # Convert
+      # TODO: Add method to peerStore (eg isBackoffExpired())
+      disconnectTime = Moment.init(storedInfo.disconnectTime, Second)  # Convert
       currentTime = Moment.init(getTime().toUnix, Second) # Current time comparable to persisted value
       backoffTime = disconnectTime + backoff - currentTime # Consider time elapsed since last disconnect
 
     trace "Respecting backoff", backoff=backoff, disconnectTime=disconnectTime, currentTime=currentTime, backoffTime=backoffTime
 
+    # TODO: This blocks the whole function. Try to connect to another peer in the meantime.
     if backoffTime > ZeroDuration:
       debug "Backing off before reconnect...", peerId=storedInfo.peerId, backoffTime=backoffTime
       # We disconnected recently and still need to wait for a backoff period before connecting
       await sleepAsync(backoffTime)
 
-    # Add to protos for peer, if it has not been added yet
-    if not pm.peerStore.get(storedInfo.peerId).protos.contains(proto):
-      let remotePeerInfo = storedInfo.toRemotePeerInfo()
-      trace "Adding newly dialed peer to manager", peerId = remotePeerInfo.peerId, addr = remotePeerInfo.addrs[0], proto = proto
-      pm.addPeer(remotePeerInfo, proto)
-
     trace "Reconnecting to peer", peerId=storedInfo.peerId
     discard await pm.dialPeer(storedInfo.peerId, toSeq(storedInfo.addrs), proto)