integrate mix protocol with extended kademlia discovery, libwaku API updates for logos-chat-poc integration

Author: chaitanyaprem

apps/chat2mix/chat2mix.nim

@@ -24,12 +24,14 @@ import
     stream/connection, # create and close stream read / write connections
     multiaddress,
       # encode different addressing schemes. For example, /ip4/7.7.7.7/tcp/6543 means it is using IPv4 protocol and TCP
+    multicodec,
     peerinfo,
       # manage the information of a peer, such as peer ID and public / private key
     peerid, # Implement how peers interact
     protobuf/minprotobuf, # message serialisation/deserialisation from and to protobufs
     nameresolving/dnsresolver,
     protocols/mix/curve25519,
+    protocols/mix/mix_protocol,
   ] # define DNS resolution
 import
   waku/[
@@ -38,6 +40,7 @@ import
     waku_lightpush/rpc,
     waku_enr,
     discovery/waku_dnsdisc,
+    discovery/waku_ext_kademlia,
     waku_node,
     node/waku_metrics,
     node/peer_manager,
@@ -84,6 +87,24 @@ type
 
 const MinMixNodePoolSize = 4
 
+proc parseKadBootstrapNode(
+    multiAddrStr: string
+): Result[(PeerId, seq[MultiAddress]), string] =
+  ## Parse a multiaddr string that includes /p2p/<peerID> into (PeerId, seq[MultiAddress])
+  let multiAddr = MultiAddress.init(multiAddrStr).valueOr:
+    return err("Invalid multiaddress: " & multiAddrStr)
+
+  let peerIdPart = multiAddr.getPart(multiCodec("p2p")).valueOr:
+    return err("Multiaddress must include /p2p/<peerID>: " & multiAddrStr)
+
+  let peerIdBytes = peerIdPart.protoArgument().valueOr:
+    return err("Failed to extract peer ID from multiaddress: " & multiAddrStr)
+
+  let peerId = PeerId.init(peerIdBytes).valueOr:
+    return err("Invalid peer ID in multiaddress: " & multiAddrStr)
+
+  ok((peerId, @[multiAddr]))
+
 #####################
 ## chat2 protobufs ##
 #####################
@@ -453,14 +474,39 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
   (await node.mountMix(conf.clusterId, mixPrivKey, conf.mixnodes)).isOkOr:
     error "failed to mount waku mix protocol: ", error = $error
     quit(QuitFailure)
-  await node.mountRendezvousClient(conf.clusterId)
+
+  # Setup extended kademlia discovery if bootstrap nodes are provided
+  if conf.kadBootstrapNodes.len > 0:
+    var kadBootstrapPeers: seq[(PeerId, seq[MultiAddress])]
+    for nodeStr in conf.kadBootstrapNodes:
+      let parsed = parseKadBootstrapNode(nodeStr).valueOr:
+        error "Failed to parse kademlia bootstrap node", node = nodeStr, error = error
+        continue
+      kadBootstrapPeers.add(parsed)
+
+    if kadBootstrapPeers.len > 0:
+      (
+        await setupExtendedKademliaDiscovery(
+          node,
+          ExtendedKademliaDiscoveryParams(
+            bootstrapNodes: kadBootstrapPeers,
+            mixPubKey: some(mixPubKey),
+            advertiseMix: false,
+          ),
+        )
+      ).isOkOr:
+        error "failed to setup kademlia discovery", error = error
+        quit(QuitFailure)
+
+  #await node.mountRendezvousClient(conf.clusterId)
 
   await node.start()
 
   node.peerManager.start()
+  node.startExtendedKademliaDiscoveryLoop(minMixPeers = MinMixNodePoolSize)
 
   await node.mountLibp2pPing()
-  await node.mountPeerExchangeClient()
+  #await node.mountPeerExchangeClient()
   let pubsubTopic = conf.getPubsubTopic(node, conf.contentTopic)
   echo "pubsub topic is: " & pubsubTopic
   let nick = await readNick(transp)
@@ -601,11 +647,6 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
     node, pubsubTopic, conf.contentTopic, servicePeerInfo, false
   )
   echo "waiting for mix nodes to be discovered..."
