address review comments

Author: chaitanyaprem

apps/chat2mix/chat2mix.nim

@@ -590,10 +590,8 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
       error "Couldn't find any service peer"
       quit(QuitFailure)
 
-  #await mountLegacyLightPush(node)
   node.peerManager.addServicePeer(servicePeerInfo, WakuLightpushCodec)
   node.peerManager.addServicePeer(servicePeerInfo, WakuPeerExchangeCodec)
-  #node.peerManager.addServicePeer(servicePeerInfo, WakuRendezVousCodec)
 
   # Start maintaining subscription
   asyncSpawn maintainSubscription(

examples/lightpush_mix/lightpush_publisher_mix.nim

@@ -163,27 +163,27 @@ proc setupAndPublish(rng: ref HmacDrbgContext, conf: LightPushMixConf) {.async.}
       ephemeral: true, # tell store nodes to not store it
       timestamp: getNowInNanosecondTime(),
     ) # current timestamp
-    try:
-      let res = await node.wakuLightpushClient.publishWithConn(
+
+    (
+      await node.wakuLightpushClient.publishWithConn(
         LightpushPubsubTopic, message, conn, dPeerId
       )
+    ).isOkOr:
+      error "failed to publish message via mix", error = error.desc
+      lp_mix_failed.inc(labelValues = ["publish_error"])
+      return
+
+    lp_mix_success.inc()
+    notice "published message",
+      text = text,
+      timestamp = message.timestamp,
+      psTopic = LightpushPubsubTopic,
+      contentTopic = LightpushContentTopic
 
-      if res.isOk():
-        lp_mix_success.inc()
-        notice "published message",
-          text = text,
-          timestamp = message.timestamp,
-          psTopic = LightpushPubsubTopic,
-          contentTopic = LightpushContentTopic
-      else:
-        error "failed to publish message", error = $res.error
-        lp_mix_failed.inc(labelValues = ["publish_error"])
-    except CatchableError as e:
-      error "exception while publishing message", error = getCurrentExceptionMsg()
     if conf.mixDisabled:
       await conn.close()
     await sleepAsync(conf.msgIntervalMilliseconds)
-  info "###########Sent all messages via mix"
+  info "Sent all messages via mix"
   quit(0)
 
 when isMainModule:

waku/common/callbacks.nim

@@ -1,4 +1,4 @@
-import ../waku_enr/capabilities, ../waku_rendezvous/waku_peer_record
+import waku/waku_enr/capabilities, waku/waku_rendezvous/waku_peer_record
 
 type GetShards* = proc(): seq[uint16] {.closure, gcsafe, raises: [].}
 

waku/waku_mix/protocol.nim

@@ -36,11 +36,11 @@ type
     multiAddr*: string
     pubKey*: Curve25519Key
 
-proc mixPoolFilter*(cluster: Option[uint16], peer: RemotePeerInfo): bool =
+proc filterMixNodes*(cluster: Option[uint16], peer: RemotePeerInfo): bool =
   # Note that origin based(discv5) filtering is not done intentionally
   # so that more mix nodes can be discovered.
   if peer.mixPubKey.isNone():
-    trace "peer has no mix Pub Key", peer = $peer
+    trace "remote peer has no mix Pub Key", peer = $peer
     return false
 
   if cluster.isSome() and peer.enr.isSome() and
@@ -75,7 +75,7 @@ func getIPv4Multiaddr*(maddrs: seq[MultiAddress]): Option[MultiAddress] =
 proc populateMixNodePool*(mix: WakuMix) =
   # populate only peers that i) are reachable ii) share cluster iii) support mix
   let remotePeers = mix.peerManager.switch.peerStore.peers().filterIt(
-      mixPoolFilter(some(mix.clusterId), it)
+      filterMixNodes(some(mix.clusterId), it)
     )
   var mixNodes = initTable[PeerId, MixPubInfo]()
 
@@ -87,19 +87,19 @@ proc populateMixNodePool*(mix: WakuMix) =
     trace "remote peer info", info = remotePeers[i]
 
     if remotePeers[i].mixPubKey.isNone():
-      trace "peer has no mix Pub Key", peer = $remotePeers[i]
+      trace "peer has no mix Pub Key", remotePeerId = $remotePeers[i]
       continue
 
     let peerMixPubKey = remotePeers[i].mixPubKey.get()
     var peerPubKey: crypto.PublicKey
     if not remotePeers[i].peerId.extractPublicKey(peerPubKey):
       warn "Failed to extract public key from peerId, skipping node",
-        peerId = remotePeers[i].peerId
+        remotePeerId = remotePeers[i].peerId
       continue
 
     if peerPubKey.scheme != PKScheme.Secp256k1:
       warn "Peer public key is not Secp256k1, skipping node",
-        peerId = remotePeers[i].peerId, scheme = peerPubKey.scheme
+        remotePeerId = remotePeers[i].peerId, scheme = peerPubKey.scheme
       continue
 
     let mixNodePubInfo = MixPubInfo.init(
@@ -109,7 +109,7 @@ proc populateMixNodePool*(mix: WakuMix) =
       peerPubKey.skkey,
     )
     trace "adding mix node to pool",
-      peerId = remotePeers[i].peerId, multiAddr = $ipv4addr
+      remotePeerId = remotePeers[i].peerId, multiAddr = $ipv4addr
     mixNodes[remotePeers[i].peerId] = mixNodePubInfo
 
   # set the mix node pool

waku/waku_rendezvous/client.nim

@@ -13,9 +13,9 @@ import
 import metrics except collect
 
 import
-  ../node/peer_manager,
-  ../waku_core/peers,
-  ../waku_core/codecs,
+  waku/node/peer_manager,
+  waku/waku_core/peers,
+  waku/waku_core/codecs,
   ./common,
   ./waku_peer_record
 
@@ -34,6 +34,9 @@ type WakuRendezVousClient* = ref object
   # Internal rendezvous instance for making requests
   rdv: GenericRendezVous[WakuPeerRecord]
 
+const MaxSimultanesousAdvertisements = 5
+const RendezVousLookupInterval = 10.seconds
+
 proc requestAll*(
     self: WakuRendezVousClient
 ): Future[Result[void, string]] {.async: (raises: []).} =
@@ -56,7 +59,8 @@ proc requestAll*(
 
   trace "waku rendezvous client request got peers", count = records.len
   for record in records:
-    rendezvousPeerFoundTotal.inc()
+    if not self.switch.peerStore.peerExists(record.peerId):
+      rendezvousPeerFoundTotal.inc()
     if record.mixKey.len == 0 or record.peerId == self.switch.peerInfo.peerId:
       continue
     trace "adding peer from rendezvous",
@@ -77,11 +81,11 @@ proc periodicRequests(self: WakuRendezVousClient) {.async.} =
 
   # infinite loop
   while true:
+    await sleepAsync(self.requestInterval)
+
     (await self.requestAll()).isOkOr:
       error "waku rendezvous requests failed", error = error
 
-    await sleepAsync(self.requestInterval)
-
     # Exponential backoff
 
 #[    TODO: Reevaluate for mix, maybe be aggresive in the start until a sizeable pool is built and then backoff
@@ -102,7 +106,7 @@ proc new*(
   let rdv = GenericRendezVous[WakuPeerRecord](
     switch: switch,
     rng: rng,
-    sema: newAsyncSemaphore(5),
+    sema: newAsyncSemaphore(MaxSimultanesousAdvertisements),
     minDuration: rendezvous.MinimumAcceptedDuration,
     maxDuration: rendezvous.MaximumDuration,
     minTTL: rendezvous.MinimumAcceptedDuration.seconds.uint64,
@@ -119,7 +123,7 @@ proc new*(
     switch: switch,
     peerManager: peerManager,
     clusterId: clusterId,
-    requestInterval: 10.seconds,
+    requestInterval: RendezVousLookupInterval,
     rdv: rdv,
   )
 

waku/waku_rendezvous/protocol.nim

@@ -45,13 +45,15 @@ type WakuRendezVous* = ref object of GenericRendezVous[WakuPeerRecord]
   registrationInterval: timer.Duration
   periodicRegistrationFut: Future[void]
 
-# Override discover method to avoid collect macro generic instantiation issues
-# TODO figure out if we can use parent generic discover
+const MaximumNamespaceLen = 255
+
 method discover*(
     self: WakuRendezVous, conn: Connection, d: Discover
 ) {.async: (raises: [CancelledError, LPStreamError]).} =
+  # Override discover method to avoid collect macro generic instantiation issues
+  # TODO figure out if we can use parent generic discover
   trace "Received Discover", peerId = conn.peerId, ns = d.ns
-  if d.ns.isSome() and d.ns.get().len > 255: #MaximumNamespaceLen
+  if d.ns.isSome() and d.ns.get().len > MaximumNamespaceLen:
     await conn.sendDiscoverResponseError(InvalidNamespace)
     return
 
@@ -115,9 +117,10 @@ proc advertise*(
   let se = SignedPayload[WakuPeerRecord].init(
     self.switch.peerInfo.privateKey, self.getPeerRecord()
   ).valueOr:
-    return err("rendezvous advertisement failed: Failed to sign Waku Peer Record")
+    return
+      err("rendezvous advertisement failed: Failed to sign Waku Peer Record: " & $error)
   let sprBuff = se.encode().valueOr:
-    return err("rendezvous advertisement failed: Wrong Signed Peer Record")
+    return err("rendezvous advertisement failed: Wrong Signed Peer Record: " & $error)
 
   # rendezvous.advertise expects already opened connections
   # must dial first
@@ -205,7 +208,7 @@ proc new*(
     clusterId: uint16,
     getShards: GetShards,
     getCapabilities: GetCapabilities,
-    getPeerRecord: GetWakuPeerRecord = nil,
+    getPeerRecord: GetWakuPeerRecord,
 ): Result[T, string] {.raises: [].} =
   let rng = newRng()
   let wrv = T(
@@ -276,7 +279,7 @@ proc start*(self: WakuRendezVous) {.async: (raises: []).} =
     await procCall GenericRendezVous[WakuPeerRecord](self).start()
   except CancelledError as exc:
     error "failed to start GenericRendezVous", cause = exc.msg
-
+    return
   # start registering forever
   self.periodicRegistrationFut = self.periodicRegistration()