refactor(networking): peermanager refactor and cleanups

apps/chat2/chat2.nim

@@ -24,6 +24,7 @@ import libp2p/[switch,                   # manage transports, a single entry poi
                nameresolving/dnsresolver]# define DNS resolution
 import
   ../../waku/v2/protocol/waku_message,
+  ../../waku/v2/protocol/waku_lightpush,
   ../../waku/v2/protocol/waku_lightpush/rpc,
   ../../waku/v2/protocol/waku_filter,
   ../../waku/v2/protocol/waku_store,
@@ -489,7 +490,7 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
       echo "Connecting to storenode: " & $(storenode.get())
 
       node.mountStoreClient()
-      node.setStorePeer(storenode.get())
+      node.peerManager.addServicePeer(storenode.get(), WakuStoreCodec)
 
       proc storeHandler(response: HistoryResponse) {.gcsafe.} =
         for msg in response.messages:
@@ -509,13 +510,13 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
     await mountLightPush(node)
 
     node.mountLightPushClient()
-    node.setLightPushPeer(conf.lightpushnode)
+    node.peerManager.addServicePeer(parseRemotePeerInfo(conf.lightpushnode), WakuLightpushCodec)
 
   if conf.filternode != "":
     await node.mountFilter()
     await node.mountFilterClient()
 
-    node.setFilterPeer(parseRemotePeerInfo(conf.filternode))
+    node.peerManager.addServicePeer(parseRemotePeerInfo(conf.filternode), WakuFilterCodec)
 
     proc filterHandler(pubsubTopic: PubsubTopic, msg: WakuMessage) {.gcsafe.} =
       trace "Hit filter handler", contentTopic=msg.contentTopic

apps/chat2bridge/chat2bridge.nim

@@ -16,6 +16,10 @@ import
   libp2p/errors,
   ../../../waku/v2/protocol/waku_message,
   ../../../waku/v2/node/waku_node,
+  ../../../waku/v2/utils/peers,
+  ../../../waku/v2/node/peer_manager,
+  ../../waku/v2/protocol/waku_filter,
+  ../../waku/v2/protocol/waku_store,
   # Chat 2 imports
   ../chat2/chat2,
   # Common cli config
@@ -281,10 +285,12 @@ when isMainModule:
     waitFor connectToNodes(bridge.nodev2, conf.staticnodes)
 
   if conf.storenode != "":
-    setStorePeer(bridge.nodev2, conf.storenode)
+    let storePeer = parseRemotePeerInfo(conf.storenode)
+    bridge.nodev2.peerManager.addServicePeer(storePeer, WakuStoreCodec)
 
   if conf.filternode != "":