-  while true:
-    if node.getMixNodePoolSize() >= MinMixNodePoolSize:
-      break
-    discard await node.fetchPeerExchangePeers()
-    await sleepAsync(1000)
 
   while node.getMixNodePoolSize() < MinMixNodePoolSize:
     info "waiting for mix nodes to be discovered",

apps/chat2mix/config_chat2mix.nim

@@ -203,13 +203,13 @@ type
     fleet* {.
       desc:
         "Select the fleet to connect to. This sets the DNS discovery URL to the selected fleet.",
-      defaultValue: Fleet.test,
+      defaultValue: Fleet.none,
       name: "fleet"
     .}: Fleet
 
     contentTopic* {.
       desc: "Content topic for chat messages.",
-      defaultValue: "/toy-chat-mix/2/huilong/proto",
+      defaultValue: "/toy-chat/2/baixa-chiado/proto",
       name: "content-topic"
     .}: string
 
@@ -228,7 +228,14 @@ type
       desc: "WebSocket Secure Support.",
       defaultValue: false,
       name: "websocket-secure-support"
-    .}: bool ## rln-relay configuration
+    .}: bool
+
+    ## Kademlia Discovery config
+    kadBootstrapNodes* {.
+      desc:
+        "Peer multiaddr for kademlia discovery bootstrap node (must include /p2p/<peerID>). Argument may be repeated.",
+      name: "kad-bootstrap-node"
+    .}: seq[string]
 
 proc parseCmdArg*(T: type MixNodePubInfo, p: string): T =
   let elements = p.split(":")

library/kernel_api/debug_node_api.nim

@@ -9,7 +9,11 @@ import
   metrics,
   ffi
 import
-  waku/factory/waku, waku/node/waku_node, waku/node/health_monitor, library/declare_lib
+  waku/factory/waku,
+  waku/node/waku_node,
+  waku/node/health_monitor,
+  library/declare_lib,
+  waku/waku_core/codecs
 
 proc getMultiaddresses(node: WakuNode): seq[string] =
   return node.info().listenAddresses
@@ -48,3 +52,19 @@ proc waku_is_online(
     ctx: ptr FFIContext[Waku], callback: FFICallBack, userData: pointer
 ) {.ffi.} =
   return ok($ctx.myLib[].healthMonitor.onlineMonitor.amIOnline())
+
+proc waku_get_mixnode_pool_size(
+    ctx: ptr FFIContext[Waku], callback: FFICallBack, userData: pointer
+) {.ffi.} =
+  ## Returns the number of mix nodes in the pool
+  if ctx.myLib[].node.wakuMix.isNil():
+    return ok("0")
+  return ok($ctx.myLib[].node.getMixNodePoolSize())
+
+proc waku_get_lightpush_peers_count(
+    ctx: ptr FFIContext[Waku], callback: FFICallBack, userData: pointer
+) {.ffi.} =
+  ## Returns the count of all peers in peerstore supporting lightpush protocol
+  let peers =
+    ctx.myLib[].node.peerManager.switch.peerStore.getPeersByProtocol(WakuLightPushCodec)
+  return ok($peers.len)

library/kernel_api/protocols/filter_api.nim

