fix: allow autosharding nodes to get peers

Author: adklempner

packages/tests/src/node/node.ts

@@ -1,3 +1,4 @@
+/* eslint-disable no-console */
 import type { PeerId } from "@libp2p/interface/peer-id";
 import { peerIdFromString } from "@libp2p/peer-id";
 import { Multiaddr, multiaddr } from "@multiformats/multiaddr";
@@ -159,6 +160,9 @@ export class NimGoNode {
             this.logPath,
             WAKU_SERVICE_NODE_PARAMS
           );
+
+          console.log("node pubsubTopic:" + mergedArgs.pubsubTopic);
+          console.log("node clusterid: " + mergedArgs.clusterId);
         } catch (error) {
           log.error("Nwaku node failed to start:", error);
           await this.stop();

packages/tests/tests/getPeers.spec.ts

@@ -51,7 +51,8 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo),
-      lightpush: true
+      lightpush: true,
+      relay: true
     });
 
     const serviceNodeMa = await serviceNode1.getMultiaddrWithId();
@@ -77,12 +78,18 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       shards: [1]
     };
 
+    const shardInfoServiceNode: ShardInfo = {
+      clusterId: 1,
+      shards: [2]
+    };
+
     await serviceNode1.start({
       discv5Discovery: true,
       peerExchange: true,
-      clusterId: shardInfo.clusterId,
-      pubsubTopic: shardInfoToPubsubTopics(shardInfo),
-      lightpush: true
+      clusterId: shardInfoServiceNode.clusterId,
+      pubsubTopic: shardInfoToPubsubTopics(shardInfoServiceNode),
+      lightpush: true,
+      relay: true
     });
 
     const serviceNodeMa = await serviceNode1.getMultiaddrWithId();
@@ -120,7 +127,8 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo1.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo1),
-      lightpush: true
+      lightpush: true,
+      relay: true
     });
 
     // and another node in the same cluster cluster as our node
@@ -129,15 +137,16 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo2.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo2),
-      lightpush: true
+      lightpush: true,
+      relay: true
     });
 
     const serviceNode1Ma = await serviceNode1.getMultiaddrWithId();
     const serviceNode2Ma = await serviceNode2.getMultiaddrWithId();
 
     waku = await createLightNode({ shardInfo: shardInfo2 });
-    await waku.libp2p.dialProtocol(serviceNode1Ma, LightPushCodec);
-    await waku.libp2p.dialProtocol(serviceNode2Ma, LightPushCodec);
+    await waku.dial(serviceNode1Ma, [Protocols.LightPush]);
+    await waku.dial(serviceNode2Ma, [Protocols.LightPush]);
 
     await waku.start();
     await waitForRemotePeer(waku, [Protocols.LightPush]);
@@ -170,7 +179,8 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo1.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo1),
-      lightpush: true
+      lightpush: true,
+      relay: true
     });
 
     // and another node in the same cluster cluster as our node
@@ -180,15 +190,16 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo2.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo2),
-      lightpush: true
+      lightpush: true,
+      relay: true
     });
 
     const serviceNodeMa1 = await serviceNode1.getMultiaddrWithId();
     const serviceNodeMa2 = await serviceNode2.getMultiaddrWithId();
 
     waku = await createLightNode({ shardInfo: shardInfo2 });
-    await waku.libp2p.dialProtocol(serviceNodeMa1, LightPushCodec);
-    await waku.libp2p.dialProtocol(serviceNodeMa2, LightPushCodec);
+    await waku.dial(serviceNodeMa1, [Protocols.LightPush]);
+    await waku.dial(serviceNodeMa2, [Protocols.LightPush]);
     await waku.start();
     await waitForRemotePeer(waku, [Protocols.LightPush]);
 

packages/tests/tests/light-push/multiple_pubsub.node.spec.ts

