sharded peer manager

Fix possible out of bound & logic error

Filter peers per protocol & rename proc

Fix out of bound & refactor dialling

Fix catching raise VS timeout & tests fixes

Fix test to connect to all peer per proto

Fix test

Div target per shard count

Logging & stuff

Fixes

Log peer count

More logs

Remove protobook override & clean up

Fix relay peer management & logs

Mics Fixes

Fixes

Author: SionoiS

tests/test_peer_manager.nim

@@ -1,7 +1,7 @@
 {.used.}
 
 import
-  std/[options, sequtils, times],
+  std/[options, sequtils, times, sugar],
   stew/shims/net as stewNet,
   testutils/unittests,
   chronos,
@@ -21,6 +21,7 @@ import
   ../../waku/node/peer_manager/peer_manager,
   ../../waku/node/peer_manager/peer_store/waku_peer_storage,
   ../../waku/waku_node,
+  ../../waku/waku_core/topics,
   ../../waku/waku_relay,
   ../../waku/waku_store,
   ../../waku/waku_filter,
@@ -128,7 +129,6 @@ procSuite "Peer Manager":
 
     await node.stop()
 
-
   asyncTest "Peer manager keeps track of connections":
     # Create 2 nodes
     let nodes = toSeq(0..<2).mapIt(newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("0.0.0.0"), Port(0)))
@@ -225,18 +225,36 @@ procSuite "Peer Manager":
     let
       database = SqliteDatabase.new(":memory:")[]
       storage = WakuPeerStorage.new(database)[]
-      node1 = newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("0.0.0.0"), Port(0), peerStorage = storage)
-      node2 = newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("0.0.0.0"), Port(0))
-      peerInfo2 = node2.switch.peerInfo
+      node1 = newTestWakuNode(
+        generateSecp256k1Key(),
+        ValidIpAddress.init("127.0.0.1"),
+        Port(44048),
+        peerStorage = storage
+        )
+      node2 = newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("127.0.0.1"), Port(34023))
+      
+    node1.mountMetadata(0).expect("Mounted Waku Metadata")
+    node2.mountMetadata(0).expect("Mounted Waku Metadata")
 
     await node1.start()
     await node2.start()
 
     await node1.mountRelay()
     await node2.mountRelay()
+    
+    let peerInfo2 = node2.switch.peerInfo
+    var remotePeerInfo2 = peerInfo2.toRemotePeerInfo()
+    remotePeerInfo2.enr = some(node2.enr)
 
-    require:
-      (await node1.peerManager.connectRelay(peerInfo2.toRemotePeerInfo())) == true
+    let is12Connected = await node1.peerManager.connectRelay(remotePeerInfo2)
+    assert is12Connected == true, "Node 1 and 2 not connected"
+
+    # When node use 0.0.0.0 and port 0
+    # After connecting the peer store is updated with the wrong address
+    check:
+      node1.peerManager.peerStore[AddressBook][remotePeerInfo2.peerId] == remotePeerInfo2.addrs
+
+    # wait for the peer store update
     await sleepAsync(chronos.milliseconds(500))
 
     check:
@@ -246,24 +264,34 @@ procSuite "Peer Manager":
       node1.peerManager.peerStore.connectedness(peerInfo2.peerId) == Connected
 
     # Simulate restart by initialising a new node using the same storage
-    let
-      node3 = newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("0.0.0.0"), Port(0), peerStorage = storage)
+    let node3 = newTestWakuNode(
+      generateSecp256k1Key(),
+      ValidIpAddress.init("127.0.0.1"),
+      Port(56037),
+      peerStorage = storage
+      )
+
+    node3.mountMetadata(0).expect("Mounted Waku Metadata")
 
     await node3.start()
+
     check:
       # Node2 has been loaded after "restart", but we have not yet reconnected
       node3.peerManager.peerStore.peers().len == 1
       node3.peerManager.peerStore.peers().anyIt(it.peerId == peerInfo2.peerId)
       node3.peerManager.peerStore.connectedness(peerInfo2.peerId) == NotConnected
 
     await node3.mountRelay()
-    await node3.peerManager.connectToRelayPeers()
+
+    await node3.peerManager.manageRelayPeers()
+
+    await sleepAsync(chronos.milliseconds(500))
 
     check:
       # Reconnected to node2 after "restart"
       node3.peerManager.peerStore.peers().len == 1
       node3.peerManager.peerStore.peers().anyIt(it.peerId == peerInfo2.peerId)
-      node3.peerManager.peerStore.connectedness(peerInfo2.peerId) == Connected
+      node3.peerManager.peerStore.connectedness(peerInfo2.peerId) == Connected 
 
     await allFutures([node1.stop(), node2.stop(), node3.stop()])
 
@@ -297,9 +325,9 @@ procSuite "Peer Manager":
         topics = @["/waku/2/rs/4/0"],
       )
 
-    discard node1.mountMetadata(clusterId3)
-    discard node2.mountMetadata(clusterId4)
-    discard node3.mountMetadata(clusterId4)
+    node1.mountMetadata(clusterId3).expect("Mounted Waku Metadata")
+    node2.mountMetadata(clusterId4).expect("Mounted Waku Metadata")
+    node3.mountMetadata(clusterId4).expect("Mounted Waku Metadata")
 
     # Start nodes
     await allFutures([node1.start(), node2.start(), node3.start()])
@@ -318,7 +346,6 @@ procSuite "Peer Manager":
       conn2.isNone
       conn3.isSome
 
-
   # TODO: nwaku/issues/1377
   xasyncTest "Peer manager support multiple protocol IDs when reconnecting to peers":
     let
@@ -380,19 +407,24 @@ procSuite "Peer Manager":
     let nodes = toSeq(0..<4).mapIt(newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("0.0.0.0"), Port(0)))
 
     # Start them
-    await allFutures(nodes.mapIt(it.start()))
+    discard nodes.mapIt(it.mountMetadata(0))
     await allFutures(nodes.mapIt(it.mountRelay()))
+    await allFutures(nodes.mapIt(it.start()))
 
     # Get all peer infos
-    let peerInfos = nodes.mapIt(it.switch.peerInfo.toRemotePeerInfo())
+    let peerInfos = collect:
+      for i in 0..nodes.high:
+        let peerInfo = nodes[i].switch.peerInfo.toRemotePeerInfo()
+        peerInfo.enr = some(nodes[i].enr)
+        peerInfo
 
     # Add all peers (but self) to node 0
     nodes[0].peerManager.addPeer(peerInfos[1])
     nodes[0].peerManager.addPeer(peerInfos[2])
     nodes[0].peerManager.addPeer(peerInfos[3])
 
     # Connect to relay peers
-    await nodes[0].peerManager.connectToRelayPeers()
+    await nodes[0].peerManager.manageRelayPeers()
 
     check:
       # Peerstore track all three peers

tests/testlib/wakunode.nim

@@ -66,7 +66,9 @@ proc newTestWakuNode*(nodeKey: crypto.PrivateKey,
     if (extIp.isSome() or dns4DomainName.isSome()) and extPort.isNone(): some(Port(60000))
     else: extPort
 
-  let conf = defaultTestWakuNodeConf()
+  var conf = defaultTestWakuNodeConf()
+
+  conf.topics = topics
 
   if dns4DomainName.isSome() and extIp.isNone():
     # If there's an error resolving the IP, an exception is thrown and test fails