@@ -47,8 +47,10 @@ proc waku_filter_subscribe(
     error "fail filter subscribe", error = errorMsg
     return err(errorMsg)
 
+  let pubsubTopicOpt =
+    if ($pubsubTopic).len > 0: some(PubsubTopic($pubsubTopic)) else: none(PubsubTopic)
   let subFut = ctx.myLib[].node.filterSubscribe(
-    some(PubsubTopic($pubsubTopic)),
+    pubsubTopicOpt,
     ($contentTopics).split(",").mapIt(ContentTopic(it)),
     peer,
   )

library/kernel_api/protocols/lightpush_api.nim

@@ -39,13 +39,15 @@ proc waku_lightpush_publish(
     let errorMsg = "failed to lightpublish message, no suitable remote peers"
     error "PUBLISH failed", error = errorMsg
     return err(errorMsg)
+  let topic =
+    if ($pubsubTopic).len == 0:
+      none(PubsubTopic)
+    else:
+      some(PubsubTopic($pubsubTopic))
 
-  let msgHashHex = (
-    await ctx.myLib[].node.wakuLegacyLightpushClient.publish(
-      $pubsubTopic, msg, peer = peerOpt.get()
-    )
-  ).valueOr:
-    error "PUBLISH failed", error = error
-    return err($error)
+  discard (await ctx.myLib[].node.lightpushPublish(topic, msg, peerOpt)).valueOr:
+    let errorMsg = error.desc.get($error.code.int)
+    error "PUBLISH failed", error = errorMsg
+    return err(errorMsg)
 
-  return ok(msgHashHex)
+  return ok("")

library/kernel_api/protocols/relay_api.nim

@@ -85,30 +85,47 @@ proc waku_relay_subscribe(
     callback: FFICallBack,
     userData: pointer,
     pubSubTopic: cstring,
+    contentTopic: cstring,
 ) {.ffi.} =
-  echo "Subscribing to topic: " & $pubSubTopic & " ..."
   proc onReceivedMessage(ctx: ptr FFIContext[Waku]): WakuRelayHandler =
     return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} =
       callEventCallback(ctx, "onReceivedMessage"):
         $JsonMessageEvent.new(pubsubTopic, msg)
 
   var cb = onReceivedMessage(ctx)
 
-  ctx.myLib[].node.subscribe(
-    (kind: SubscriptionKind.PubsubSub, topic: $pubsubTopic),
-    handler = WakuRelayHandler(cb),
-  ).isOkOr:
+  # If contentTopic is provided and non-empty, use ContentSub, otherwise use PubsubSub
+  let subscription =
+    if contentTopic != nil and len($contentTopic) > 0:
+      echo "Subscribing to content topic: " & $contentTopic & " ..."
+      (kind: SubscriptionKind.ContentSub, topic: $contentTopic)
+    else:
+      echo "Subscribing to pubsub topic: " & $pubSubTopic & " ..."
+      (kind: SubscriptionKind.PubsubSub, topic: $pubsubTopic)
+
+  ctx.myLib[].node.subscribe(subscription, handler = WakuRelayHandler(cb)).isOkOr:
     error "SUBSCRIBE failed", error = error
     return err($error)
   return ok("")
 
+# NOTE: When unsubscribing via contentTopic, this will unsubscribe from the entire
+# underlying pubsub topic/shard that the content topic maps to. This affects ALL
+# content topics on the same shard, not just the specified content topic.
 proc waku_relay_unsubscribe(
     ctx: ptr FFIContext[Waku],
     callback: FFICallBack,
     userData: pointer,
     pubSubTopic: cstring,
+    contentTopic: cstring,
 ) {.ffi.} =
-  ctx.myLib[].node.unsubscribe((kind: SubscriptionKind.PubsubSub, topic: $pubsubTopic)).isOkOr:
+  # If contentTopic is provided and non-empty, use ContentUnsub, otherwise use PubsubUnsub
+  let subscription =
+    if contentTopic != nil and len($contentTopic) > 0:
+      (kind: SubscriptionKind.ContentUnsub, topic: $contentTopic)
+    else:
+      (kind: SubscriptionKind.PubsubUnsub, topic: $pubsubTopic)
+
+  ctx.myLib[].node.unsubscribe(subscription).isOkOr:
     error "UNSUBSCRIBE failed", error = error
     return err($error)
 

library/kernel_api/protocols/store_api.nim

@@ -8,8 +8,13 @@ import
   waku/waku_store/common,
   waku/waku_store/client,
   waku/common/paging,
+  waku/common/base64,
   library/declare_lib
 
+# Custom JSON serialization for seq[byte] to avoid ambiguity
+proc `%`*(data: seq[byte]): JsonNode =
+  %base64.encode(data)
+
 func fromJsonNode(jsonContent: JsonNode): Result[StoreQueryRequest, string] =
   var contentTopics: seq[string]
   if jsonContent.contains("contentTopics"):
@@ -90,5 +95,6 @@ proc waku_store_query(
   ).valueOr:
     return err("StoreRequest failed store query: " & $error)
 
-  let res = $(%*(queryResponse.toHex()))
+  let hexResponse = queryResponse.toHex()
+  let res = $(%*hexResponse)
   return ok(res) ## returning the response in json format

library/libwaku.h

@@ -85,7 +85,8 @@ extern "C"
   int waku_relay_subscribe(void *ctx,
                            FFICallBack callback,
                            void *userData,
-                           const char *pubSubTopic);
+                           const char *pubSubTopic,
+                           const char *contentTopic);
 
   int waku_relay_add_protected_shard(void *ctx,
                                      FFICallBack callback,
@@ -97,7 +98,8 @@ extern "C"
   int waku_relay_unsubscribe(void *ctx,
                              FFICallBack callback,
                              void *userData,
-                             const char *pubSubTopic);
+                             const char *pubSubTopic,
+                             const char *contentTopic);
 
   int waku_filter_subscribe(void *ctx,
                             FFICallBack callback,
@@ -247,6 +249,14 @@ extern "C"
                      FFICallBack callback,
                      void *userData);
 