@@ -290,7 +290,8 @@ describe("Waku Light Push (Autosharding): Multiple PubsubTopics", function () {
       filter: true,
       lightpush: true,
       relay: true,
-      pubsubTopic: [autoshardingPubsubTopic2]
+      pubsubTopic: [autoshardingPubsubTopic2],
+      clusterId: shardInfo.clusterId
     });
     await nwaku2.ensureSubscriptionsAutosharding([customContentTopic2]);
     await waku.dial(await nwaku2.getMultiaddrWithId());
@@ -353,10 +354,6 @@ describe("Waku Light Push (named sharding): Multiple PubsubTopics", function ()
     customContentTopic2,
     clusterId
   );
-  const contentTopicInfo: ContentTopicInfo = {
-    clusterId,
-    contentTopics: [customContentTopic1, customContentTopic2]
-  };
   const customEncoder1 = createEncoder({
     contentTopic: customContentTopic1,
     pubsubTopicShardInfo: {
@@ -372,11 +369,10 @@ describe("Waku Light Push (named sharding): Multiple PubsubTopics", function ()
 
   this.beforeEach(async function () {
     this.timeout(15000);
-    [nwaku, waku] = await runNodes(
-      this,
-      [autoshardingPubsubTopic1, autoshardingPubsubTopic2],
-      contentTopicInfo
-    );
+    [nwaku, waku] = await runNodes(this, [
+      autoshardingPubsubTopic1,
+      autoshardingPubsubTopic2
+    ]);
     messageCollector = new MessageCollector(nwaku);
     nimPeerId = await nwaku.getPeerId();
   });

packages/tests/tests/store/multiple_pubsub.spec.ts

@@ -214,7 +214,8 @@ describe("Waku Store (Autosharding), custom pubsub topic", function () {
     await nwaku.start({
       store: true,
       pubsubTopic: [autoshardingPubsubTopic1, autoshardingPubsubTopic2],
-      relay: true
+      relay: true,
+      clusterId
     });
     await nwaku.ensureSubscriptionsAutosharding([
       customContentTopic1,
@@ -287,7 +288,8 @@ describe("Waku Store (Autosharding), custom pubsub topic", function () {
     await nwaku2.start({
       store: true,
       pubsubTopic: [autoshardingPubsubTopic2],
-      relay: true
+      relay: true,
+      clusterId
     });
     await nwaku2.ensureSubscriptionsAutosharding([customContentTopic2]);
 

packages/utils/src/libp2p/index.ts

@@ -4,6 +4,7 @@ import type { ShardingParams } from "@waku/interfaces";
 
 import { bytesToUtf8 } from "../bytes/index.js";
 import { decodeRelayShard } from "../common/relay_shard_codec.js";
+import { contentTopicToShardIndex } from "../index.js";
 
 /**
  * Returns a pseudo-random peer that supports the given protocol.
@@ -89,17 +90,33 @@ export async function getConnectedPeersForProtocolAndShard(
 
     if (supportsProtocol) {
       if (shardInfo) {
-        //TODO: support auto-sharding
+        let shards;
         if (!("shards" in shardInfo)) {
-          throw new Error(
-            `Connections Manager only supports static sharding for now. Autosharding is not supported.`
+          if (!("contentTopics" in shardInfo)) {
+            throw new Error(
+              "Missing configuration for static or auto sharding"
+            );
+          }
+          // content topics to shards
+          shards = shardInfo.contentTopics.map((topic) =>
+            contentTopicToShardIndex(topic)
           );
+        } else {
+          shards = shardInfo.shards;
         }
         const encodedPeerShardInfo = peer.metadata.get("shardInfo");
         const peerShardInfo =
           encodedPeerShardInfo && decodeRelayShard(encodedPeerShardInfo);
 
-        if (peerShardInfo && shardInfo.clusterId === peerShardInfo.clusterId) {
+        // eslint-disable-next-line no-console
+        console.log("peer shards: " + peerShardInfo?.shards);
+
+        if (
+          peerShardInfo &&
+          shardInfo.clusterId === peerShardInfo.clusterId &&
+          (!("contentTopics" in shardInfo) ||
+            shards.some((shard) => peerShardInfo.shards.includes(shard)))
+        ) {
           return peer;
         }
       } else {