-    setFilterPeer(bridge.nodev2, conf.filternode)
+    let filterPeer = parseRemotePeerInfo(conf.filternode)
+    bridge.nodev2.peerManager.addServicePeer(filterPeer, WakuFilterCodec)
 
   if conf.rpc:
     let ta = initTAddress(conf.rpcAddress,

apps/wakubridge/wakubridge.nim

@@ -26,6 +26,8 @@ import
   libp2p/nameresolving/nameresolver,
   ../../waku/v2/utils/time,
   ../../waku/v2/protocol/waku_message,
+  ../../waku/v2/protocol/waku_store,
+  ../../waku/v2/protocol/waku_filter,
   ../../waku/v2/node/message_cache,
   ../../waku/v2/node/waku_node,
   ../../waku/v2/node/peer_manager,
@@ -428,11 +430,13 @@ when isMainModule:
 
   if conf.storenode != "":
     mountStoreClient(bridge.nodev2)
-    setStorePeer(bridge.nodev2, conf.storenode)
+    let storeNode = parseRemotePeerInfo(conf.storenode)
+    bridge.nodev2.peerManager.addServicePeer(storeNode, WakuStoreCodec)
 
   if conf.filternode != "":
     waitFor mountFilterClient(bridge.nodev2)
-    setFilterPeer(bridge.nodev2, conf.filternode)
+    let filterNode = parseRemotePeerInfo(conf.filternode)
+    bridge.nodev2.peerManager.addServicePeer(filterNode, WakuFilterCodec)
 
   if conf.rpc:
     let ta = initTAddress(conf.rpcAddress,

tests/v2/test_peer_manager.nim

@@ -96,10 +96,9 @@ procSuite "Peer Manager":
     await node.mountSwap()
     node.mountStoreClient()
 
-    node.wakuSwap.setPeer(swapPeer.toRemotePeerInfo())
-
-    node.setStorePeer(storePeer.toRemotePeerInfo())
-    node.setFilterPeer(filterPeer.toRemotePeerInfo())
+    node.peerManager.addServicePeer(swapPeer.toRemotePeerInfo(), WakuSwapCodec)
+    node.peerManager.addServicePeer(storePeer.toRemotePeerInfo(), WakuStoreCodec)
+    node.peerManager.addServicePeer(filterPeer.toRemotePeerInfo(), WakuFilterCodec)
 
     # Check peers were successfully added to peer manager
     check:
@@ -127,7 +126,7 @@ procSuite "Peer Manager":
     await allFutures(nodes.mapIt(it.mountRelay()))
 
     # Test default connectedness for new peers
-    nodes[0].peerManager.addPeer(nodes[1].peerInfo.toRemotePeerInfo(), WakuRelayCodec)
+    nodes[0].peerManager.addPeer(nodes[1].peerInfo.toRemotePeerInfo())
     check:
       # No information about node2's connectedness
       nodes[0].peerManager.peerStore.connectedness(nodes[1].peerInfo.peerId) == NotConnected
@@ -160,7 +159,7 @@ procSuite "Peer Manager":
 
     await nodes[0].start()
     await nodes[0].mountRelay()
-    nodes[0].peerManager.addPeer(nodes[1].peerInfo.toRemotePeerInfo(), WakuRelayCodec)
+    nodes[0].peerManager.addPeer(nodes[1].peerInfo.toRemotePeerInfo())
 
     # Set a low backoff to speed up test: 2, 4, 8, 16
     nodes[0].peerManager.initialBackoffInSec = 2
@@ -236,7 +235,8 @@ procSuite "Peer Manager":
       node3.peerManager.peerStore.peers().anyIt(it.peerId == peerInfo2.peerId)
       node3.peerManager.peerStore.connectedness(peerInfo2.peerId) == NotConnected
 
-    await node3.mountRelay()  # This should trigger a reconnect
+    await node3.mountRelay()
+    await node3.peerManager.connectToRelayPeers()
 
     check:
       # Reconnected to node2 after "restart"
@@ -313,12 +313,12 @@ procSuite "Peer Manager":
     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)
+    nodes[0].peerManager.addPeer(peerInfos[1])
+    nodes[0].peerManager.addPeer(peerInfos[2])
+    nodes[0].peerManager.addPeer(peerInfos[3])
 
-    # Attempt to connect to all known peers supporting a given protocol
-    await nodes[0].peerManager.reconnectPeers(WakuRelayCodec, protocolMatcher(WakuRelayCodec))
+    # Connect to relay peers
+    await nodes[0].peerManager.connectToRelayPeers()
 
     check:
       # Peerstore track all three peers
@@ -512,7 +512,7 @@ procSuite "Peer Manager":
 
     # Create 15 peers and add them to the peerstore
     let peers = toSeq(1..15).mapIt(parseRemotePeerInfo("/ip4/0.0.0.0/tcp/0/p2p/" & $PeerId.random().get()))
-    for p in peers: pm.addPeer(p, "")
+    for p in peers: pm.addPeer(p)
 
     # Check that we have 15 peers in the peerstore
     check:

tests/v2/wakunode_jsonrpc/test_jsonrpc_admin.nim

@@ -167,8 +167,12 @@ procSuite "Waku v2 JSON-RPC API - Admin":
       filterPeer = PeerInfo.new(generateEcdsaKey(), @[locationAddr])
       storePeer = PeerInfo.new(generateEcdsaKey(), @[locationAddr])
 
-    node.setStorePeer(storePeer.toRemotePeerInfo())
-    node.setFilterPeer(filterPeer.toRemotePeerInfo())
+    node.peerManager.addServicePeer(filterPeer.toRemotePeerInfo(), WakuFilterCodec)
+    node.peerManager.addServicePeer(storePeer.toRemotePeerInfo(), WakuStoreCodec)
+
+    # Mock that we connected in the past so Identify populated this
+    node.peerManager.peerStore[ProtoBook][filterPeer.peerId] = @[WakuFilterCodec]
+    node.peerManager.peerStore[ProtoBook][storePeer.peerId] = @[WakuStoreCodec]
 
     let response = await client.get_waku_v2_admin_v1_peers()
 

tests/v2/wakunode_jsonrpc/test_jsonrpc_filter.nim

@@ -14,6 +14,7 @@ import
   ../../../waku/v2/node/jsonrpc/filter/handlers as filter_api,
   ../../../waku/v2/node/jsonrpc/filter/client as filter_api_client,
   ../../../waku/v2/protocol/waku_message,
+  ../../../waku/v2/protocol/waku_filter,
   ../../../waku/v2/protocol/waku_filter/rpc,
   ../../../waku/v2/protocol/waku_filter/client,
   ../../../waku/v2/utils/peers,
@@ -40,7 +41,7 @@ procSuite "Waku v2 JSON-RPC API - Filter":
     await node1.mountFilter()
     await node2.mountFilterClient()
 
-    node2.setFilterPeer(node1.peerInfo.toRemotePeerInfo())
+    node2.peerManager.addServicePeer(node1.peerInfo.toRemotePeerInfo(), WakuFilterCodec)
 
     # RPC server setup
     let

tests/v2/wakunode_jsonrpc/test_jsonrpc_store.nim

@@ -16,6 +16,7 @@ import
   ../../../waku/v2/protocol/waku_message,
   ../../../waku/v2/protocol/waku_archive,
   ../../../waku/v2/protocol/waku_archive/driver/queue_driver,
+  ../../../waku/v2/protocol/waku_store,
   ../../../waku/v2/protocol/waku_store/rpc,
   ../../../waku/v2/utils/peers,
   ../../../waku/v2/utils/time,
@@ -66,7 +67,7 @@ procSuite "Waku v2 JSON-RPC API - Store":
     var listenSwitch = newStandardSwitch(some(key))
     await listenSwitch.start()
 
-    node.setStorePeer(listenSwitch.peerInfo.toRemotePeerInfo())
+    node.peerManager.addServicePeer(listenSwitch.peerInfo.toRemotePeerInfo(), WakuStoreCodec)
 
     listenSwitch.mount(node.wakuRelay)
     listenSwitch.mount(node.wakuStore)