+  int waku_get_mixnode_pool_size(void *ctx,
+                                  FFICallBack callback,
+                                  void *userData);
+
+  int waku_get_lightpush_peers_count(void *ctx,
+                                      FFICallBack callback,
+                                      void *userData);
+
 #ifdef __cplusplus
 }
 #endif

nix/default.nix

@@ -45,9 +45,6 @@ in stdenv.mkDerivation {
     pkgs.darwin.cctools gcc # Necessary for libbacktrace
   ];
 
-  # Environment variables required for Android builds
-  ANDROID_SDK_ROOT="${pkgs.androidPkgs.sdk}";
-  ANDROID_NDK_HOME="${pkgs.androidPkgs.ndk}";
   NIMFLAGS = "-d:disableMarchNative -d:git_revision_override=${revision}";
   XDG_CACHE_HOME = "/tmp";
 
@@ -61,6 +58,15 @@ in stdenv.mkDerivation {
 
   configurePhase = ''
     patchShebangs . vendor/nimbus-build-system > /dev/null
+
+    # build_nim.sh guards "rm -rf dist/checksums" with NIX_BUILD_TOP != "/build",
+    # but on macOS the nix sandbox uses /private/tmp/... so the check fails and
+    # dist/checksums (provided via preBuild) gets deleted. Fix the check to skip
+    # the removal whenever NIX_BUILD_TOP is set (i.e. any nix build).
+    substituteInPlace vendor/nimbus-build-system/scripts/build_nim.sh \
+      --replace 'if [[ "''${NIX_BUILD_TOP}" != "/build" ]]; then' \
+                'if [[ -z "''${NIX_BUILD_TOP}" ]]; then'
+
     make nimbus-build-system-paths
     make nimbus-build-system-nimble-dir
   '';
@@ -98,6 +104,9 @@ in stdenv.mkDerivation {
     cp library/libwaku.h $out/include/
   '';
 
+  ANDROID_SDK_ROOT = "${pkgs.androidPkgs.sdk}";
+  ANDROID_NDK_HOME = "${pkgs.androidPkgs.ndk}";
+
   meta = with pkgs.lib; {
     description = "NWaku derivation to build libwaku for mobile targets using Android NDK and Rust.";
     homepage = "https://github.com/status-im/nwaku";

simulations/mixnet/config.toml

@@ -1,16 +1,17 @@
-log-level = "INFO"
+log-level = "TRACE"
 relay = true
 mix = true
 filter = true
 store = false
 lightpush = true
 max-connections = 150
-peer-exchange = true
+peer-exchange = false
 metrics-logging = false
 cluster-id = 2
-discv5-discovery = true
+discv5-discovery = false
 discv5-udp-port = 9000
 discv5-enr-auto-update = true
+enable-kad-discovery = true
 rest = true
 rest-admin = true
 ports-shift = 1
@@ -19,7 +20,9 @@ shard = [0]
 agent-string = "nwaku-mix"
 nodekey = "f98e3fba96c32e8d1967d460f1b79457380e1a895f7971cecc8528abe733781a"
 mixkey = "a87db88246ec0eedda347b9b643864bee3d6933eb15ba41e6d58cb678d813258"
-rendezvous = true
+rendezvous = false
 listen-address = "127.0.0.1"
 nat = "extip:127.0.0.1"
+ext-multiaddr = ["/ip4/127.0.0.1/tcp/60001"]
+ext-multiaddr-only = true
 ip-colocation-limit=0